"""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", ]