You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
241 lines
8.6 KiB
Python
241 lines
8.6 KiB
Python
"""Learn mode: capture page artifacts and invoke Claude Code to generate adapters."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
|
|
from gimme_job.config import GlobalConfig
|
|
from gimme_job.runtime.claude_cli import ClaudeCodeClient, DEFAULT_LEARN_TOOLS
|
|
|
|
|
|
class LearnMode:
|
|
def __init__(self, global_config: GlobalConfig, claude_client: ClaudeCodeClient):
|
|
self.cfg = global_config
|
|
self.claude = claude_client
|
|
|
|
def learn_site(
|
|
self,
|
|
site_id: str,
|
|
url: str,
|
|
profile_name: str = "JobAgent",
|
|
headed: bool = True,
|
|
timeout: int = 180,
|
|
) -> bool:
|
|
from gimme_job.utils.paths import (
|
|
a11y_snapshot_path, html_dump_path, learn_screenshot_path,
|
|
manifests_dir, project_root, sites_dir,
|
|
)
|
|
from rich.console import Console
|
|
console = Console()
|
|
|
|
logger.info(f"[learn] Starting learn mode for {site_id}")
|
|
|
|
# Step 1: Open browser and capture artifacts
|
|
console.print(f"[cyan]Opening browser for {site_id}...[/cyan]")
|
|
artifacts = self._capture_artifacts(
|
|
site_id=site_id,
|
|
url=url,
|
|
profile_name=profile_name,
|
|
headed=headed,
|
|
timeout=timeout,
|
|
)
|
|
|
|
if not artifacts:
|
|
logger.error("[learn] Failed to capture artifacts")
|
|
return False
|
|
|
|
html_path, a11y_path, ss_path = artifacts
|
|
|
|
# Step 2: Build the learn prompt
|
|
prompt = self._build_prompt(
|
|
site_id=site_id,
|
|
url=url,
|
|
html_path=html_path,
|
|
a11y_path=a11y_path,
|
|
ss_path=ss_path,
|
|
)
|
|
|
|
# Step 3: Call Claude Code
|
|
console.print("[cyan]Calling Claude Code...[/cyan]")
|
|
result = self.claude.run_prompt(
|
|
prompt=prompt,
|
|
cwd=project_root(),
|
|
allowed_tools=DEFAULT_LEARN_TOOLS,
|
|
)
|
|
|
|
if not result.success:
|
|
logger.error(f"[learn] Claude Code failed: {result.error}")
|
|
return False
|
|
|
|
logger.info(f"[learn] Claude Code succeeded (session={result.session_id})")
|
|
|
|
# Step 4: Verify generated files
|
|
required_files = [
|
|
sites_dir() / f"{site_id}.yaml",
|
|
project_root() / "gimme_job" / "adapters" / f"{site_id}.py",
|
|
project_root() / "tests" / "adapters" / f"test_{site_id}.py",
|
|
]
|
|
missing = [f for f in required_files if not f.exists()]
|
|
if missing:
|
|
logger.error(f"[learn] Missing generated files: {missing}")
|
|
# One retry
|
|
console.print("[yellow]Some files missing — retrying...[/yellow]")
|
|
retry_prompt = f"The following files were not created: {missing}. Please create them now.\n\n" + prompt
|
|
result2 = self.claude.run_prompt(
|
|
prompt=retry_prompt,
|
|
cwd=project_root(),
|
|
allowed_tools=DEFAULT_LEARN_TOOLS,
|
|
)
|
|
if not result2.success:
|
|
return False
|
|
missing = [f for f in required_files if not f.exists()]
|
|
if missing:
|
|
logger.error(f"[learn] Still missing after retry: {missing}")
|
|
return False
|
|
|
|
# Step 5: Run smoke test
|
|
console.print(f"[cyan]Running smoke test for {site_id}...[/cyan]")
|
|
import subprocess
|
|
test_result = subprocess.run(
|
|
["uv", "run", "pytest", f"tests/adapters/test_{site_id}.py", "-v", "--tb=short"],
|
|
cwd=str(project_root()),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
)
|
|
|
|
if test_result.returncode != 0:
|
|
logger.warning(f"[learn] Smoke test failed:\n{test_result.stdout[-1000:]}")
|
|
# One repair attempt
|
|
console.print("[yellow]Smoke test failed — asking Claude to fix...[/yellow]")
|
|
fix_prompt = (
|
|
f"The smoke test for {site_id} failed. Here is the output:\n\n"
|
|
f"```\n{test_result.stdout[-2000:]}\n```\n\n"
|
|
"Please fix the adapter and/or manifest so the test passes."
|
|
)
|
|
self.claude.run_prompt(
|
|
prompt=fix_prompt,
|
|
cwd=project_root(),
|
|
allowed_tools=DEFAULT_LEARN_TOOLS,
|
|
)
|
|
# Re-run test
|
|
test_result2 = subprocess.run(
|
|
["uv", "run", "pytest", f"tests/adapters/test_{site_id}.py", "-v"],
|
|
cwd=str(project_root()),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
)
|
|
if test_result2.returncode != 0:
|
|
logger.error("[learn] Smoke test still failing after retry")
|
|
return False
|
|
|
|
logger.info(f"[learn] {site_id} learned successfully")
|
|
return True
|
|
|
|
def _capture_artifacts(
|
|
self,
|
|
site_id: str,
|
|
url: str,
|
|
profile_name: str,
|
|
headed: bool,
|
|
timeout: int,
|
|
) -> Optional[tuple[Path, Path, Path]]:
|
|
"""Open browser, navigate to URL, wait for user login, capture artifacts."""
|
|
from gimme_job.runtime.browser import BrowserManager
|
|
from gimme_job.utils.paths import (
|
|
a11y_snapshot_path, html_dump_path, learn_screenshot_path,
|
|
)
|
|
from gimme_job.utils.json_io import write_json
|
|
|
|
bm = BrowserManager(profile_name=profile_name, headless=not headed)
|
|
try:
|
|
context = bm.open_context(headless=not headed)
|
|
page = context.new_page()
|
|
page.set_default_timeout(timeout * 1000)
|
|
page.goto(url, timeout=timeout * 1000)
|
|
|
|
# Prompt user to login / handle cookie banners
|
|
import typer
|
|
print(f"\n Browser opened at: {url}")
|
|
print(" Please:")
|
|
print(" 1. Log in if required")
|
|
print(" 2. Dismiss cookie banners")
|
|
print(" 3. Make sure job results are visible")
|
|
typer.confirm(" Press Enter when the page is ready", default=True)
|
|
|
|
# Capture artifacts
|
|
html_path = html_dump_path(site_id)
|
|
html_path.parent.mkdir(parents=True, exist_ok=True)
|
|
html_path.write_text(page.content(), encoding="utf-8")
|
|
logger.info(f"HTML dump saved: {html_path}")
|
|
|
|
a11y_path = a11y_snapshot_path(site_id)
|
|
try:
|
|
a11y = page.accessibility.snapshot()
|
|
write_json(a11y_path, a11y or {})
|
|
logger.info(f"A11y snapshot saved: {a11y_path}")
|
|
except Exception as e:
|
|
logger.warning(f"A11y snapshot failed: {e}")
|
|
write_json(a11y_path, {})
|
|
|
|
ss_path = learn_screenshot_path(site_id)
|
|
ss_path.parent.mkdir(parents=True, exist_ok=True)
|
|
page.screenshot(path=str(ss_path), full_page=False)
|
|
logger.info(f"Screenshot saved: {ss_path}")
|
|
|
|
return html_path, a11y_path, ss_path
|
|
|
|
except Exception as e:
|
|
logger.error(f"[learn] Artifact capture failed: {e}")
|
|
return None
|
|
finally:
|
|
bm.close()
|
|
|
|
def _build_prompt(
|
|
self,
|
|
site_id: str,
|
|
url: str,
|
|
html_path: Path,
|
|
a11y_path: Path,
|
|
ss_path: Path,
|
|
) -> str:
|
|
from jinja2 import Environment, FileSystemLoader
|
|
from gimme_job.utils.paths import project_root
|
|
from gimme_job.utils.json_io import write_yaml
|
|
|
|
prompts_dir = project_root() / "gimme_job" / "prompts"
|
|
|
|
# Read system prompt
|
|
system_md = (prompts_dir / "learn_system.md").read_text()
|
|
|
|
# Build global config yaml snippet
|
|
import yaml
|
|
from gimme_job.config import load_global_config
|
|
cfg = load_global_config()
|
|
global_cfg_yaml = yaml.dump(
|
|
{"keywords": cfg.search_defaults.keywords, "date_mode": cfg.search_defaults.date_mode},
|
|
allow_unicode=True,
|
|
)
|
|
|
|
# Read base adapter source for context
|
|
base_src_path = project_root() / "gimme_job" / "adapters" / "base.py"
|
|
base_src = base_src_path.read_text() if base_src_path.exists() else ""
|
|
|
|
env = Environment(loader=FileSystemLoader(str(prompts_dir)))
|
|
template = env.get_template("learn_user.md.j2")
|
|
user_prompt = template.render(
|
|
site_id=site_id,
|
|
site_url=url,
|
|
html_dump_path=str(html_path),
|
|
dom_snapshot_path=str(html_path),
|
|
screenshot_path=str(ss_path),
|
|
global_config_yaml=global_cfg_yaml,
|
|
base_adapter_source=base_src[:3000], # truncate for prompt size
|
|
)
|
|
|
|
return f"{system_md}\n\n---\n\n{user_prompt}"
|