"""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 quote_plus, 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" ECOMMERCE_EXPANSION_POLICY = "read_only_pixelrag_ecommerce_expansion_plan_v1" CRAWLER_FAILURE_ACTION_POLICY = "read_only_pixelrag_crawler_failure_action_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" DEFAULT_VIEWPORT = {"name": "desktop-1440", "width": 1440, "height": 950} DEFAULT_TILE_SIZE = {"width": 512, "height": 512} ECOMMERCE_PLATFORM_PROFILES: dict[str, dict[str, Any]] = { "momo": { "display_name": "MOMO", "hosts": ["m.momoshop.com.tw", "www.momoshop.com.tw"], "search_url_template": "https://m.momoshop.com.tw/search.momo?searchKeyword={query}", "priority": "incumbent_structured_parser_plus_visual_fallback", "best_for": ["rendered promo blocks", "bundle/spec cards", "parser-empty search pages"], "data_targets": ["title", "price", "bundle_spec", "promo_badge", "image"], }, "pchome": { "display_name": "PChome 24h", "hosts": ["24h.pchome.com.tw", "ecshweb.pchome.com.tw", "ecapi-cdn.pchome.com.tw"], "search_url_template": "https://24h.pchome.com.tw/search/?q={query}", "priority": "incumbent_api_parser_plus_visual_fallback", "best_for": ["search result cards", "campaign badges", "rendered price cards"], "data_targets": ["title", "price", "product_id", "campaign_badge", "stock_hint"], }, "shopee_tw": { "display_name": "Shopee Taiwan", "hosts": ["shopee.tw", "mall.shopee.tw"], "search_url_template": "https://shopee.tw/search?keyword={query}", "priority": "P1_marketplace_expansion", "best_for": ["JS-heavy marketplace cards", "seller badges", "shipping/promo chips"], "data_targets": ["title", "price", "shop", "sold_count", "rating", "shipping_badge"], }, "coupang_tw": { "display_name": "Coupang Taiwan", "hosts": ["www.tw.coupang.com", "tw.coupang.com"], "search_url_template": "https://www.tw.coupang.com/search?q={query}", "priority": "P1_marketplace_expansion", "best_for": ["Rocket delivery badges", "discount panels", "spec tables"], "data_targets": ["title", "price", "discount", "delivery_badge", "spec", "review_count"], }, "yahoo_shopping_tw": { "display_name": "Yahoo Shopping Taiwan", "hosts": ["tw.buy.yahoo.com"], "search_url_template": "https://tw.buy.yahoo.com/search/product?p={query}", "priority": "P2_marketplace_expansion", "best_for": ["search cards", "campaign coupons", "rating and sales badges"], "data_targets": ["title", "price", "coupon", "rating", "sales_hint"], }, "etmall_tw": { "display_name": "ETMall", "hosts": ["www.etmall.com.tw", "etmall.com.tw"], "search_url_template": "https://www.etmall.com.tw/Search?keyword={query}", "priority": "P2_marketplace_expansion", "best_for": ["promotion blocks", "coupon and member badges"], "data_targets": ["title", "price", "promo_badge", "member_price", "shipping_hint"], }, "friday_tw": { "display_name": "friDay Shopping", "hosts": ["shopping.friday.tw", "ec-m.shopping.friday.tw", "ec-w.shopping.friday.tw"], "search_url_template": "https://ec-m.shopping.friday.tw/search?keyword={query}", "priority": "P3_probe_required", "best_for": ["campaign cards", "member reward hints", "mobile rendered offers"], "data_targets": ["title", "price", "reward_badge", "campaign_badge"], }, "rakuten_tw": { "display_name": "Rakuten Taiwan", "hosts": ["www.rakuten.com.tw", "rakuten.com.tw"], "search_url_template": "https://www.rakuten.com.tw/search/{query}/", "priority": "P3_probe_required", "best_for": ["storefront cards", "cross-store price comparisons", "coupon badges"], "data_targets": ["title", "price", "shop", "coupon", "point_reward"], }, "market_intel": { "display_name": "Market intelligence URL", "hosts": [], "search_url_template": "", "priority": "custom_url_only", "best_for": ["controlled external market diagnostics"], "data_targets": ["public_page_visual_evidence"], }, "external_market": { "display_name": "External market URL", "hosts": [], "search_url_template": "", "priority": "custom_url_only", "best_for": ["controlled public external URL diagnostics"], "data_targets": ["public_page_visual_evidence"], }, } ALLOWED_PLATFORMS = tuple(ECOMMERCE_PLATFORM_PROFILES) MARKETPLACE_EXPANSION_ORDER = ( "momo", "pchome", "shopee_tw", "coupang_tw", "yahoo_shopping_tw", "etmall_tw", "friday_tw", "rakuten_tw", ) 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" def _default_manifest_queue_root() -> str: container_root = Path("/app/data/ai_automation") if container_root.exists(): return str(container_root / "pixelrag_manifest_queue") return "runtime_artifacts/pixelrag_manifest_queue" DEFAULT_ARTIFACT_ROOT = os.getenv("PIXELRAG_VISUAL_EVIDENCE_ROOT", _default_artifact_root()) DEFAULT_MANIFEST_QUEUE_ROOT = os.getenv( "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_QUEUE_ROOT", _default_manifest_queue_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 _visual_barrier_reason(receipt: Mapping[str, Any]) -> str: http_status = _safe_int(receipt.get("http_status")) page_metrics = receipt.get("page_metrics") or {} final_url = str(page_metrics.get("final_url") or "").lower() title = str(page_metrics.get("title") or "").lower() if http_status >= 400: return f"http_status_{http_status}" barrier_tokens = ( "access denied", "verify/traffic/error", "captcha", "blocked", "robot", ) haystack = f"{final_url} {title}" for token in barrier_tokens: if token in haystack: return token.replace("/", "_").replace(" ", "_") return "" def _platform_profile(platform: str) -> dict[str, Any] | None: return ECOMMERCE_PLATFORM_PROFILES.get(str(platform or "").strip().lower()) def _search_url_for_platform(platform: str, keyword: str) -> str: profile = _platform_profile(platform) or {} template = str(profile.get("search_url_template") or "") query = quote_plus(str(keyword or "").strip()) return template.format(query=query) if template and query else "" 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_ecommerce_expansion_plan( *, target_platforms: tuple[str, ...] | list[str] | None = None, ) -> dict[str, Any]: """Return a machine-readable PixelRAG rollout plan for public commerce sites.""" requested = [ str(platform or "").strip().lower() for platform in (target_platforms or MARKETPLACE_EXPANSION_ORDER) if str(platform or "").strip() ] selected = [ platform for platform in requested if platform in ECOMMERCE_PLATFORM_PROFILES ] unknown = [ platform for platform in requested if platform not in ECOMMERCE_PLATFORM_PROFILES ] profiles = [] for platform in selected: profile = deepcopy(ECOMMERCE_PLATFORM_PROFILES[platform]) profile["platform"] = platform profile["manifest_ready"] = bool(profile.get("search_url_template")) profile["capture_scope"] = "public_read_only_no_login_no_cart" profiles.append(profile) return { "success": True, "policy": ECOMMERCE_EXPANSION_POLICY, "generated_at": datetime.now(timezone.utc).isoformat(), "status": "marketplace_profiles_ready", "platform_count": len(profiles), "unknown_platforms": unknown, "platforms": profiles, "advantages": [ "Covers JS-heavy cards and visual promotion badges that HTML parsers often miss.", "Creates replayable screenshots and tiles before adding platform-specific parsers.", "Lets AI compare public offer presentation across marketplaces without writing price truth.", "Keeps each platform behind host allowlists and read-only artifact boundaries.", ], "limitations": [ "Screenshots are slower and heavier than structured APIs.", "Visual evidence is not a formal price source until OCR/VLM replay is verified.", "Login-only, cart-only, coupon-personalized, or anti-bot-gated states are out of scope.", "Search URL patterns still need runtime probes before production scheduling per platform.", ], "next_actions": [ { "priority": "P0", "action": "capture_public_search_visual_receipts", "target_platforms": ["shopee_tw", "coupang_tw"], "status": "ready_for_controlled_probe", }, { "priority": "P1", "action": "add_visual_card_extraction_replay", "target_fields": ["title", "price", "promo_badge", "shop", "rating"], "status": "requires_receipt_samples", }, { "priority": "P2", "action": "promote_stable_platform_parser", "status": "only_after_replay_beats_visual_only_diagnostics", }, ], "controlled_apply": _controlled_apply_boundary(), } def build_pixelrag_marketplace_search_manifest( *, platform: str, keyword: str, crawler: str = "PixelRAGMarketplaceSearch.visual_fallback", trigger_reason: str = "marketplace_search_visual_scan", evidence_intent: str = "collect_public_marketplace_offer_cards", 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 visual evidence manifest from a public marketplace keyword.""" platform_code = str(platform or "").strip().lower() keyword_text = str(keyword or "").strip() profile = _platform_profile(platform_code) errors: list[str] = [] if not profile: errors.append(f"platform must be one of: {', '.join(ALLOWED_PLATFORMS)}.") if not keyword_text: errors.append("keyword is required.") search_url = _search_url_for_platform(platform_code, keyword_text) if profile and not search_url: errors.append("platform does not define a public search URL template.") if errors: return { "success": False, "policy": MANIFEST_POLICY, "result": RESULT_MANIFEST_REJECTED, "errors": errors, "controlled_apply": _controlled_apply_boundary(), } manifest = build_pixelrag_visual_evidence_manifest( url=search_url, platform=platform_code, crawler=crawler, trigger_reason=trigger_reason, evidence_intent=evidence_intent, viewport=viewport, page_size=page_size, tile_size=tile_size, ) if manifest.get("success"): manifest["marketplace_profile"] = { "platform": platform_code, "display_name": profile.get("display_name"), "hosts": profile.get("hosts"), "priority": profile.get("priority"), "data_targets": profile.get("data_targets"), "keyword": keyword_text, } return manifest 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 persist_pixelrag_visual_evidence_manifest( manifest: Mapping[str, Any], *, queue_root: str | Path | None = None, ) -> dict[str, Any]: """Persist a capture manifest to the local artifact queue.""" root = Path(queue_root or DEFAULT_MANIFEST_QUEUE_ROOT) platform = str((manifest.get("capture_target") or {}).get("platform") or "unknown").strip().lower() manifest_id = str(manifest.get("manifest_id") or "").strip() errors: list[str] = [] if manifest.get("policy") != MANIFEST_POLICY: errors.append(f"manifest policy must be {MANIFEST_POLICY}") if not manifest.get("success"): errors.append("manifest must be successful before persistence") if not manifest_id: errors.append("manifest_id is required") if platform not in ALLOWED_PLATFORMS: errors.append(f"platform must be one of: {', '.join(ALLOWED_PLATFORMS)}.") if errors: return { "success": False, "status": "rejected", "errors": errors, "queue_root": str(root), "controlled_apply": { "artifact_write": False, "db_write": False, "secret_read": False, "production_price_write": False, }, } target_dir = root / platform target_dir.mkdir(parents=True, exist_ok=True) target_path = target_dir / f"{manifest_id}.json" temp_path = target_path.with_suffix(".json.tmp") payload = dict(manifest) payload["queued_at"] = datetime.now(timezone.utc).isoformat() payload["queue_policy"] = CRAWLER_FAILURE_ACTION_POLICY temp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") temp_path.replace(target_path) return { "success": True, "status": "queued", "path": str(target_path), "queue_root": str(root), "controlled_apply": { "artifact_write": True, "db_write": False, "secret_read": False, "production_price_write": False, }, } def build_pixelrag_crawler_failure_action( *, url: str, platform: str, crawler: str, parser_success: bool, parsed_item_count: int = 0, confidence_band: str = "", missing_fields: tuple[str, ...] | list[str] | None = None, failure_reason: str = "", evidence_intent: str = "recover_visual_offer_evidence", persist_manifest: bool = False, queue_root: str | Path | None = None, viewport: Mapping[str, Any] | None = None, page_size: Mapping[str, Any] | None = None, tile_size: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Turn a crawler parse failure into a queued PixelRAG capture action.""" selector = should_emit_visual_evidence_fallback( parser_success=parser_success, parsed_item_count=parsed_item_count, confidence_band=confidence_band, missing_fields=missing_fields, failure_reason=failure_reason, ) if not selector["should_emit"]: return { "success": True, "policy": CRAWLER_FAILURE_ACTION_POLICY, "status": "skipped_structured_evidence_sufficient", "selector": selector, "manifest": None, "persistence": None, "next_machine_action": "keep_structured_crawler_result", "controlled_apply": _controlled_apply_boundary(), } manifest = build_pixelrag_visual_evidence_manifest( url=url, platform=platform, crawler=crawler, trigger_reason=selector["fallback_reason"], evidence_intent=evidence_intent, viewport=viewport, page_size=page_size, tile_size=tile_size, ) persistence = None if persist_manifest and manifest.get("success"): persistence = persist_pixelrag_visual_evidence_manifest( manifest, queue_root=queue_root, ) manifest_ready = bool(manifest.get("success")) queued = bool((persistence or {}).get("success")) return { "success": manifest_ready and (not persist_manifest or queued), "policy": CRAWLER_FAILURE_ACTION_POLICY, "generated_at": datetime.now(timezone.utc).isoformat(), "status": ( "manifest_persisted" if queued else ("manifest_ready" if manifest_ready else "manifest_rejected") ), "selector": selector, "manifest": manifest, "persistence": persistence, "next_machine_action": ( "run_pixelrag_visual_capture_worker" if manifest_ready else "fix_manifest_input" ), "controlled_apply": { "network_call": False, "artifact_write": bool(queued), "db_write": False, "secret_read": False, "production_price_write": False, }, } 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 _portable_receipt_file_path(receipt_path: Path, raw_path: Any, kind: Any) -> Path: path = Path(str(raw_path or "")) if path.exists(): return path receipt_dir = receipt_path.parent basename = path.name if not basename: return path if kind == "tile": tile_path = receipt_dir / "tiles" / basename if tile_path.exists(): return tile_path sibling_path = receipt_dir / basename if sibling_path.exists(): return sibling_path parts = path.parts if receipt_dir.name in parts: index = parts.index(receipt_dir.name) suffix = parts[index + 1:] if suffix: relocated = receipt_dir.joinpath(*suffix) if relocated.exists(): return relocated return path 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 = _portable_receipt_file_path(latest_path, item.get("path"), item.get("kind")) 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 {} page_metrics = receipt.get("page_metrics") or {} http_status = _safe_int(receipt.get("http_status")) barrier_reason = _visual_barrier_reason(receipt) status = ( "ok" if ( receipt.get("status") == "captured" and missing_file_count == 0 and not stale and not barrier_reason ) 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": http_status, "age_hours": round(age_hours, 3) if age_hours is not None else None, "stale": stale, "visual_barrier_detected": bool(barrier_reason), "visual_barrier_reason": barrier_reason, "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": page_metrics, "tile_plan": tile_plan, "files": file_statuses, }, "next_machine_action": ( "keep_pixelrag_visual_evidence_receipt" if status == "ok" else ( "run_platform_probe_or_use_structured_api" if barrier_reason else "run_pixelrag_visual_capture_worker" ) ), "controlled_apply": { "network_call": False, "db_write": False, "secret_read": False, "production_price_write": False, }, } __all__ = [ "CRAWLER_FAILURE_ACTION_POLICY", "ECOMMERCE_EXPANSION_POLICY", "ECOMMERCE_PLATFORM_PROFILES", "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_crawler_failure_action", "build_pixelrag_ecommerce_expansion_plan", "build_pixelrag_marketplace_search_manifest", "build_pixelrag_visual_evidence_readback", "build_pixelrag_visual_evidence_manifest", "persist_pixelrag_visual_evidence_manifest", "should_emit_visual_evidence_fallback", ]