"""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 json import math import os from copy import deepcopy from datetime import datetime, timezone from pathlib import Path 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" READBACK_POLICY = "read_only_pixelrag_visual_evidence_readback_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} def _default_artifact_root() -> str: container_root = Path("/app/data/ai_automation") if container_root.exists(): return str(container_root / "pixelrag_visual_evidence") return "runtime_artifacts/pixelrag_visual_evidence" DEFAULT_ARTIFACT_ROOT = os.getenv("PIXELRAG_VISUAL_EVIDENCE_ROOT", _default_artifact_root()) DEFAULT_ARTIFACT_MAX_AGE_HOURS = int(os.getenv("PIXELRAG_VISUAL_EVIDENCE_MAX_AGE_HOURS", "168")) 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 _parse_iso_datetime(value: Any) -> datetime | None: if not value: return None try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _safe_int(value: Any, default: int = 0) -> int: try: return int(value or default) except (TypeError, ValueError): return default 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"runtime_artifacts/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 def _receipt_candidates( root: Path, *, platform: str | None = None, manifest_id: str | None = None, ) -> list[Path]: if manifest_id and platform: candidate = root / platform / manifest_id / "capture_receipt.json" return [candidate] if candidate.exists() else [] pattern = "*/capture_receipt.json" if platform and manifest_id else "*/**/capture_receipt.json" search_root = root / platform if platform and not manifest_id else root if not search_root.exists(): return [] return sorted(search_root.glob(pattern), key=lambda path: path.stat().st_mtime, reverse=True) def build_pixelrag_visual_evidence_readback( *, artifact_root: str | Path | None = None, platform: str | None = None, manifest_id: str | None = None, max_age_hours: int | None = None, ) -> dict[str, Any]: """Read latest PixelRAG capture receipt and file presence without running capture.""" root = Path(artifact_root or DEFAULT_ARTIFACT_ROOT) max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS)) platform_filter = str(platform or "").strip().lower() or None manifest_filter = str(manifest_id or "").strip() or None candidates = _receipt_candidates(root, platform=platform_filter, manifest_id=manifest_filter) now = datetime.now(timezone.utc) if not candidates: return { "success": False, "policy": READBACK_POLICY, "status": "warning", "artifact_root": str(root), "platform": platform_filter, "manifest_id": manifest_filter, "summary": { "receipt_count": 0, "tile_file_count": 0, "missing_file_count": 0, "age_hours": None, "stale": True, }, "latest_receipt": None, "next_machine_action": "run_pixelrag_visual_capture_worker", "controlled_apply": { "network_call": False, "db_write": False, "secret_read": False, "production_price_write": False, }, } latest_path = candidates[0] errors: list[str] = [] try: receipt = json.loads(latest_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: receipt = {} errors.append(str(exc)[:300]) generated_at = _parse_iso_datetime(receipt.get("generated_at")) age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None stale = age_hours is None or age_hours > max_age files = list(receipt.get("files") or []) file_statuses = [] missing_file_count = 0 tile_file_count = 0 for item in files: path = Path(str(item.get("path") or "")) exists = path.exists() if not exists: missing_file_count += 1 if item.get("kind") == "tile": tile_file_count += 1 file_statuses.append({ "kind": item.get("kind"), "path": str(path), "exists": exists, }) capture_target = receipt.get("capture_target") or {} tile_plan = receipt.get("tile_plan") or {} status = "ok" if receipt.get("status") == "captured" and missing_file_count == 0 and not stale else "warning" if errors: status = "critical" return { "success": not errors, "policy": READBACK_POLICY, "status": status, "artifact_root": str(root), "platform": platform_filter, "manifest_id": manifest_filter or receipt.get("manifest_id"), "summary": { "receipt_count": len(candidates), "tile_file_count": tile_file_count, "missing_file_count": missing_file_count, "planned_tile_count": _safe_int(tile_plan.get("planned_tile_count")), "emitted_tile_count": _safe_int(tile_plan.get("emitted_tile_count")), "http_status": _safe_int(receipt.get("http_status")), "age_hours": round(age_hours, 3) if age_hours is not None else None, "stale": stale, "errors": errors, }, "latest_receipt": { "path": str(latest_path), "generated_at": receipt.get("generated_at"), "capture_status": receipt.get("status"), "capture_target": capture_target, "page_metrics": receipt.get("page_metrics") or {}, "tile_plan": tile_plan, "files": file_statuses, }, "next_machine_action": ( "keep_pixelrag_visual_evidence_receipt" if status == "ok" else "run_pixelrag_visual_capture_worker" ), "controlled_apply": { "network_call": False, "db_write": False, "secret_read": False, "production_price_write": False, }, } __all__ = [ "MANIFEST_POLICY", "POLICY", "READBACK_POLICY", "RESULT_PHASE1_READY", "RESULT_BLOCKED_NO_CAPTURE", "RESULT_MANIFEST_READY", "RESULT_MANIFEST_REJECTED", "build_pixelrag_crawler_integration_assessment", "build_pixelrag_visual_evidence_readback", "build_pixelrag_visual_evidence_manifest", "should_emit_visual_evidence_fallback", ]