feat(crawler): expand pixelrag marketplace automation

.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/

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
 ```

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]

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(),
 };

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(),
 }

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(

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)}", []
 
 

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

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
 

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