From c3824119cd4ca70fdd70efa86f72505c80f4e785 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:09:00 +0800 Subject: [PATCH] feat(crawler): expand pixelrag marketplace automation --- .gitignore | 1 + docs/guides/browse_sh_crawler_playbook.md | 38 +- routes/system_public_routes.py | 16 + .../ops/capture_pixelrag_visual_evidence.js | 6 + .../ops/capture_pixelrag_visual_evidence.py | 6 + .../report_pixelrag_crawler_integration.py | 45 ++ services/pchome_crawler.py | 83 ++++ .../pixelrag_crawler_integration_service.py | 439 +++++++++++++++++- tests/test_pchome_crawler_search.py | 26 ++ ...st_pixelrag_crawler_integration_service.py | 131 ++++++ 10 files changed, 784 insertions(+), 7 deletions(-) diff --git a/.gitignore b/.gitignore index 8a8d457..5126dbc 100644 --- a/.gitignore +++ b/.gitignore @@ -91,6 +91,7 @@ data/excel_exports/ data/*_cache/ data/ai_automation_smoke_history.jsonl data/ai_automation/sitewide_visual_qa_latest.json +runtime_artifacts/ # 上傳檔案 web/static/uploads/ diff --git a/docs/guides/browse_sh_crawler_playbook.md b/docs/guides/browse_sh_crawler_playbook.md index ba291ea..39f1bc6 100644 --- a/docs/guides/browse_sh_crawler_playbook.md +++ b/docs/guides/browse_sh_crawler_playbook.md @@ -54,7 +54,7 @@ python scripts/tools/browse_sh_probe.py -- screenshot ## PixelRAG-style 視覺證據 fallback -2026-07-09 評估 PixelRAG 後，結論是「可導入，但不可直接取代正式爬蟲」。PixelRAG 的核心價值是把渲染後頁面截圖切成 tiles，讓 AI 讀到 HTML parser 可能丟失的視覺結構；本專案第一階段只採用這個視覺證據思路，不直接拉外部 runtime、不用外部 embedding API、不把像素結果寫入正式價格表。 +2026-07-09 評估 PixelRAG 後，結論是「可導入，而且不應只限 MOMO/PChome」。PixelRAG 的核心價值是把渲染後頁面截圖切成 tiles，讓 AI 讀到 HTML parser 可能丟失的視覺結構；本專案第一階段採用這個視覺證據思路，擴充成多電商 public marketplace visual evidence layer，不直接拉外部 runtime、不用外部 embedding API、不把像素結果寫入正式價格表。 導入順序： @@ -65,11 +65,44 @@ python scripts/tools/browse_sh_probe.py -- screenshot 5. Phase 4 若需要索引，優先設計 pgvector-compatible evidence metadata；FAISS 只能先當本地研究/ADR 候選，不直接進 production。 6. Phase 5 才談 crawler fusion；正式 `competitor_prices` / `competitor_price_history` 寫入仍需 matcher replay/canary 證據。 +多電商優先順序： + +| Priority | Platform | PixelRAG 用途 | 邊界 | +|---|---|---|---| +| P0 | MOMO / PChome | 既有 structured crawler 失敗時自動產生 visual manifest queue | 正式價格仍以既有 parser/API/matcher 為準 | +| P1 | Shopee Taiwan / Coupang Taiwan | 抓公共搜尋頁商品卡、促銷 badge、店家/配送/評分視覺證據 | 不登入、不碰購物車、不繞過 anti-bot | +| P2 | Yahoo Shopping / ETMall | 補促銷券、快速到貨、活動價格等視覺訊號 | 先做 controlled probe，再決定 parser | +| P3 | friDay / Rakuten Taiwan | 先建立 probe profile 與 search manifest | URL pattern 需 runtime receipt 驗證後再排程 | + +已支援的 platform code： + +```text +momo +pchome +shopee_tw +coupang_tw +yahoo_shopping_tw +etmall_tw +friday_tw +rakuten_tw +``` + +自動觸發： + +- `PChomeCrawler.fetch_region_page` 找不到商品 ID 時會 queue PixelRAG manifest。 +- `PChomeCrawler.fetch_product_details` API 回傳但無可解析商品時會 queue 商品頁 manifest。 +- `PChomeCrawler.search_products` 搜尋無商品 ID 或 request error 時會 queue 公開搜尋頁 manifest。 +- queue 預設路徑：正式 app 容器 `/app/data/ai_automation/pixelrag_manifest_queue`，本機為 `runtime_artifacts/pixelrag_manifest_queue`。 +- queue 只寫 artifact，不寫 DB、不寫正式價格、不讀 secret；capture worker 後續消費 manifest 產生 screenshot/tile receipt。 + 機器可讀評估： ```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 --ecommerce-expansion-plan +python scripts/ops/report_pixelrag_crawler_integration.py --platform shopee_tw --marketplace-keyword "防曬乳" +python scripts/ops/report_pixelrag_crawler_integration.py --platform coupang_tw --marketplace-keyword "iphone" 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 \ @@ -84,6 +117,9 @@ python scripts/ops/report_pixelrag_crawler_integration.py \ ```text /api/ai-automation/pixelrag-crawler-integration?platform=momo +/api/ai-automation/pixelrag-crawler-integration?expansion_plan=true +/api/ai-automation/pixelrag-crawler-integration?platform=shopee_tw&marketplace_keyword=%E9%98%B2%E6%9B%AC%E4%B9%B3 +/api/ai-automation/pixelrag-crawler-integration?platform=coupang_tw&marketplace_keyword=iphone /api/ai-automation/pixelrag-crawler-integration?platform=pchome&manifest_url=https://24h.pchome.com.tw/prod/TEST-000000001&crawler=PChomeCrawler.search_products&trigger_reason=parser_empty /api/ai-automation/pixelrag-visual-evidence-readback?platform=pchome&manifest_id=4a93e95e5afb414bc8c3 ``` diff --git a/routes/system_public_routes.py b/routes/system_public_routes.py index 82d4fe9..030f885 100644 --- a/routes/system_public_routes.py +++ b/routes/system_public_routes.py @@ -640,10 +640,26 @@ def ai_automation_pixelrag_crawler_integration_api(): """Read-only PixelRAG-style crawler visual evidence assessment.""" from services.pixelrag_crawler_integration_service import ( build_pixelrag_crawler_integration_assessment, + build_pixelrag_ecommerce_expansion_plan, + build_pixelrag_marketplace_search_manifest, build_pixelrag_visual_evidence_manifest, ) platforms = request.args.getlist('platform') or None + if str(request.args.get('expansion_plan') or '').lower() in {'1', 'true', 'yes'}: + return jsonify(build_pixelrag_ecommerce_expansion_plan( + target_platforms=tuple(platforms) if platforms else None, + )) + marketplace_keyword = str(request.args.get('marketplace_keyword') or '').strip() + if marketplace_keyword: + platform = (platforms or ['shopee_tw'])[0] + return jsonify(build_pixelrag_marketplace_search_manifest( + platform=platform, + keyword=marketplace_keyword, + crawler=str(request.args.get('crawler') or 'PixelRAGMarketplaceSearch.visual_fallback'), + trigger_reason=str(request.args.get('trigger_reason') or 'marketplace_search_visual_scan'), + evidence_intent=str(request.args.get('evidence_intent') or 'collect_public_marketplace_offer_cards'), + )) manifest_url = str(request.args.get('manifest_url') or '').strip() if manifest_url: platform = (platforms or ['momo'])[0] diff --git a/scripts/ops/capture_pixelrag_visual_evidence.js b/scripts/ops/capture_pixelrag_visual_evidence.js index 8973a57..beea5ae 100644 --- a/scripts/ops/capture_pixelrag_visual_evidence.js +++ b/scripts/ops/capture_pixelrag_visual_evidence.js @@ -21,6 +21,12 @@ const DEFAULT_MAX_TILES = 80; const ALLOWED_PLATFORM_HOSTS = { momo: new Set(['m.momoshop.com.tw', 'www.momoshop.com.tw']), pchome: new Set(['24h.pchome.com.tw', 'ecshweb.pchome.com.tw', 'ecapi-cdn.pchome.com.tw']), + shopee_tw: new Set(['shopee.tw', 'mall.shopee.tw']), + coupang_tw: new Set(['www.tw.coupang.com', 'tw.coupang.com']), + yahoo_shopping_tw: new Set(['tw.buy.yahoo.com']), + etmall_tw: new Set(['www.etmall.com.tw', 'etmall.com.tw']), + friday_tw: new Set(['shopping.friday.tw', 'ec-m.shopping.friday.tw', 'ec-w.shopping.friday.tw']), + rakuten_tw: new Set(['www.rakuten.com.tw', 'rakuten.com.tw']), market_intel: new Set(), external_market: new Set(), }; diff --git a/scripts/ops/capture_pixelrag_visual_evidence.py b/scripts/ops/capture_pixelrag_visual_evidence.py index 6263cd2..f9c2a1f 100644 --- a/scripts/ops/capture_pixelrag_visual_evidence.py +++ b/scripts/ops/capture_pixelrag_visual_evidence.py @@ -26,6 +26,12 @@ DEFAULT_MAX_TILES = 80 ALLOWED_PLATFORM_HOSTS = { "momo": {"m.momoshop.com.tw", "www.momoshop.com.tw"}, "pchome": {"24h.pchome.com.tw", "ecshweb.pchome.com.tw", "ecapi-cdn.pchome.com.tw"}, + "shopee_tw": {"shopee.tw", "mall.shopee.tw"}, + "coupang_tw": {"www.tw.coupang.com", "tw.coupang.com"}, + "yahoo_shopping_tw": {"tw.buy.yahoo.com"}, + "etmall_tw": {"www.etmall.com.tw", "etmall.com.tw"}, + "friday_tw": {"shopping.friday.tw", "ec-m.shopping.friday.tw", "ec-w.shopping.friday.tw"}, + "rakuten_tw": {"www.rakuten.com.tw", "rakuten.com.tw"}, "market_intel": set(), "external_market": set(), } diff --git a/scripts/ops/report_pixelrag_crawler_integration.py b/scripts/ops/report_pixelrag_crawler_integration.py index 0a9f226..ef95edd 100644 --- a/scripts/ops/report_pixelrag_crawler_integration.py +++ b/scripts/ops/report_pixelrag_crawler_integration.py @@ -15,6 +15,8 @@ if str(ROOT) not in sys.path: from services.pixelrag_crawler_integration_service import ( # noqa: E402 build_pixelrag_crawler_integration_assessment, + build_pixelrag_ecommerce_expansion_plan, + build_pixelrag_marketplace_search_manifest, build_pixelrag_visual_evidence_manifest, ) @@ -62,6 +64,12 @@ def main() -> int: help="覆寫 readiness 假設，用於模擬下一階段條件。", ) parser.add_argument("--manifest-url", help="輸出單一 URL 的視覺證據 manifest。") + parser.add_argument("--marketplace-keyword", help="依平台與關鍵字輸出多電商視覺搜尋 manifest。") + parser.add_argument( + "--ecommerce-expansion-plan", + action="store_true", + help="輸出 MOMO/PChome 以外的 PixelRAG 多電商擴充計畫。", + ) parser.add_argument("--crawler", default="crawler_diagnostics", help="manifest 來源 crawler 名稱。") parser.add_argument( "--trigger-reason", @@ -90,6 +98,43 @@ def main() -> int: ) args = parser.parse_args() + if args.ecommerce_expansion_plan: + payload = build_pixelrag_ecommerce_expansion_plan( + 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 args.marketplace_keyword: + platform = (args.platforms or ["shopee_tw"])[0] + crawler = ( + "PixelRAGMarketplaceSearch.visual_fallback" + if args.crawler == "crawler_diagnostics" + else args.crawler + ) + trigger_reason = ( + "marketplace_search_visual_scan" + if args.trigger_reason == "parser_empty_or_low_confidence" + else args.trigger_reason + ) + evidence_intent = ( + "collect_public_marketplace_offer_cards" + if args.evidence_intent == "recover_visual_offer_evidence" + else args.evidence_intent + ) + payload = build_pixelrag_marketplace_search_manifest( + platform=platform, + keyword=args.marketplace_keyword, + crawler=crawler, + trigger_reason=trigger_reason, + evidence_intent=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 + if args.manifest_url: platform = (args.platforms or ["momo"])[0] payload = build_pixelrag_visual_evidence_manifest( diff --git a/services/pchome_crawler.py b/services/pchome_crawler.py index bcd4f8a..8b2158f 100644 --- a/services/pchome_crawler.py +++ b/services/pchome_crawler.py @@ -18,6 +18,7 @@ import logging from typing import List, Dict, Optional, Tuple from dataclasses import dataclass, asdict from datetime import datetime +from urllib.parse import urlencode import requests @@ -139,6 +140,7 @@ class PChomeCrawler: self.session = requests.Session() self.session.headers.update(self.DEFAULT_HEADERS) self._last_request_time = 0 + self.last_visual_evidence_action = None def _rate_limit(self): """速率限制""" @@ -188,6 +190,58 @@ class PChomeCrawler: raise last_error raise requests.RequestException(f"GET failed: {url}") + def _public_search_url(self, keyword: str) -> str: + return f"{self.BASE_URL}/search/?{urlencode({'q': str(keyword or '').strip()})}" + + def _emit_visual_evidence_action( + self, + *, + url: str, + crawler: str, + failure_reason: str, + parsed_item_count: int = 0, + missing_fields: Tuple[str, ...] = ("product_id", "title", "price"), + ) -> Optional[dict]: + """Queue a PixelRAG visual manifest for parser-empty crawler cases.""" + try: + from services.pixelrag_crawler_integration_service import ( + build_pixelrag_crawler_failure_action, + ) + + action = build_pixelrag_crawler_failure_action( + url=url, + platform="pchome", + crawler=crawler, + parser_success=False, + parsed_item_count=parsed_item_count, + missing_fields=missing_fields, + failure_reason=failure_reason, + persist_manifest=True, + ) + self.last_visual_evidence_action = action + if action.get("success"): + logger.info( + "[PChome] PixelRAG visual evidence manifest queued | crawler=%s status=%s path=%s", + crawler, + action.get("status"), + ((action.get("persistence") or {}).get("path") or ""), + ) + else: + logger.warning( + "[PChome] PixelRAG visual evidence action failed | crawler=%s status=%s", + crawler, + action.get("status"), + ) + return action + except Exception as exc: + logger.warning("[PChome] PixelRAG visual evidence action skipped: %s", exc) + self.last_visual_evidence_action = { + "success": False, + "status": "exception", + "error": str(exc)[:300], + } + return self.last_visual_evidence_action + def _normalize_product_id(self, product_id: str) -> str: """ 正規化商品 ID 格式 @@ -248,11 +302,22 @@ class PChomeCrawler: product_ids = self._extract_product_ids_from_html(response.text) logger.info(f"從 {url} 取得 {len(product_ids)} 個商品 ID") + if not product_ids: + self._emit_visual_evidence_action( + url=url, + crawler="PChomeCrawler.fetch_region_page", + failure_reason="HTML parser found no product IDs in region page", + ) return True, f"成功取得 {len(product_ids)} 個商品", product_ids except requests.RequestException as e: logger.error(f"爬取 {url} 失敗: {e}") + self._emit_visual_evidence_action( + url=url, + crawler="PChomeCrawler.fetch_region_page", + failure_reason=f"request_error: {e}", + ) return False, f"請求失敗: {str(e)}", [] def fetch_product_details(self, product_ids: List[str], batch_size: int = 20) -> Tuple[bool, str, List[PChomeProduct]]: @@ -327,6 +392,14 @@ class PChomeCrawler: message = f"成功取得 {len(all_products)} 個商品資料" if failed_count > 0: message += f"，{failed_count} 個失敗" + if product_ids and not all_products: + first_id = self._normalize_product_id(str(product_ids[0])) + self._emit_visual_evidence_action( + url=f"{self.BASE_URL}/prod/{first_id}", + crawler="PChomeCrawler.fetch_product_details", + failure_reason=message, + missing_fields=("title", "price", "product_id", "image_url"), + ) return len(all_products) > 0, message, all_products @@ -487,6 +560,11 @@ class PChomeCrawler: break if not product_ids: + self._emit_visual_evidence_action( + url=self._public_search_url(keyword), + crawler="PChomeCrawler.search_products", + failure_reason="search API returned no product IDs", + ) return False, "沒有找到符合的商品", [] # 取得詳細資料 @@ -497,6 +575,11 @@ class PChomeCrawler: except requests.RequestException as e: logger.error(f"搜尋失敗: {e}") + self._emit_visual_evidence_action( + url=self._public_search_url(keyword), + crawler="PChomeCrawler.search_products", + failure_reason=f"request_error: {e}", + ) return False, f"搜尋失敗: {str(e)}", [] diff --git a/services/pixelrag_crawler_integration_service.py b/services/pixelrag_crawler_integration_service.py index 647e284..4f254cd 100644 --- a/services/pixelrag_crawler_integration_service.py +++ b/services/pixelrag_crawler_integration_service.py @@ -15,20 +15,116 @@ from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from typing import Any, Mapping -from urllib.parse import urlparse +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" -ALLOWED_PLATFORMS = ("momo", "pchome", "market_intel", "external_market") 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(): @@ -36,7 +132,18 @@ def _default_artifact_root() -> str: 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", @@ -188,6 +295,38 @@ def _safe_int(value: Any, default: int = 0) -> int: 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"]) @@ -316,6 +455,131 @@ def should_emit_visual_evidence_fallback( } +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, @@ -425,6 +689,144 @@ def build_pixelrag_visual_evidence_manifest( } +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, @@ -634,7 +1036,19 @@ def build_pixelrag_visual_evidence_readback( 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" + 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" @@ -651,9 +1065,11 @@ def build_pixelrag_visual_evidence_readback( "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")), + "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": { @@ -661,14 +1077,18 @@ def build_pixelrag_visual_evidence_readback( "generated_at": receipt.get("generated_at"), "capture_status": receipt.get("status"), "capture_target": capture_target, - "page_metrics": receipt.get("page_metrics") or {}, + "page_metrics": page_metrics, "tile_plan": tile_plan, "files": file_statuses, }, "next_machine_action": ( "keep_pixelrag_visual_evidence_receipt" if status == "ok" - else "run_pixelrag_visual_capture_worker" + else ( + "run_platform_probe_or_use_structured_api" + if barrier_reason + else "run_pixelrag_visual_capture_worker" + ) ), "controlled_apply": { "network_call": False, @@ -680,6 +1100,9 @@ def build_pixelrag_visual_evidence_readback( __all__ = [ + "CRAWLER_FAILURE_ACTION_POLICY", + "ECOMMERCE_EXPANSION_POLICY", + "ECOMMERCE_PLATFORM_PROFILES", "MANIFEST_POLICY", "POLICY", "READBACK_POLICY", @@ -688,7 +1111,11 @@ __all__ = [ "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", ] diff --git a/tests/test_pchome_crawler_search.py b/tests/test_pchome_crawler_search.py index 6b859b6..6596d74 100644 --- a/tests/test_pchome_crawler_search.py +++ b/tests/test_pchome_crawler_search.py @@ -69,6 +69,32 @@ def test_pchome_search_scans_multiple_pages_until_limit(monkeypatch): assert [product.product_id for product in products] == ["A001", "A002", "A003"] +def test_pchome_search_auto_queues_pixelrag_manifest_on_empty_result(monkeypatch, tmp_path): + from services import pixelrag_crawler_integration_service as pixelrag + from services.pchome_crawler import PChomeCrawler + + monkeypatch.setattr(pixelrag, "DEFAULT_MANIFEST_QUEUE_ROOT", str(tmp_path)) + crawler = PChomeCrawler(timeout=1, delay=0, max_retries=0) + + class FakeSession: + headers = {} + + def get(self, url, params=None, timeout=None): + return _FakeResponse({"Prods": []}) + + crawler.session = FakeSession() + + success, message, products = crawler.search_products("找不到的商品", limit=3, max_pages=1) + + assert success is False + assert message == "沒有找到符合的商品" + assert products == [] + action = crawler.last_visual_evidence_action + assert action["status"] == "manifest_persisted" + assert action["manifest"]["capture_target"]["crawler"] == "PChomeCrawler.search_products" + assert action["persistence"]["path"].startswith(str(tmp_path)) + + def test_pchome_get_retries_transient_timeout(): from services.pchome_crawler import PChomeCrawler diff --git a/tests/test_pixelrag_crawler_integration_service.py b/tests/test_pixelrag_crawler_integration_service.py index f2a386c..263828d 100644 --- a/tests/test_pixelrag_crawler_integration_service.py +++ b/tests/test_pixelrag_crawler_integration_service.py @@ -124,6 +124,67 @@ def test_visual_fallback_selector_routes_parser_empty_and_low_confidence_cases() assert clean_parser["fallback_reason"] == "structured_evidence_sufficient" +def test_pixelrag_ecommerce_expansion_plan_includes_shopee_and_coupang(): + from services.pixelrag_crawler_integration_service import ( + ECOMMERCE_EXPANSION_POLICY, + build_pixelrag_ecommerce_expansion_plan, + ) + + payload = build_pixelrag_ecommerce_expansion_plan() + platforms = {item["platform"]: item for item in payload["platforms"]} + + assert payload["policy"] == ECOMMERCE_EXPANSION_POLICY + assert payload["status"] == "marketplace_profiles_ready" + assert platforms["shopee_tw"]["manifest_ready"] is True + assert platforms["coupang_tw"]["manifest_ready"] is True + assert "price" in platforms["shopee_tw"]["data_targets"] + assert payload["next_actions"][0]["target_platforms"] == ["shopee_tw", "coupang_tw"] + assert payload["controlled_apply"]["production_price_write"] is False + + +def test_marketplace_search_manifest_builds_shopee_visual_capture_target(): + from services.pixelrag_crawler_integration_service import build_pixelrag_marketplace_search_manifest + + payload = build_pixelrag_marketplace_search_manifest( + platform="shopee_tw", + keyword="防曬乳", + page_size={"width": 1440, "height": 1024}, + tile_size={"width": 512, "height": 512}, + ) + + assert payload["success"] is True + assert payload["capture_target"]["platform"] == "shopee_tw" + assert payload["capture_target"]["url"].startswith("https://shopee.tw/search?keyword=") + assert "%E9%98%B2%E6%9B%AC%E4%B9%B3" in payload["capture_target"]["url"] + assert payload["marketplace_profile"]["display_name"] == "Shopee Taiwan" + assert payload["tile_plan"]["tile_count"] == 6 + + +def test_crawler_failure_action_persists_manifest_queue(tmp_path): + from services.pixelrag_crawler_integration_service import build_pixelrag_crawler_failure_action + + payload = build_pixelrag_crawler_failure_action( + url="https://24h.pchome.com.tw/search/?q=test", + platform="pchome", + crawler="PChomeCrawler.search_products", + parser_success=False, + parsed_item_count=0, + missing_fields=["price", "product_id"], + failure_reason="search API returned no product IDs", + persist_manifest=True, + queue_root=tmp_path, + ) + + assert payload["success"] is True + assert payload["status"] == "manifest_persisted" + assert payload["next_machine_action"] == "run_pixelrag_visual_capture_worker" + assert payload["controlled_apply"]["artifact_write"] is True + manifest_path = payload["persistence"]["path"] + queued = json.loads(open(manifest_path, encoding="utf-8").read()) + assert queued["capture_target"]["platform"] == "pchome" + assert queued["queue_policy"] == "read_only_pixelrag_crawler_failure_action_v1" + + def test_visual_evidence_manifest_builds_tile_plan_without_runtime_writes(): from services.pixelrag_crawler_integration_service import ( RESULT_MANIFEST_READY, @@ -226,6 +287,17 @@ def test_pixelrag_ai_automation_route_returns_assessment_and_manifest(): ): manifest_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() manifest = manifest_response.get_json() + with app.test_request_context( + "/api/ai-automation/pixelrag-crawler-integration?expansion_plan=true" + ): + expansion_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() + expansion = expansion_response.get_json() + with app.test_request_context( + "/api/ai-automation/pixelrag-crawler-integration" + "?platform=shopee_tw&marketplace_keyword=防曬乳" + ): + marketplace_response = routes.ai_automation_pixelrag_crawler_integration_api.__wrapped__() + marketplace = marketplace_response.get_json() assert assessment["policy"] == "read_only_pixelrag_crawler_integration_assessment_v1" assert assessment["target_platforms"] == ["momo"] @@ -233,6 +305,10 @@ def test_pixelrag_ai_automation_route_returns_assessment_and_manifest(): assert manifest["policy"] == "read_only_pixelrag_visual_evidence_manifest_v1" assert manifest["result"] == "PIXELRAG_VISUAL_EVIDENCE_MANIFEST_READY" assert manifest["capture_target"]["platform"] == "pchome" + assert expansion["policy"] == "read_only_pixelrag_ecommerce_expansion_plan_v1" + assert expansion["next_actions"][0]["target_platforms"] == ["shopee_tw", "coupang_tw"] + assert marketplace["capture_target"]["platform"] == "shopee_tw" + assert marketplace["marketplace_profile"]["keyword"] == "防曬乳" def _write_capture_receipt(root, *, platform="pchome", manifest_id="manifest-1"): @@ -303,6 +379,30 @@ def test_pixelrag_visual_evidence_readback_reports_latest_receipt(tmp_path): assert payload["controlled_apply"]["db_write"] is False +def test_pixelrag_visual_evidence_readback_warns_on_access_denied_page(tmp_path): + from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback + + receipt_path = _write_capture_receipt(tmp_path, platform="coupang_tw", manifest_id="denied") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + receipt["http_status"] = 403 + receipt["page_metrics"] = { + "final_url": "https://www.tw.coupang.com/search?q=iphone", + "title": "Access Denied", + } + receipt_path.write_text(json.dumps(receipt), encoding="utf-8") + + payload = build_pixelrag_visual_evidence_readback( + artifact_root=tmp_path, + platform="coupang_tw", + manifest_id="denied", + ) + + assert payload["status"] == "warning" + assert payload["summary"]["visual_barrier_detected"] is True + assert payload["summary"]["visual_barrier_reason"] == "http_status_403" + assert payload["next_machine_action"] == "run_platform_probe_or_use_structured_api" + + def test_pixelrag_visual_evidence_readback_warns_when_missing(tmp_path): from services.pixelrag_crawler_integration_service import build_pixelrag_visual_evidence_readback @@ -467,3 +567,34 @@ def test_pixelrag_python_capture_worker_dry_run_plans_artifact_outputs(tmp_path) assert payload["controlled_apply"]["artifact_write"] is False assert payload["tile_plan"]["planned_tile_count"] == 4 assert payload["capture_target"]["platform"] == "pchome" + + +def test_pixelrag_python_capture_worker_accepts_shopee_manifest(tmp_path): + from services.pixelrag_crawler_integration_service import build_pixelrag_marketplace_search_manifest + + manifest = build_pixelrag_marketplace_search_manifest( + platform="shopee_tw", + keyword="防曬乳", + page_size={"width": 1024, "height": 1024}, + tile_size={"width": 512, "height": 512}, + ) + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/capture_pixelrag_visual_evidence.py", + "--manifest-json", + json.dumps(manifest), + "--output-dir", + str(tmp_path), + "--dry-run", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["status"] == "dry_run_ready" + assert payload["capture_target"]["platform"] == "shopee_tw" + assert payload["tile_plan"]["planned_tile_count"] == 4