diff --git a/docs/guides/browse_sh_crawler_playbook.md b/docs/guides/browse_sh_crawler_playbook.md index 4490592..b0a1381 100644 --- a/docs/guides/browse_sh_crawler_playbook.md +++ b/docs/guides/browse_sh_crawler_playbook.md @@ -51,3 +51,39 @@ python scripts/tools/browse_sh_probe.py -- screenshot - PChome：目前已有搜尋 API 與商品 API，`browse` 只用於確認 API 參數、分頁行為、前端是否切新 endpoint。 - MOMO：若既有 BeautifulSoup selector 失效，先用 `browse` 找出前端實際 XHR；找到 API 時優先改成 structured API parser。 - Matcher：`browse` 只提供候選證據；是否為同款仍由 `marketplace_product_matcher.score_marketplace_match()` 決定。 + +## PixelRAG-style 視覺證據 fallback + +2026-07-09 評估 PixelRAG 後，結論是「可導入，但不可直接取代正式爬蟲」。PixelRAG 的核心價值是把渲染後頁面截圖切成 tiles，讓 AI 讀到 HTML parser 可能丟失的視覺結構；本專案第一階段只採用這個視覺證據思路，不直接拉外部 runtime、不用外部 embedding API、不把像素結果寫入正式價格表。 + +導入順序： + +1. 既有 PChome API / MOMO structured parser 先跑；若 parser 結果為空、價格欄位缺失、規格/組合資訊疑似只存在渲染畫面，才啟動視覺證據 fallback。 +2. Phase 1 只輸出 screenshot/tile manifest：URL、platform、crawler、parse failure、viewport、tile coordinates、artifact path、evidence intent。 +3. Phase 2 用 manifest 收集 replay samples，對照 `marketplace_product_matcher` 與 `competitor_match_attempts` 的診斷結果，評估是否提高身份證據覆蓋率。 +4. Phase 3 才評估 Ollama-first multimodal embedding；未完成 GCP-A → GCP-B → 111 視覺 embedding 驗證前，不做自動像素檢索。 +5. Phase 4 若需要索引，優先設計 pgvector-compatible evidence metadata；FAISS 只能先當本地研究/ADR 候選，不直接進 production。 +6. Phase 5 才談 crawler fusion；正式 `competitor_prices` / `competitor_price_history` 寫入仍需 matcher replay/canary 證據。 + +機器可讀評估： + +```bash +python scripts/ops/report_pixelrag_crawler_integration.py +python scripts/ops/report_pixelrag_crawler_integration.py --platform momo +python scripts/ops/report_pixelrag_crawler_integration.py --capability ollama_multimodal_ready --capability pgvector_visual_ready +python scripts/ops/report_pixelrag_crawler_integration.py \ + --platform momo \ + --manifest-url "https://m.momoshop.com.tw/search.momo?searchKeyword=test" \ + --crawler MomoCrawler.search_products \ + --trigger-reason parser_empty \ + --page-size page=1440x1900 \ + --tile-size tile=512x512 +``` + +安全邊界： + +- read-only；不登入、不下單、不加入購物車、不寫第三方狀態。 +- 不從像素結果直接寫正式價格或同款判斷，只寫 artifact / review diagnostics。 +- 不使用 GitHub runtime 依賴；不讀 secrets / sessions / cookies。 +- 不呼叫 hosted embedding / VLM API；如需 embedding，必須走 Ollama-first 並保留成本/品質 benchmark。 +- 若穩定 XHR/API 可取得同樣資訊，仍優先回補 structured parser，不把視覺 fallback 變成主路徑。 diff --git a/scripts/ops/report_pixelrag_crawler_integration.py b/scripts/ops/report_pixelrag_crawler_integration.py new file mode 100644 index 0000000..0a9f226 --- /dev/null +++ b/scripts/ops/report_pixelrag_crawler_integration.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +"""Report PixelRAG-style crawler integration readiness.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from services.pixelrag_crawler_integration_service import ( # noqa: E402 + build_pixelrag_crawler_integration_assessment, + build_pixelrag_visual_evidence_manifest, +) + + +CAPABILITY_FLAGS = { + "no_playwright": ("playwright_artifact_pipeline", False), + "ollama_multimodal_ready": ("ollama_multimodal_embedding_ready", True), + "pgvector_visual_ready": ("pgvector_visual_index_ready", True), + "faiss_allowed": ("faiss_allowed_in_production", True), +} + + +def _size_arg(value: str) -> dict[str, int | str]: + try: + name, size = value.split("=", 1) + except ValueError: + name, size = "", value + try: + width_text, height_text = size.lower().split("x", 1) + width = int(width_text) + height = int(height_text) + except (ValueError, TypeError): + raise argparse.ArgumentTypeError("size must be name=WIDTHxHEIGHT or WIDTHxHEIGHT") + payload: dict[str, int | str] = {"width": width, "height": height} + if name: + payload["name"] = name + return payload + + +def main() -> int: + parser = argparse.ArgumentParser( + description="輸出 PixelRAG 導入現有爬蟲解決方案的機器可讀評估。" + ) + parser.add_argument( + "--platform", + action="append", + dest="platforms", + help="限制目標平台，可重複指定；預設 momo + pchome。", + ) + parser.add_argument( + "--capability", + action="append", + choices=sorted(CAPABILITY_FLAGS), + default=[], + help="覆寫 readiness 假設，用於模擬下一階段條件。", + ) + parser.add_argument("--manifest-url", help="輸出單一 URL 的視覺證據 manifest。") + parser.add_argument("--crawler", default="crawler_diagnostics", help="manifest 來源 crawler 名稱。") + parser.add_argument( + "--trigger-reason", + default="parser_empty_or_low_confidence", + help="啟動視覺 fallback 的原因。", + ) + parser.add_argument( + "--evidence-intent", + default="recover_visual_offer_evidence", + help="視覺證據用途。", + ) + parser.add_argument( + "--viewport", + type=_size_arg, + help="viewport，例如 desktop-1440=1440x950。", + ) + parser.add_argument( + "--page-size", + type=_size_arg, + help="預估頁面尺寸，例如 page=1440x1900。", + ) + parser.add_argument( + "--tile-size", + type=_size_arg, + help="tile 尺寸，例如 tile=512x512。", + ) + args = parser.parse_args() + + if args.manifest_url: + platform = (args.platforms or ["momo"])[0] + payload = build_pixelrag_visual_evidence_manifest( + url=args.manifest_url, + platform=platform, + crawler=args.crawler, + trigger_reason=args.trigger_reason, + evidence_intent=args.evidence_intent, + viewport=args.viewport, + page_size=args.page_size, + tile_size=args.tile_size, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + capabilities = { + key: value + for flag in args.capability + for key, value in [CAPABILITY_FLAGS[flag]] + } + payload = build_pixelrag_crawler_integration_assessment( + capabilities=capabilities or None, + target_platforms=tuple(args.platforms) if args.platforms else None, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/pixelrag_crawler_integration_service.py b/services/pixelrag_crawler_integration_service.py new file mode 100644 index 0000000..ff0b9ec --- /dev/null +++ b/services/pixelrag_crawler_integration_service.py @@ -0,0 +1,503 @@ +"""PixelRAG-style visual evidence lane for crawler diagnostics. + +This module does not call external services, read credentials, or write DB data. +It turns the PixelRAG research pattern into a controlled, machine-readable +integration assessment for the existing MOMO/PChome crawler stack. +""" + +from __future__ import annotations + +import hashlib +import math +from copy import deepcopy +from datetime import datetime, timezone +from typing import Any, Mapping +from urllib.parse import urlparse + + +POLICY = "read_only_pixelrag_crawler_integration_assessment_v1" +MANIFEST_POLICY = "read_only_pixelrag_visual_evidence_manifest_v1" +RESULT_PHASE1_READY = "PIXELRAG_VISUAL_EVIDENCE_PHASE1_READY" +RESULT_BLOCKED_NO_CAPTURE = "PIXELRAG_BLOCKED_NO_VISUAL_CAPTURE" +RESULT_MANIFEST_READY = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY" +RESULT_MANIFEST_REJECTED = "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_REJECTED" + +ALLOWED_PLATFORMS = ("momo", "pchome", "market_intel", "external_market") +DEFAULT_VIEWPORT = {"name": "desktop-1440", "width": 1440, "height": 950} +DEFAULT_TILE_SIZE = {"width": 512, "height": 512} +VISUAL_FALLBACK_CONFIDENCE_TRIGGERS = { + "low", + "manual_review", + "identity_review", + "true_low_confidence", + "variant_selection_review", +} + +DEFAULT_CAPABILITIES: dict[str, bool] = { + "structured_crawler_api": True, + "momo_html_parser_fallback": True, + "matcher_guardrails": True, + "playwright_artifact_pipeline": True, + "browse_sh_optional_probe": True, + "ollama_multimodal_embedding_ready": False, + "pgvector_visual_index_ready": False, + "faiss_allowed_in_production": False, + "production_price_auto_write": False, +} + +TARGET_PLATFORMS = ("momo", "pchome") + +SOURCE_FINDINGS: tuple[dict[str, str], ...] = ( + { + "code": "pixel_native_retrieval", + "finding": ( + "PixelRAG represents web pages as screenshots, retrieves screenshot " + "tiles, and lets a vision-language model read the visual evidence." + ), + "adoption": "Use as visual evidence fallback after structured parsers fail.", + }, + { + "code": "playwright_screenshot_tiles", + "finding": ( + "The research pipeline starts with Playwright rendering, screenshot " + "capture, and image tiling." + ), + "adoption": "Ready for Phase 1 because this repo already has Playwright artifact patterns.", + }, + { + "code": "qwen3_vl_embedding_optional", + "finding": ( + "Full pixel-space retrieval needs a multimodal embedding model such as " + "Qwen3-VL-Embedding." + ), + "adoption": "Deferred until Ollama-first multimodal hosting is verified.", + }, + { + "code": "faiss_index_policy_gap", + "finding": "The paper uses FAISS for visual retrieval indexes.", + "adoption": "Do not add FAISS to production before pgvector/ADR review.", + }, +) + +SAFETY_GUARDS: tuple[dict[str, Any], ...] = ( + { + "code": "read_only_visual_capture", + "status": "enforced", + "reason": "Screenshots and tile manifests are diagnostic evidence only.", + }, + { + "code": "no_price_write_from_pixels", + "status": "enforced", + "reason": ( + "Pixel evidence cannot directly write competitor_prices or " + "competitor_price_history." + ), + }, + { + "code": "no_external_embedding_api", + "status": "enforced", + "reason": "Embedding must stay Ollama-first and cannot use hosted visual APIs by default.", + }, + { + "code": "no_github_runtime_dependency", + "status": "enforced", + "reason": "The integration is derived from public research notes, not GitHub code.", + }, + { + "code": "polite_crawler_boundary", + "status": "enforced", + "reason": "Visual capture must respect delay, target allowlists, and read-only behavior.", + }, +) + + +def _merge_capabilities(overrides: Mapping[str, Any] | None) -> dict[str, bool]: + capabilities = dict(DEFAULT_CAPABILITIES) + for key, value in (overrides or {}).items(): + if key in capabilities: + capabilities[key] = bool(value) + return capabilities + + +def _phase_status_counts(phases: list[dict[str, Any]]) -> dict[str, int]: + counts: dict[str, int] = {} + for phase in phases: + status = str(phase.get("status") or "unknown") + counts[status] = counts.get(status, 0) + 1 + return counts + + +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 _dimension_payload(value: Mapping[str, Any] | None, fallback: Mapping[str, Any]) -> dict[str, Any]: + payload = dict(fallback) + for key, raw_value in (value or {}).items(): + if key in {"width", "height"}: + payload[key] = _positive_int(raw_value, int(fallback[key])) + elif key == "name": + payload[key] = str(raw_value or fallback.get("name") or "").strip() + return payload + + +def _controlled_apply_boundary() -> dict[str, Any]: + return { + "network_call": False, + "db_write": False, + "secret_read": False, + "github_dependency": False, + "production_price_write": False, + "rollback": "Disable the visual fallback selector or ignore generated artifacts.", + } + + +def _build_phases(capabilities: Mapping[str, bool]) -> list[dict[str, Any]]: + capture_ready = bool(capabilities["playwright_artifact_pipeline"]) + structured_ready = bool(capabilities["structured_crawler_api"]) + guard_ready = bool(capabilities["matcher_guardrails"]) + phase1_ready = capture_ready and structured_ready and guard_ready + embedding_ready = bool(capabilities["ollama_multimodal_embedding_ready"]) + visual_index_ready = bool(capabilities["pgvector_visual_index_ready"]) + faiss_allowed = bool(capabilities["faiss_allowed_in_production"]) + + return [ + { + "id": "PXR-1", + "name": "Visual capture fallback selector", + "status": "ready_to_start" if phase1_ready else "blocked_prerequisite", + "integration_point": [ + "MomoCrawler.search_products parser-empty fallback", + "PChomeCrawler search/detail parser anomaly fallback", + "market_intel.html_diagnostics low-confidence pages", + ], + "deliverable": "Emit a screenshot capture request when HTML/API evidence is missing.", + "writes": [], + "blocking_reason": "" if phase1_ready else "Needs structured crawler and Playwright artifact readiness.", + }, + { + "id": "PXR-2", + "name": "Tile manifest artifact", + "status": "ready_to_start" if phase1_ready else "blocked_prerequisite", + "integration_point": [ + "data/ai_automation visual artifacts", + "crawler diagnostics receipt", + ], + "deliverable": ( + "Record URL, viewport, tile coordinates, source crawler, parse failure, " + "and evidence intent without model inference." + ), + "writes": ["artifact_file_only"], + "blocking_reason": "" if phase1_ready else "Phase 1 selector must be available first.", + }, + { + "id": "PXR-3", + "name": "Ollama-first multimodal embedding benchmark", + "status": "ready_to_benchmark" if embedding_ready else "deferred_model_not_verified", + "integration_point": [ + "Hermes embedding lane", + "Ollama GCP-A -> GCP-B -> 111 routing", + ], + "deliverable": "Benchmark local visual embeddings on saved tiles before enabling retrieval.", + "writes": ["benchmark_artifact"], + "blocking_reason": "" if embedding_ready else "No verified Ollama-hosted visual embedding model yet.", + }, + { + "id": "PXR-4", + "name": "Visual retrieval index", + "status": ( + "ready_to_design" + if visual_index_ready + else "deferred_index_policy" + ), + "integration_point": [ + "pgvector-backed evidence index", + "AI knowledge retrieval", + ], + "deliverable": "Use pgvector-compatible visual evidence metadata before any FAISS adoption.", + "writes": ["design_artifact"], + "blocking_reason": ( + "" + if visual_index_ready + else "Production vector policy is pgvector-first; FAISS requires ADR or local-only proof." + ), + "faiss_allowed_in_production": faiss_allowed, + }, + { + "id": "PXR-5", + "name": "Crawler fusion and price-write guard", + "status": "deferred_replay_required", + "integration_point": [ + "marketplace_product_matcher", + "competitor_match_attempts", + "manual/AI verification queue", + ], + "deliverable": ( + "Fuse text/API evidence and visual evidence into review diagnostics; " + "do not auto-write formal price rows until replay proves confidence." + ), + "writes": ["review_diagnostics_only"], + "blocking_reason": "Needs replay/canary evidence before production price decisions.", + }, + ] + + +def should_emit_visual_evidence_fallback( + *, + parser_success: bool, + parsed_item_count: int = 0, + confidence_band: str = "", + missing_fields: tuple[str, ...] | list[str] | None = None, + failure_reason: str = "", +) -> dict[str, Any]: + """Decide whether a crawler result should emit a visual evidence manifest.""" + triggers: list[str] = [] + missing = [str(field) for field in (missing_fields or []) if str(field).strip()] + normalized_confidence = str(confidence_band or "").strip().lower() + normalized_failure = str(failure_reason or "").strip().lower() + + if not parser_success: + triggers.append("parser_failed") + if int(parsed_item_count or 0) <= 0: + triggers.append("parsed_empty") + if normalized_confidence in VISUAL_FALLBACK_CONFIDENCE_TRIGGERS: + triggers.append(f"confidence:{normalized_confidence}") + if any(field in {"price", "product_id", "title", "spec", "image_url"} for field in missing): + triggers.append("critical_field_missing") + if any(token in normalized_failure for token in ("html", "selector", "render", "visual", "price")): + triggers.append("failure_reason_visual_or_parser_related") + + should_emit = bool(triggers) + return { + "should_emit": should_emit, + "triggers": triggers, + "fallback_reason": ", ".join(triggers) if should_emit else "structured_evidence_sufficient", + "missing_fields": missing, + "parser_success": bool(parser_success), + "parsed_item_count": int(parsed_item_count or 0), + "confidence_band": confidence_band, + "policy": MANIFEST_POLICY, + } + + +def build_pixelrag_visual_evidence_manifest( + *, + url: str, + platform: str, + crawler: str, + trigger_reason: str, + evidence_intent: str = "recover_visual_offer_evidence", + viewport: Mapping[str, Any] | None = None, + page_size: Mapping[str, Any] | None = None, + tile_size: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Build a read-only visual evidence manifest without capturing the page.""" + errors: list[str] = [] + raw_url = str(url or "").strip() + parsed = urlparse(raw_url) + platform_code = str(platform or "").strip().lower() + crawler_name = str(crawler or "").strip() + reason = str(trigger_reason or "").strip() + intent = str(evidence_intent or "recover_visual_offer_evidence").strip() + + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + errors.append("URL must be an absolute http(s) URL.") + if parsed.username or parsed.password: + errors.append("URL credentials are not allowed in visual evidence manifests.") + if platform_code not in ALLOWED_PLATFORMS: + errors.append(f"platform must be one of: {', '.join(ALLOWED_PLATFORMS)}.") + if not crawler_name: + errors.append("crawler is required.") + if not reason: + errors.append("trigger_reason is required.") + + boundary = _controlled_apply_boundary() + if errors: + return { + "success": False, + "policy": MANIFEST_POLICY, + "result": RESULT_MANIFEST_REJECTED, + "errors": errors, + "controlled_apply": boundary, + } + + viewport_payload = _dimension_payload(viewport, DEFAULT_VIEWPORT) + tile_payload = _dimension_payload(tile_size, DEFAULT_TILE_SIZE) + page_payload = _dimension_payload( + page_size, + { + "name": "estimated-page", + "width": viewport_payload["width"], + "height": viewport_payload["height"], + }, + ) + tiles_x = max(1, math.ceil(page_payload["width"] / tile_payload["width"])) + tiles_y = max(1, math.ceil(page_payload["height"] / tile_payload["height"])) + tile_count = tiles_x * tiles_y + manifest_key = "|".join( + [ + platform_code, + crawler_name, + raw_url, + reason, + str(viewport_payload["width"]), + str(viewport_payload["height"]), + str(tile_payload["width"]), + str(tile_payload["height"]), + ] + ) + manifest_id = hashlib.sha256(manifest_key.encode("utf-8")).hexdigest()[:20] + + return { + "success": True, + "policy": MANIFEST_POLICY, + "generated_at": datetime.now(timezone.utc).isoformat(), + "result": RESULT_MANIFEST_READY, + "manifest_id": manifest_id, + "status": "capture_requested", + "capture_target": { + "url": raw_url, + "platform": platform_code, + "crawler": crawler_name, + "trigger_reason": reason, + "evidence_intent": intent, + }, + "viewport": viewport_payload, + "page_size": page_payload, + "tile_size": tile_payload, + "tile_plan": { + "tiles_x": tiles_x, + "tiles_y": tiles_y, + "tile_count": tile_count, + "overlap_px": 0, + }, + "artifact": { + "kind": "pixelrag_visual_evidence_manifest", + "suggested_path": ( + f"data/ai_automation/pixelrag_visual_evidence/" + f"{platform_code}/{manifest_id}.json" + ), + "writes": ["artifact_file_only"], + }, + "safety_guards": deepcopy(list(SAFETY_GUARDS)), + "controlled_apply": boundary, + "next_action": { + "action": "capture_screenshot_and_tiles_from_manifest", + "status": "ready_for_capture_worker", + "reason": "Manifest is validated; capture worker can run under read-only crawler policy.", + }, + } + + +def build_pixelrag_crawler_integration_assessment( + *, + capabilities: Mapping[str, Any] | None = None, + target_platforms: tuple[str, ...] | list[str] | None = None, +) -> dict[str, Any]: + """Build a safe integration assessment for PixelRAG-style crawler fallback.""" + merged_capabilities = _merge_capabilities(capabilities) + platforms = tuple(target_platforms or TARGET_PLATFORMS) + phases = _build_phases(merged_capabilities) + phase_counts = _phase_status_counts(phases) + + phase1_ready = all( + phase.get("status") == "ready_to_start" + for phase in phases[:2] + ) + full_pixelrag_ready = all( + merged_capabilities[key] + for key in ( + "playwright_artifact_pipeline", + "ollama_multimodal_embedding_ready", + "pgvector_visual_index_ready", + ) + ) and not merged_capabilities["faiss_allowed_in_production"] + + result = RESULT_PHASE1_READY if phase1_ready else RESULT_BLOCKED_NO_CAPTURE + recommended_mode = ( + "pixelrag_inspired_visual_evidence_fallback" + if phase1_ready + else "prepare_visual_capture_prerequisites" + ) + + payload = { + "success": True, + "policy": POLICY, + "generated_at": datetime.now(timezone.utc).isoformat(), + "result": result, + "feasible": phase1_ready, + "can_start_now": phase1_ready, + "full_pixelrag_ready": full_pixelrag_ready, + "recommended_mode": recommended_mode, + "plain_assessment": ( + "可導入，但第一階段應定位為爬蟲低信心時的視覺證據 fallback；" + "完整像素向量檢索需等 Ollama-first 視覺 embedding 與 pgvector/ADR 驗證。" + if phase1_ready + else "目前不應啟動，需先補齊 Playwright artifact 或 crawler guardrail 前置條件。" + ), + "target_platforms": list(platforms), + "capabilities": deepcopy(merged_capabilities), + "source_findings": deepcopy(list(SOURCE_FINDINGS)), + "crawler_fit": { + "best_for": [ + "dynamic pages whose rendered price/spec block is lost by HTML parsing", + "visual tables, badges, bundle/spec cards, and image-heavy offer evidence", + "MOMO/PChome parser-empty or identity-evidence-missing diagnostics", + ], + "not_for": [ + "stable public APIs that already return structured price and product IDs", + "bypassing robots, login walls, anti-bot controls, or rate limits", + "directly deciding exact product identity without matcher replay", + ], + "routing_rule": ( + "Use API/structured parser first; if parser evidence is empty or low-confidence, " + "emit visual evidence artifact; matcher remains the authority for identity." + ), + }, + "phases": phases, + "phase_status_counts": phase_counts, + "safety_guards": deepcopy(list(SAFETY_GUARDS)), + "controlled_apply": _controlled_apply_boundary(), + "next_actions": [ + { + "priority": "P0", + "action": "add_visual_evidence_manifest_lane", + "status": "ready" if phase1_ready else "blocked_prerequisite", + "reason": ( + "Starts crawler automation without changing formal price truth." + if phase1_ready + else "Requires Playwright artifact readiness before capture manifests can start." + ), + }, + { + "priority": "P1", + "action": "collect_replay_samples_from_parser_empty_cases", + "status": "ready_after_p0" if phase1_ready else "deferred", + "reason": "Builds evidence for whether visual capture improves coverage.", + }, + { + "priority": "P2", + "action": "benchmark_ollama_multimodal_embedding", + "status": "deferred", + "reason": "Needed before real pixel retrieval; cannot call hosted APIs by default.", + }, + ], + } + return payload + + +__all__ = [ + "MANIFEST_POLICY", + "POLICY", + "RESULT_PHASE1_READY", + "RESULT_BLOCKED_NO_CAPTURE", + "RESULT_MANIFEST_READY", + "RESULT_MANIFEST_REJECTED", + "build_pixelrag_crawler_integration_assessment", + "build_pixelrag_visual_evidence_manifest", + "should_emit_visual_evidence_fallback", +] diff --git a/tests/test_pixelrag_crawler_integration_service.py b/tests/test_pixelrag_crawler_integration_service.py new file mode 100644 index 0000000..bacb18c --- /dev/null +++ b/tests/test_pixelrag_crawler_integration_service.py @@ -0,0 +1,199 @@ +import json +import subprocess +import sys + + +def test_pixelrag_assessment_starts_visual_evidence_fallback_not_price_write(): + from services.pixelrag_crawler_integration_service import ( + RESULT_PHASE1_READY, + build_pixelrag_crawler_integration_assessment, + ) + + payload = build_pixelrag_crawler_integration_assessment() + + assert payload["success"] is True + assert payload["result"] == RESULT_PHASE1_READY + assert payload["feasible"] is True + assert payload["can_start_now"] is True + assert payload["recommended_mode"] == "pixelrag_inspired_visual_evidence_fallback" + assert payload["full_pixelrag_ready"] is False + assert payload["controlled_apply"] == { + "network_call": False, + "db_write": False, + "secret_read": False, + "github_dependency": False, + "production_price_write": False, + "rollback": "Disable the visual fallback selector or ignore generated artifacts.", + } + + phase_ids = {phase["id"]: phase for phase in payload["phases"]} + assert phase_ids["PXR-1"]["status"] == "ready_to_start" + assert phase_ids["PXR-2"]["writes"] == ["artifact_file_only"] + assert phase_ids["PXR-3"]["status"] == "deferred_model_not_verified" + assert phase_ids["PXR-5"]["writes"] == ["review_diagnostics_only"] + assert any( + guard["code"] == "no_price_write_from_pixels" + for guard in payload["safety_guards"] + ) + + +def test_pixelrag_assessment_blocks_when_visual_capture_is_not_ready(): + from services.pixelrag_crawler_integration_service import ( + RESULT_BLOCKED_NO_CAPTURE, + build_pixelrag_crawler_integration_assessment, + ) + + payload = build_pixelrag_crawler_integration_assessment( + capabilities={"playwright_artifact_pipeline": False} + ) + + assert payload["result"] == RESULT_BLOCKED_NO_CAPTURE + assert payload["feasible"] is False + assert payload["can_start_now"] is False + assert payload["recommended_mode"] == "prepare_visual_capture_prerequisites" + assert payload["phases"][0]["status"] == "blocked_prerequisite" + + +def test_pixelrag_assessment_keeps_github_and_hosted_embedding_out_of_runtime(): + from services.pixelrag_crawler_integration_service import ( + build_pixelrag_crawler_integration_assessment, + ) + + payload = build_pixelrag_crawler_integration_assessment( + capabilities={ + "ollama_multimodal_embedding_ready": True, + "pgvector_visual_index_ready": True, + } + ) + + findings = {finding["code"]: finding for finding in payload["source_findings"]} + guards = {guard["code"]: guard for guard in payload["safety_guards"]} + + assert payload["full_pixelrag_ready"] is True + assert "Ollama" in findings["qwen3_vl_embedding_optional"]["adoption"] + assert guards["no_external_embedding_api"]["status"] == "enforced" + assert guards["no_github_runtime_dependency"]["status"] == "enforced" + assert payload["controlled_apply"]["github_dependency"] is False + + +def test_pixelrag_report_cli_outputs_machine_readable_json(): + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/report_pixelrag_crawler_integration.py", + "--platform", + "momo", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["success"] is True + assert payload["target_platforms"] == ["momo"] + assert payload["next_actions"][0]["action"] == "add_visual_evidence_manifest_lane" + + +def test_visual_fallback_selector_routes_parser_empty_and_low_confidence_cases(): + from services.pixelrag_crawler_integration_service import ( + should_emit_visual_evidence_fallback, + ) + + parser_empty = should_emit_visual_evidence_fallback( + parser_success=False, + parsed_item_count=0, + missing_fields=["price"], + failure_reason="HTML selector did not find rendered price block", + ) + clean_parser = should_emit_visual_evidence_fallback( + parser_success=True, + parsed_item_count=8, + confidence_band="high", + ) + + assert parser_empty["should_emit"] is True + assert "parser_failed" in parser_empty["triggers"] + assert "critical_field_missing" in parser_empty["triggers"] + assert clean_parser["should_emit"] is False + assert clean_parser["fallback_reason"] == "structured_evidence_sufficient" + + +def test_visual_evidence_manifest_builds_tile_plan_without_runtime_writes(): + from services.pixelrag_crawler_integration_service import ( + RESULT_MANIFEST_READY, + build_pixelrag_visual_evidence_manifest, + ) + + payload = build_pixelrag_visual_evidence_manifest( + url="https://m.momoshop.com.tw/search.momo?searchKeyword=test", + platform="momo", + crawler="MomoCrawler.search_products", + trigger_reason="parser_empty", + viewport={"name": "desktop-1440", "width": 1440, "height": 950}, + page_size={"width": 1440, "height": 1900}, + tile_size={"width": 512, "height": 512}, + ) + + assert payload["success"] is True + assert payload["result"] == RESULT_MANIFEST_READY + assert payload["status"] == "capture_requested" + assert payload["tile_plan"] == { + "tiles_x": 3, + "tiles_y": 4, + "tile_count": 12, + "overlap_px": 0, + } + assert payload["artifact"]["writes"] == ["artifact_file_only"] + assert payload["controlled_apply"]["db_write"] is False + assert payload["controlled_apply"]["network_call"] is False + assert payload["controlled_apply"]["production_price_write"] is False + + +def test_visual_evidence_manifest_rejects_urls_with_credentials(): + from services.pixelrag_crawler_integration_service import ( + RESULT_MANIFEST_REJECTED, + build_pixelrag_visual_evidence_manifest, + ) + + payload = build_pixelrag_visual_evidence_manifest( + url="https://user:pass@example.test/product", + platform="momo", + crawler="MomoCrawler", + trigger_reason="parser_empty", + ) + + assert payload["success"] is False + assert payload["result"] == RESULT_MANIFEST_REJECTED + assert "URL credentials are not allowed" in payload["errors"][0] + + +def test_pixelrag_report_cli_outputs_visual_manifest(): + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/report_pixelrag_crawler_integration.py", + "--platform", + "pchome", + "--manifest-url", + "https://24h.pchome.com.tw/prod/TEST-000000001", + "--crawler", + "PChomeCrawler.search_products", + "--trigger-reason", + "parser_empty", + "--page-size", + "page=1024x1024", + "--tile-size", + "tile=512x512", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["result"] == "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY" + assert payload["capture_target"]["platform"] == "pchome" + assert payload["tile_plan"]["tile_count"] == 4