diff --git a/docs/guides/browse_sh_crawler_playbook.md b/docs/guides/browse_sh_crawler_playbook.md index 9895afa..6ca468e 100644 --- a/docs/guides/browse_sh_crawler_playbook.md +++ b/docs/guides/browse_sh_crawler_playbook.md @@ -107,6 +107,12 @@ node scripts/ops/capture_pixelrag_visual_evidence.js \ node scripts/ops/capture_pixelrag_visual_evidence.js \ --manifest-file /tmp/pixelrag_manifest.json \ --output-dir data/ai_automation/pixelrag_visual_evidence + +# Production host can use the Python Playwright worker when Node modules are not installed. +python3 scripts/ops/capture_pixelrag_visual_evidence.py \ + --manifest-file /tmp/pixelrag_manifest.json \ + --output-dir data/ai_automation/pixelrag_visual_evidence \ + --max-tiles 12 ``` 安全邊界： diff --git a/scripts/ops/capture_pixelrag_visual_evidence.py b/scripts/ops/capture_pixelrag_visual_evidence.py new file mode 100644 index 0000000..100e55a --- /dev/null +++ b/scripts/ops/capture_pixelrag_visual_evidence.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +"""Read-only PixelRAG-style visual evidence capture worker. + +This Python worker is the production-friendly counterpart of the Node worker: +it can dry-run without Playwright and only imports Playwright when actual capture +is requested. +""" + +from __future__ import annotations + +import argparse +import json +import math +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.parse import urlparse + + +CAPTURE_POLICY = "read_only_pixelrag_visual_capture_artifact_v1" +MANIFEST_POLICY = "read_only_pixelrag_visual_evidence_manifest_v1" +DEFAULT_OUTPUT_DIR = "data/ai_automation/pixelrag_visual_evidence" +DEFAULT_TIMEOUT_MS = 30000 +DEFAULT_SETTLE_MS = 500 +DEFAULT_MAX_TILES = 80 +ALLOWED_PLATFORM_HOSTS = { + "momo": {"m.momoshop.com.tw", "www.momoshop.com.tw"}, + "pchome": {"24h.pchome.com.tw", "ecshweb.pchome.com.tw", "ecapi-cdn.pchome.com.tw"}, + "market_intel": set(), + "external_market": set(), +} + + +def _positive_int(value: Any, fallback: int) -> int: + try: + parsed = int(value) + except (TypeError, ValueError): + return fallback + return parsed if parsed > 0 else fallback + + +def _safe_name(value: Any) -> str: + text = "".join(ch if ch.isalnum() else "_" for ch in str(value or "").lower()) + text = "_".join(part for part in text.split("_") if part) + return text or "unknown" + + +def _load_manifest(args: argparse.Namespace) -> dict[str, Any]: + if args.manifest_file: + return json.loads(Path(args.manifest_file).read_text(encoding="utf-8")) + return json.loads(args.manifest_json) + + +def _validate_manifest(manifest: dict[str, Any], *, require_allowed_host: bool) -> list[str]: + errors: list[str] = [] + if manifest.get("policy") != MANIFEST_POLICY: + errors.append(f"manifest policy must be {MANIFEST_POLICY}") + target = manifest.get("capture_target") or {} + raw_url = str(target.get("url") or "") + parsed = urlparse(raw_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + errors.append("capture_target.url must be an absolute http(s) URL") + if parsed.username or parsed.password: + errors.append("URL credentials are not allowed") + platform = str(target.get("platform") or "") + allowed_hosts = ALLOWED_PLATFORM_HOSTS.get(platform) + if ( + require_allowed_host + and parsed.netloc + and allowed_hosts + and parsed.hostname not in allowed_hosts + ): + errors.append(f"host {parsed.hostname} is not allowed for platform {platform}") + if not platform: + errors.append("capture_target.platform is required") + if not target.get("crawler"): + errors.append("capture_target.crawler is required") + if not target.get("trigger_reason"): + errors.append("capture_target.trigger_reason is required") + return errors + + +def _build_tile_plan( + *, + width: int, + height: int, + tile_width: int, + tile_height: int, + max_tiles: int, +) -> dict[str, Any]: + tiles_x = max(1, math.ceil(width / tile_width)) + tiles_y = max(1, math.ceil(height / tile_height)) + tiles: list[dict[str, int]] = [] + for row in range(tiles_y): + for col in range(tiles_x): + if len(tiles) >= max_tiles: + break + x = col * tile_width + y = row * tile_height + tiles.append({ + "index": len(tiles), + "row": row, + "col": col, + "x": x, + "y": y, + "width": min(tile_width, max(1, width - x)), + "height": min(tile_height, max(1, height - y)), + }) + planned = tiles_x * tiles_y + return { + "tiles_x": tiles_x, + "tiles_y": tiles_y, + "planned_tile_count": planned, + "emitted_tile_count": len(tiles), + "capped": len(tiles) < planned, + "tiles": tiles, + } + + +def _output_paths(manifest: dict[str, Any], output_dir: str) -> dict[str, Path]: + target = manifest.get("capture_target") or {} + manifest_id = manifest.get("manifest_id") or _safe_name( + f"{target.get('platform')}-{target.get('crawler')}-{target.get('url')}" + )[:20] + root = Path(output_dir).resolve() / _safe_name(target.get("platform")) / manifest_id + return { + "root": root, + "receipt": root / "capture_receipt.json", + "screenshot": root / "fullpage.png", + "tiles_dir": root / "tiles", + } + + +def _base_artifact( + manifest: dict[str, Any], + args: argparse.Namespace, + errors: list[str], +) -> dict[str, Any]: + viewport = manifest.get("viewport") or {} + page_size = manifest.get("page_size") or {} + tile_size = manifest.get("tile_size") or {} + width = _positive_int(page_size.get("width"), _positive_int(viewport.get("width"), 1440)) + height = _positive_int(page_size.get("height"), _positive_int(viewport.get("height"), 950)) + tile_width = _positive_int(tile_size.get("width"), 512) + tile_height = _positive_int(tile_size.get("height"), 512) + paths = _output_paths(manifest, args.output_dir) + return { + "success": not errors, + "policy": CAPTURE_POLICY, + "status": "rejected" if errors else ("dry_run_ready" if args.dry_run else "capture_pending"), + "generated_at": datetime.now(timezone.utc).isoformat(), + "errors": errors, + "manifest_id": manifest.get("manifest_id") or "", + "manifest_policy": manifest.get("policy") or "", + "capture_target": manifest.get("capture_target") or {}, + "controlled_apply": { + "db_write": False, + "secret_read": False, + "github_dependency": False, + "production_price_write": False, + "artifact_write": bool(not args.dry_run and not errors), + }, + "planned_output": { + "root": str(paths["root"]), + "receipt": str(paths["receipt"]), + "screenshot": str(paths["screenshot"]), + "tiles_dir": str(paths["tiles_dir"]), + }, + "viewport": { + "name": viewport.get("name") or "desktop-1440", + "width": _positive_int(viewport.get("width"), 1440), + "height": _positive_int(viewport.get("height"), 950), + }, + "page_size": {"width": width, "height": height}, + "tile_size": {"width": tile_width, "height": tile_height}, + "tile_plan": _build_tile_plan( + width=width, + height=height, + tile_width=tile_width, + tile_height=tile_height, + max_tiles=_positive_int(args.max_tiles, DEFAULT_MAX_TILES), + ), + "files": [], + } + + +def _capture( + manifest: dict[str, Any], + args: argparse.Namespace, + artifact: dict[str, Any], +) -> dict[str, Any]: + from playwright.sync_api import sync_playwright + + paths = _output_paths(manifest, args.output_dir) + paths["tiles_dir"].mkdir(parents=True, exist_ok=True) + target = manifest.get("capture_target") or {} + viewport = artifact["viewport"] + timeout_ms = _positive_int(args.timeout * 1000, DEFAULT_TIMEOUT_MS) + + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=True) + context = browser.new_context(ignore_https_errors=True, viewport={ + "width": viewport["width"], + "height": viewport["height"], + }) + page = context.new_page() + try: + response = page.goto(target["url"], wait_until=args.wait_until, timeout=timeout_ms) + try: + page.wait_for_selector("body", timeout=min(timeout_ms, 5000)) + except Exception: + pass + if args.settle_ms > 0: + page.wait_for_timeout(args.settle_ms) + metrics = page.evaluate( + """() => { + const root = document.documentElement; + const body = document.body; + return { + final_url: location.href, + title: document.title, + scroll_width: Math.max(root.scrollWidth, body.scrollWidth, window.innerWidth), + scroll_height: Math.max(root.scrollHeight, body.scrollHeight, window.innerHeight) + }; + }""" + ) + page.screenshot(path=str(paths["screenshot"]), full_page=True) + tile_plan = _build_tile_plan( + width=_positive_int(metrics.get("scroll_width"), artifact["page_size"]["width"]), + height=_positive_int(metrics.get("scroll_height"), artifact["page_size"]["height"]), + tile_width=artifact["tile_size"]["width"], + tile_height=artifact["tile_size"]["height"], + max_tiles=_positive_int(args.max_tiles, DEFAULT_MAX_TILES), + ) + files: list[dict[str, Any]] = [ + {"kind": "fullpage_screenshot", "path": str(paths["screenshot"])}, + ] + for tile in tile_plan["tiles"]: + file_path = paths["tiles_dir"] / f"tile_{tile['index']:03d}.png" + page.screenshot(path=str(file_path), clip={ + "x": tile["x"], + "y": tile["y"], + "width": tile["width"], + "height": tile["height"], + }) + files.append({"kind": "tile", "path": str(file_path), "tile": tile}) + captured = { + **artifact, + "success": True, + "status": "captured", + "http_status": response.status if response else 0, + "page_metrics": metrics, + "page_size": { + "width": _positive_int(metrics.get("scroll_width"), artifact["page_size"]["width"]), + "height": _positive_int(metrics.get("scroll_height"), artifact["page_size"]["height"]), + }, + "tile_plan": tile_plan, + "files": files, + } + paths["receipt"].write_text(json.dumps(captured, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + captured["files"].append({"kind": "receipt", "path": str(paths["receipt"])}) + return captured + finally: + context.close() + browser.close() + + +def main() -> int: + parser = argparse.ArgumentParser(description="Capture PixelRAG visual evidence from a manifest.") + parser.add_argument("--manifest-file") + parser.add_argument("--manifest-json") + parser.add_argument("--output-dir", default=DEFAULT_OUTPUT_DIR) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--timeout", type=int, default=30) + parser.add_argument("--settle-ms", type=int, default=DEFAULT_SETTLE_MS) + parser.add_argument("--max-tiles", type=int, default=DEFAULT_MAX_TILES) + parser.add_argument("--wait-until", default="domcontentloaded") + parser.add_argument("--allow-any-host", action="store_true") + args = parser.parse_args() + if not args.manifest_file and not args.manifest_json: + parser.error("Pass --manifest-file or --manifest-json.") + + manifest = _load_manifest(args) + errors = _validate_manifest(manifest, require_allowed_host=not args.allow_any_host) + artifact = _base_artifact(manifest, args, errors) + if errors or args.dry_run: + print(json.dumps(artifact, ensure_ascii=False, indent=2)) + return 1 if errors else 0 + captured = _capture(manifest, args, artifact) + print(json.dumps(captured, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_pixelrag_crawler_integration_service.py b/tests/test_pixelrag_crawler_integration_service.py index 38bd30b..72a75e7 100644 --- a/tests/test_pixelrag_crawler_integration_service.py +++ b/tests/test_pixelrag_crawler_integration_service.py @@ -302,3 +302,40 @@ def test_pixelrag_capture_worker_rejects_wrong_platform_host(tmp_path): payload = json.loads(completed.stdout) assert payload["status"] == "rejected" assert any("not allowed for platform momo" in error for error in payload["errors"]) + + +def test_pixelrag_python_capture_worker_dry_run_plans_artifact_outputs(tmp_path): + from services.pixelrag_crawler_integration_service import ( + build_pixelrag_visual_evidence_manifest, + ) + + manifest = build_pixelrag_visual_evidence_manifest( + url="https://24h.pchome.com.tw/", + platform="pchome", + crawler="PChomeCrawler.visual_fallback", + trigger_reason="parser_empty", + page_size={"width": 1024, "height": 1024}, + tile_size={"width": 512, "height": 512}, + ) + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/capture_pixelrag_visual_evidence.py", + "--manifest-json", + json.dumps(manifest), + "--output-dir", + str(tmp_path), + "--dry-run", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["policy"] == "read_only_pixelrag_visual_capture_artifact_v1" + assert payload["status"] == "dry_run_ready" + assert payload["controlled_apply"]["artifact_write"] is False + assert payload["tile_plan"]["planned_tile_count"] == 4 + assert payload["capture_target"]["platform"] == "pchome"