From 199c857353fd859aebe4338380381a8a0c507343 Mon Sep 17 00:00:00 2001 From: ogt Date: Fri, 10 Jul 2026 01:21:39 +0800 Subject: [PATCH] feat(ai): add PixelRAG platform probe readiness --- config.py | 2 +- docs/AI_INTELLIGENCE_MODULE_SOT.md | 2 + routes/system_public_routes.py | 22 + .../ops/capture_pixelrag_visual_evidence.js | 39 +- .../ops/capture_pixelrag_visual_evidence.py | 46 +- scripts/ops/report_pixelrag_platform_probe.py | 65 ++ services/ai_automation_smoke_service.py | 102 ++++ services/pixelrag_platform_probe_service.py | 565 ++++++++++++++++++ tests/test_ai_automation_smoke_service.py | 40 +- tests/test_pixelrag_platform_probe_service.py | 240 ++++++++ 10 files changed, 1113 insertions(+), 10 deletions(-) create mode 100644 scripts/ops/report_pixelrag_platform_probe.py create mode 100644 services/pixelrag_platform_probe_service.py create mode 100644 tests/test_pixelrag_platform_probe_service.py diff --git a/config.py b/config.py index e474d56..7e287c8 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.758" +SYSTEM_VERSION = "V10.759" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index 22f0c97..d239c2c 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -118,6 +118,7 @@ - 2026-07-09 起 PixelRAG 可整合/可運用範圍必須有機器可讀 application portfolio：`/api/ai-automation/pixelrag-application-portfolio` 與 `scripts/ops/report_pixelrag_application_portfolio.py` 需輸出 commerce、RAG、UX、ops、marketing、governance lanes，每條 lane 必須包含 priority、status、integrates_with、use_cases、current_capability、next_machine_action、no-write 邊界與 forbidden guardrails。此 portfolio 不抓外站、不呼叫模型、不讀 secret、不寫 DB；其用途是把「還可以整合哪些」變成可排程、可驗證、可拒絕違規場景的主線工作項目。 - 2026-07-09 起 PixelRAG ready receipts 必須有 Ollama-first VLM replay worker：`/api/ai-automation/pixelrag-vlm-replay-worker` 與 `scripts/ops/run_pixelrag_vlm_replay_worker.py` 預設 dry-run，`execute=true&write_receipt=true` 才呼叫 approved Ollama VLM route 並寫 artifact receipt；`/api/ai-automation/smoke` 需包含 `PixelRAG VLM replay worker`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_vlm_replay_worker` family。此 worker 不讀 secret、不寫 DB、不寫 `ai_insights`、不寫正式價格表；blocked / 403 / captcha / access denied receipt 自動跳 platform probe 或 structured API，ready receipt 的 VLM 結果仍須 identity matcher replay 與 PromotionGate 才能進候選知識層。 - 2026-07-09 起 PixelRAG VLM route readiness 必須在 execute 前可讀回：`/api/ai-automation/pixelrag-vlm-route-readiness` 與 `scripts/ops/report_pixelrag_vlm_route_readiness.py` 只讀 approved Ollama `/api/tags`，輸出 configured model、candidate model、reachable host、model_route_ready 與 next machine action；`/api/ai-automation/smoke` 需包含 `PixelRAG VLM route readiness`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_vlm_route_readiness` family。此 readback 不呼叫 `/api/generate`、不讀 secret、不寫 DB；worker execute 必須使用它自動避開缺模型路由，完全缺候選時寫 `model_route_not_ready` artifact receipt。 +- 2026-07-10 起 PixelRAG platform probe 必須把 blocked / interstitial receipt 轉成機器可執行下一步：`/api/ai-automation/pixelrag-platform-probe` 與 `scripts/ops/report_pixelrag_platform_probe.py` 需讀取 capture receipts 與 VLM replay receipts，辨識 Shopee language / region / generic marketplace slogan / traffic verification 與 Coupang 403 / access denied，輸出 public browser context probe、structured source fallback、platform backoff、blocked page not product data 與 no-write 邊界；`/api/ai-automation/smoke` 需包含 `PixelRAG platform probe readiness`，`/api/ai-automation/scheduled-health-summary` 需輸出 `pixelrag_platform_probe` family。此 readback 不抓外站、不讀 secret/cookie/session、不寫 DB、不寫 `ai_insights`、不寫正式價格表；capture worker 可消化 `public_browser_context` 的 locale/timezone/Accept-Language，但必須濾掉 Cookie、Authorization、token/secret/key 類 header。 - 2026-07-02 起 `/ai_intelligence` 商品明細與單品作戰詳情的四格價格證據必須可測：PChome 價格、MOMO 參考價、差距、可信度需以 `data-evidence` 固定，並以 `aria-label="價格證據"` 對應可掃描區塊；候選待確認或缺資料只能顯示「候選待確認 / 待補」，不得捏造價格或讓使用者打開 raw payload 才知道判斷依據。 - 2026-07-02 起 `/ai_intelligence` 必須是密集 AI 工作台，不得退回大段文字說明頁：首屏與明細可見內容只保留短狀態、數字、四格證據與下一步按鈕；KPI note、benchmark detail、alert 副句、策略說明、decision copy、來源長句與單品 reason list 不得佔用第一層視覺。`tests/test_ai_intelligence_text_density_guardrails.py` 必須鎖住 `data-density-guardrail="compact-ai-workbench"`、短任務文案、detail meta 與 hidden explanatory copy。 - 2026-07-02 起 `/observability/overview` 也必須採密集 AI 觀測工作台：首屏以 `data-density-guardrail="compact-observability-workbench"`、`AI 觀測 / 風險優先 / 下一步` 與 golden signals 先呈現狀態、數字與操作入口；hero lede、signal note、route desc、host meta 與資料來源長句不得佔用第一層視覺。`tests/test_observability_text_density_guardrails.py` 必須鎖住 compact marker 與 hidden explanatory copy。 @@ -888,6 +889,7 @@ POSTGRES_HOST=momo-db | 2026-07-10 | PixelRAG VLM 空欄位與 Shopee language interstitial 必須自動轉 platform probe | V10.756 起 VLM prompt 明確把 language / region / app-download / landing / loading / logo-only / cookie consent pages 視為非商品卡；若 VLM 回傳所有 offer fields 皆空且 required title / price 缺失，worker 會標記 `non_product_or_interstitial_detected=true`，receipt 下一步改為 `run_platform_probe_or_use_structured_api`，不得反覆要求人工審核或無限 rerun VLM。 | | 2026-07-10 | PixelRAG VLM worker preflight 必須容納 111 VLM 冷啟延遲 | V10.757 起 `run_pixelrag_vlm_replay_worker.py --execute` 與對應 API 的 `route_generate_probe_timeout` 預設為 20 秒；111 `minicpm-v` 在正式環境可於約 4 到 20 秒間完成小型 `/api/generate` probe，8 秒容易造成假陰性。若 VLM receipt 已判定 `run_platform_probe_or_use_structured_api`，run-level `next_machine_action` 也必須同樣輸出 platform probe，不得停留在 rerun VLM 摘要。 | | 2026-07-10 | PixelRAG VLM language note / generic marketplace slogan 必須視為 interstitial | V10.758 起若 VLM notes 或 title 顯示 `Language selection page`、語言/地區選擇，或只抽到 Shopee `蝦皮購物 | 花得更少買得更好` 這類網站 slogan 且 price/shop 缺失，worker validation 會標記 `interstitial_signal_detected` / `generic_marketplace_title_detected` 與 `non_product_or_interstitial_detected=true`，run-level 與 item-level 下一步都輸出 `run_platform_probe_or_use_structured_api`。 | +| 2026-07-10 | PixelRAG platform probe 必須自動消化 marketplace interstitial / blocked receipt | V10.759 起 `/api/ai-automation/pixelrag-platform-probe`、`scripts/ops/report_pixelrag_platform_probe.py`、`/api/ai-automation/smoke` 與 `/api/ai-automation/scheduled-health-summary` 必須輸出 platform probe / `pixelrag_platform_probe` family；Shopee traffic / language / generic landing 會轉成 `run_shopee_public_context_probe_then_structured_source_fallback`，Coupang 403 / access denied 會轉成 `use_structured_source_or_platform_backoff_policy`。probe plan 只產生 public empty browser context、structured adapter fallback 與 no-write contract，不讀或注入 cookie/session/login，不寫 DB、不寫 `ai_insights`、不寫正式價格表；capture worker 只允許 locale/timezone/非敏感 header 並濾掉 Cookie / Authorization / token / secret / key 類 header。 | | 2026-06-29 | PChome DB apply 授權 lane 必須先通過 no-write guard / decision preflight / decision closeout / issuer gate / signing-decision preflight / signing-decision closeout / signing-issuer guard | V10.725 的 PChome mapping backlog auto-policy 已新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-lane-guard`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-decision-preflight`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-decision-closeout`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-issuer-gate`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-decision-preflight`、`/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-decision-closeout` 與 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-issuer-guard`；這些 endpoint 只驗證 final exact request package、same-run production truth requirement、secret rejection、rollback boundary、lane entry requirements、decision input requirements、rejection policy、post-apply verifier、future authorization decision package、final nonsecret authorization envelope、signing decision preflight inputs、unsigned signing decision package 與 signable request boundary，不讀 secret、不執行 shell/SQL、不寫 DB，也不簽發 database apply authorization。 | | 2026-06-29 | PChome DB apply 授權簽署發行者 lane 必須先產出 final signable request package | V10.725 的 PChome mapping backlog auto-policy 新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-issuer-closeout`；此 endpoint 只把 signing-issuer guard 的 signable request boundary 收斂成 final signable request package 與 closeout contract，確認 fresh production truth、post-apply verifier、migration hash、secret boundary 與 no-side-effect checks，不讀 secret、不簽發 authorization、不執行 shell/SQL、不寫 DB，也不代表正式 DB apply 已授權。 | | 2026-06-29 | PChome DB apply 授權簽署執行 lane 必須先通過 operator-held secret boundary preflight | V10.725 的 PChome mapping backlog auto-policy 新增 `/api/ai/pchome-growth/mapping-backlog/auto-policy-db-apply-authorization-signing-execution-preflight`；此 endpoint 只把 final signable request package 轉成 future signing execution preflight package、operator-held secret boundary contract、nonsecret signing inputs、command-shape preview、rollback boundary 與 abort conditions，不讀 secret、不接受 plaintext secret、不簽發 authorization、不執行 shell/SQL、不寫 DB，也不代表正式 DB apply 已授權。 | diff --git a/routes/system_public_routes.py b/routes/system_public_routes.py index e5d0177..86b3259 100644 --- a/routes/system_public_routes.py +++ b/routes/system_public_routes.py @@ -840,6 +840,28 @@ def ai_automation_pixelrag_vlm_route_readiness_api(): )) +@system_public_bp.route('/api/ai-automation/pixelrag-platform-probe') +@login_required +def ai_automation_pixelrag_platform_probe_api(): + """Read-only PixelRAG platform probe readiness from barrier receipts.""" + from services.pixelrag_platform_probe_service import ( + build_pixelrag_platform_probe_readiness, + ) + + platforms = tuple( + str(item or '').strip() + for item in request.args.getlist('platform') + if str(item or '').strip() + ) + max_age_hours = request.args.get('max_age_hours', 168, type=int) + limit = request.args.get('limit', 50, type=int) + return jsonify(build_pixelrag_platform_probe_readiness( + platform=platforms, + max_age_hours=max(1, min(max_age_hours or 168, 720)), + limit=max(1, min(limit or 50, 250)), + )) + + @system_public_bp.route('/api/ai-automation/external-mcp-rag-integration') @login_required def ai_automation_external_mcp_rag_integration_api(): diff --git a/scripts/ops/capture_pixelrag_visual_evidence.js b/scripts/ops/capture_pixelrag_visual_evidence.js index beea5ae..d7953e7 100644 --- a/scripts/ops/capture_pixelrag_visual_evidence.js +++ b/scripts/ops/capture_pixelrag_visual_evidence.js @@ -17,6 +17,7 @@ const DEFAULT_OUTPUT_DIR = 'runtime_artifacts/pixelrag_visual_evidence'; const DEFAULT_TIMEOUT_MS = 30000; const DEFAULT_SETTLE_MS = 500; const DEFAULT_MAX_TILES = 80; +const FORBIDDEN_CONTEXT_HEADER_TOKENS = ['cookie', 'authorization', 'token', 'secret', 'key']; const ALLOWED_PLATFORM_HOSTS = { momo: new Set(['m.momoshop.com.tw', 'www.momoshop.com.tw']), @@ -184,6 +185,30 @@ function positiveInt(value, fallback) { return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; } +function safePublicBrowserContext(manifest) { + const input = manifest.public_browser_context || {}; + const headers = {}; + const rawHeaders = input.extra_http_headers || {}; + for (const [key, value] of Object.entries(rawHeaders)) { + const cleanKey = String(key || '').trim(); + const lowerKey = cleanKey.toLowerCase(); + if (!cleanKey || FORBIDDEN_CONTEXT_HEADER_TOKENS.some((token) => lowerKey.includes(token))) { + continue; + } + headers[cleanKey] = String(value || ''); + } + return { + context_policy: 'public_empty_browser_context_no_login', + locale: String(input.locale || 'zh-TW'), + timezone_id: String(input.timezone_id || input.timezoneId || 'Asia/Taipei'), + extra_http_headers: headers, + credentialed_session_allowed: false, + storage_state_allowed: false, + raw_cookie_or_session_read_allowed: false, + login_allowed: false, + }; +} + function buildTilePlan({ width, height, tileWidth, tileHeight, maxTiles }) { const tiles = []; const tilesX = Math.max(1, Math.ceil(width / tileWidth)); @@ -273,6 +298,7 @@ function buildBaseArtifact(manifest, options, errors = []) { tile_size: { width: tileWidth, height: tileHeight }, tile_plan: tilePlan, files: [], + public_browser_context: safePublicBrowserContext(manifest), }; } @@ -288,8 +314,19 @@ async function capture(manifest, options, artifact) { launchOptions.executablePath = chromePath; } + const publicContext = safePublicBrowserContext(manifest); + const contextOptions = { + ignoreHTTPSErrors: true, + viewport, + locale: publicContext.locale, + timezoneId: publicContext.timezone_id, + }; + if (Object.keys(publicContext.extra_http_headers).length) { + contextOptions.extraHTTPHeaders = publicContext.extra_http_headers; + } + const browser = await chromium.launch(launchOptions); - const context = await browser.newContext({ ignoreHTTPSErrors: true, viewport }); + const context = await browser.newContext(contextOptions); const page = await context.newPage(); try { const response = await page.goto(manifest.capture_target.url, { diff --git a/scripts/ops/capture_pixelrag_visual_evidence.py b/scripts/ops/capture_pixelrag_visual_evidence.py index f9c2a1f..6ffba51 100644 --- a/scripts/ops/capture_pixelrag_visual_evidence.py +++ b/scripts/ops/capture_pixelrag_visual_evidence.py @@ -23,6 +23,7 @@ DEFAULT_OUTPUT_DIR = "runtime_artifacts/pixelrag_visual_evidence" DEFAULT_TIMEOUT_MS = 30000 DEFAULT_SETTLE_MS = 500 DEFAULT_MAX_TILES = 80 +FORBIDDEN_CONTEXT_HEADER_TOKENS = ("cookie", "authorization", "token", "secret", "key") 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"}, @@ -123,6 +124,33 @@ def _build_tile_plan( } +def _safe_public_browser_context(manifest: dict[str, Any]) -> dict[str, Any]: + raw_context = manifest.get("public_browser_context") or {} + raw_headers = raw_context.get("extra_http_headers") or {} + headers: dict[str, str] = {} + if isinstance(raw_headers, dict): + for key, value in raw_headers.items(): + clean_key = str(key or "").strip() + lower_key = clean_key.lower() + if not clean_key: + continue + if any(token in lower_key for token in FORBIDDEN_CONTEXT_HEADER_TOKENS): + continue + headers[clean_key] = str(value or "") + return { + "context_policy": "public_empty_browser_context_no_login", + "locale": str(raw_context.get("locale") or "zh-TW"), + "timezone_id": str( + raw_context.get("timezone_id") or raw_context.get("timezoneId") or "Asia/Taipei" + ), + "extra_http_headers": headers, + "credentialed_session_allowed": False, + "storage_state_allowed": False, + "raw_cookie_or_session_read_allowed": False, + "login_allowed": False, + } + + def _output_paths(manifest: dict[str, Any], output_dir: str) -> dict[str, Path]: target = manifest.get("capture_target") or {} manifest_id = manifest.get("manifest_id") or _safe_name( @@ -187,6 +215,7 @@ def _base_artifact( max_tiles=_positive_int(args.max_tiles, DEFAULT_MAX_TILES), ), "files": [], + "public_browser_context": _safe_public_browser_context(manifest), } @@ -205,10 +234,19 @@ def _capture( with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) - context = browser.new_context(ignore_https_errors=True, viewport={ - "width": viewport["width"], - "height": viewport["height"], - }) + public_context = _safe_public_browser_context(manifest) + context_kwargs = { + "ignore_https_errors": True, + "viewport": { + "width": viewport["width"], + "height": viewport["height"], + }, + "locale": public_context["locale"], + "timezone_id": public_context["timezone_id"], + } + if public_context["extra_http_headers"]: + context_kwargs["extra_http_headers"] = public_context["extra_http_headers"] + context = browser.new_context(**context_kwargs) page = context.new_page() try: response = page.goto(target["url"], wait_until=args.wait_until, timeout=timeout_ms) diff --git a/scripts/ops/report_pixelrag_platform_probe.py b/scripts/ops/report_pixelrag_platform_probe.py new file mode 100644 index 0000000..95e59d4 --- /dev/null +++ b/scripts/ops/report_pixelrag_platform_probe.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +"""Report PixelRAG platform probe readiness from barrier receipts.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from services.pixelrag_platform_probe_service import ( # noqa: E402 + build_pixelrag_platform_probe_readiness, +) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="輸出 PixelRAG platform probe readiness 的機器可讀讀回。" + ) + parser.add_argument( + "--artifact-root", + help="PixelRAG visual evidence artifact root；預設使用 production/container 設定。", + ) + parser.add_argument( + "--vlm-receipt-root", + help="PixelRAG VLM replay receipt root；預設使用 production/container 設定。", + ) + parser.add_argument( + "--platform", + action="append", + dest="platforms", + help="限制平台，可重複指定，例如 --platform shopee_tw --platform coupang_tw。", + ) + parser.add_argument( + "--max-age-hours", + type=int, + default=168, + help="receipt 最大新鮮度小時數。", + ) + parser.add_argument( + "--limit", + type=int, + default=50, + help="最多輸出 probe candidate 數。", + ) + args = parser.parse_args() + + payload = build_pixelrag_platform_probe_readiness( + artifact_root=args.artifact_root, + vlm_receipt_root=args.vlm_receipt_root, + platform=tuple(args.platforms or ()), + max_age_hours=args.max_age_hours, + limit=args.limit, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/ai_automation_smoke_service.py b/services/ai_automation_smoke_service.py index a8a58d8..b0ee851 100644 --- a/services/ai_automation_smoke_service.py +++ b/services/ai_automation_smoke_service.py @@ -552,6 +552,11 @@ def build_scheduled_automation_health_summary( if not pixelrag_vlm_replay_worker or not pixelrag_vlm_replay_worker_details: pixelrag_vlm_replay_worker = _pixelrag_vlm_replay_worker_check() pixelrag_vlm_replay_worker_details = pixelrag_vlm_replay_worker.get("details") or {} + pixelrag_platform_probe = _find_check(source_result, "PixelRAG platform probe readiness") + pixelrag_platform_probe_details = pixelrag_platform_probe.get("details") or {} + if not pixelrag_platform_probe or not pixelrag_platform_probe_details: + pixelrag_platform_probe = _pixelrag_platform_probe_check() + pixelrag_platform_probe_details = pixelrag_platform_probe.get("details") or {} smoke_status = source_result.get("status") or ("warning" if latest_history else "warning") freshness_family = _history_freshness_family( latest_history, @@ -4732,6 +4737,44 @@ def build_scheduled_automation_health_summary( "primary_human_gate_count": 0, }, }, + { + "key": "pixelrag_platform_probe", + "label": "PixelRAG platform probe readiness", + "status": pixelrag_platform_probe.get("status") or "warning", + "summary": ( + pixelrag_platform_probe.get("summary") + or "PixelRAG platform probe readiness has no latest readback." + ), + "next_machine_action": pixelrag_platform_probe_details.get("next_machine_action") + or "run_pixelrag_platform_probe_readback", + "details": { + "policy": pixelrag_platform_probe_details.get("policy"), + "probe_candidate_count": int( + pixelrag_platform_probe_details.get("probe_candidate_count") or 0 + ), + "ready_for_probe_count": int( + pixelrag_platform_probe_details.get("ready_for_probe_count") or 0 + ), + "shopee_public_context_probe_count": int( + pixelrag_platform_probe_details.get("shopee_public_context_probe_count") or 0 + ), + "language_or_region_interstitial_count": int( + pixelrag_platform_probe_details.get("language_or_region_interstitial_count") or 0 + ), + "traffic_verification_count": int( + pixelrag_platform_probe_details.get("traffic_verification_count") or 0 + ), + "access_denied_count": int( + pixelrag_platform_probe_details.get("access_denied_count") or 0 + ), + "structured_source_fallback_count": int( + pixelrag_platform_probe_details.get("structured_source_fallback_count") or 0 + ), + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + }, freshness_family, { "key": "daily_summary_delivery", @@ -13573,6 +13616,64 @@ def _pixelrag_vlm_route_readiness_check() -> Dict[str, Any]: ) +def _pixelrag_platform_probe_check() -> Dict[str, Any]: + """Read-only sentinel for PixelRAG platform probe automation readiness.""" + try: + from services.pixelrag_platform_probe_service import ( + build_pixelrag_platform_probe_readiness, + ) + + readback = build_pixelrag_platform_probe_readiness() + summary = readback.get("summary") or {} + candidate_count = int(summary.get("probe_candidate_count") or 0) + ready_count = int(summary.get("ready_for_probe_count") or 0) + shopee_context_count = int(summary.get("shopee_public_context_probe_count") or 0) + structured_count = int(summary.get("structured_source_fallback_count") or 0) + status = readback.get("status") or "warning" + summary_text = ( + f"PixelRAG platform probe candidates={candidate_count}, " + f"ready={ready_count}, shopee_context={shopee_context_count}, " + f"structured_fallback={structured_count}" + ) + return _check( + "PixelRAG platform probe readiness", + status, + summary_text, + { + "policy": readback.get("policy"), + "probe_candidate_count": candidate_count, + "ready_for_probe_count": ready_count, + "invalid_count": int(summary.get("invalid_count") or 0), + "capture_source_count": int(summary.get("capture_source_count") or 0), + "vlm_source_count": int(summary.get("vlm_source_count") or 0), + "shopee_public_context_probe_count": shopee_context_count, + "language_or_region_interstitial_count": int( + summary.get("language_or_region_interstitial_count") or 0 + ), + "traffic_verification_count": int( + summary.get("traffic_verification_count") or 0 + ), + "access_denied_count": int(summary.get("access_denied_count") or 0), + "structured_source_fallback_count": structured_count, + "next_machine_action": readback.get("next_machine_action"), + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + except Exception as exc: + return _check( + "PixelRAG platform probe readiness", + "critical", + f"PixelRAG platform probe readiness 無法執行：{exc}", + { + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + + def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: int = 20) -> Dict[str, Any]: checks: List[Dict[str, Any]] = [ _event_router_check(), @@ -13611,6 +13712,7 @@ def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: i _pixelrag_ocr_vlm_replay_check(), _pixelrag_vlm_route_readiness_check(), _pixelrag_vlm_replay_worker_check(), + _pixelrag_platform_probe_check(), ] worst = max(checks, key=lambda item: STATUS_RANK.get(item["status"], 2))["status"] result = { diff --git a/services/pixelrag_platform_probe_service.py b/services/pixelrag_platform_probe_service.py new file mode 100644 index 0000000..772c2b6 --- /dev/null +++ b/services/pixelrag_platform_probe_service.py @@ -0,0 +1,565 @@ +"""Read-only PixelRAG platform probe readiness. + +This module turns PixelRAG capture/VLM barrier evidence into concrete machine +actions for public marketplace probing. It does not read sessions/cookies, +perform network calls, write DB rows, or promote price truth. +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping +from urllib.parse import parse_qs, unquote_plus, urlparse + +from services.market_intel.adapters.registry import get_adapter +from services.pixelrag_crawler_integration_service import ( + DEFAULT_ARTIFACT_MAX_AGE_HOURS, + DEFAULT_ARTIFACT_ROOT, + build_pixelrag_marketplace_search_manifest, + _parse_iso_datetime, +) +from services.pixelrag_rag_candidate_replay_service import ( + build_pixelrag_rag_candidate_replay_readback, +) + + +POLICY = "read_only_pixelrag_platform_probe_readiness_v1" +DEFAULT_LIMIT = 50 +DEFAULT_ACCEPT_LANGUAGE = "zh-TW,zh-Hant;q=0.95,zh;q=0.9,en;q=0.7" +DEFAULT_VLM_RECEIPT_ROOT = os.getenv( + "PIXELRAG_VLM_REPLAY_RECEIPT_ROOT", + "/app/data/ai_automation/pixelrag_vlm_replay_receipts" + if Path("/app/data").exists() + else "runtime_artifacts/pixelrag_vlm_replay_receipts", +) +PLATFORM_ADAPTER_ALIASES = { + "shopee_tw": "shopee", + "coupang_tw": "coupang", + "yahoo_shopping_tw": "yahoo", +} + + +def _normalise_platforms( + platform: str | tuple[str, ...] | list[str] | None, +) -> tuple[str, ...]: + if isinstance(platform, str): + value = platform.strip().lower() + return (value,) if value else () + return tuple( + str(item or "").strip().lower() + for item in (platform or ()) + if str(item or "").strip() + ) + + +def _read_json(path: Path) -> tuple[dict[str, Any], str]: + try: + parsed = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return {}, str(exc)[:300] + return parsed if isinstance(parsed, dict) else {}, "" + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value or default) + except (TypeError, ValueError): + return default + + +def _safe_text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, Mapping): + return " ".join(_safe_text(item) for item in value.values()) + if isinstance(value, list): + return " ".join(_safe_text(item) for item in value) + return str(value) + + +def _extract_keyword(url: str) -> str: + parsed = urlparse(str(url or "")) + query = parse_qs(parsed.query) + for key in ("keyword", "q", "query", "p", "search", "searchKeyword"): + values = query.get(key) or [] + for value in values: + text = unquote_plus(str(value or "")).strip() + if text: + return text + return "" + + +def _adapter_code(platform: str) -> str: + clean = str(platform or "").strip().lower() + return PLATFORM_ADAPTER_ALIASES.get(clean, clean) + + +def _structured_source_plan(platform: str) -> dict[str, Any]: + adapter = get_adapter(_adapter_code(platform)) + if not adapter: + return { + "available": False, + "adapter_code": _adapter_code(platform), + "network_request_allowed": False, + "database_write_allowed": False, + "sources": [], + } + plan = adapter.build_discovery_plan() + return { + "available": True, + "adapter_code": plan.get("platform_code"), + "platform_name": plan.get("platform_name"), + "base_url": plan.get("base_url"), + "safety_policy": plan.get("safety_policy"), + "network_request_allowed": bool(plan.get("network_request_allowed")), + "database_write_allowed": bool(plan.get("database_write_allowed")), + "sources": list(plan.get("sources") or []), + } + + +def _public_browser_context(platform: str) -> dict[str, Any]: + return { + "context_policy": "public_empty_browser_context_no_login", + "locale": "zh-TW", + "timezone_id": "Asia/Taipei", + "viewport": {"name": "desktop-1440", "width": 1440, "height": 950}, + "extra_http_headers": { + "Accept-Language": DEFAULT_ACCEPT_LANGUAGE, + }, + "credentialed_session_allowed": False, + "storage_state_allowed": False, + "raw_cookie_or_session_read_allowed": False, + "login_allowed": False, + "cart_or_checkout_allowed": False, + "platform": platform, + } + + +def _barrier_type_from_signals( + *, + platform: str, + http_status: int, + visual_barrier_reason: str, + url: str, + title: str, + validation: Mapping[str, Any] | None = None, + parsed_output: Mapping[str, Any] | None = None, +) -> str: + validation = validation or {} + parsed_output = parsed_output or {} + haystack = " ".join( + [ + visual_barrier_reason, + url, + title, + _safe_text(parsed_output.get("notes")), + _safe_text((parsed_output.get("fields") or {}).get("title")), + ] + ).lower() + if http_status in {401, 403} or "access denied" in haystack: + return "access_denied" + if "verify/traffic" in haystack or "traffic" in haystack: + return "traffic_verification_interstitial" + if ( + validation.get("interstitial_signal_detected") + or "language selection" in haystack + or "select language" in haystack + or "choose language" in haystack + or "region selection" in haystack + or "select region" in haystack + or "language" in haystack + or "語言" in haystack + ): + return "language_or_region_interstitial" + if validation.get("generic_marketplace_title_detected") or ( + platform == "shopee_tw" and "花得更少買得更好" in haystack + ): + return "generic_marketplace_landing" + if visual_barrier_reason: + return "platform_visual_barrier" + if validation.get("non_product_or_interstitial_detected"): + return "non_product_or_interstitial" + return "unknown_platform_barrier" + + +def _probe_status(platform: str, barrier_type: str) -> str: + if platform == "shopee_tw" and barrier_type in { + "language_or_region_interstitial", + "generic_marketplace_landing", + "traffic_verification_interstitial", + }: + return "ready_for_public_context_probe" + if barrier_type in {"access_denied", "traffic_verification_interstitial"}: + return "structured_source_or_backoff_required" + return "ready_for_platform_probe" + + +def _next_machine_action(platform: str, barrier_type: str, status: str) -> str: + if platform == "shopee_tw" and status == "ready_for_public_context_probe": + return "run_shopee_public_context_probe_then_structured_source_fallback" + if status == "structured_source_or_backoff_required": + return "use_structured_source_or_platform_backoff_policy" + return "run_public_platform_context_probe" + + +def _recommended_actions(platform: str, barrier_type: str, status: str) -> list[dict[str, Any]]: + actions: list[dict[str, Any]] = [] + if status == "ready_for_public_context_probe": + actions.append({ + "order": 1, + "action": "rerun_visual_capture_with_public_browser_context", + "machine_runnable": True, + "context_keys": ["locale", "timezone_id", "extra_http_headers", "viewport"], + }) + actions.append({ + "order": len(actions) + 1, + "action": "read_structured_market_intel_adapter_sources", + "machine_runnable": True, + "adapter_code": _adapter_code(platform), + }) + if barrier_type in {"access_denied", "traffic_verification_interstitial"}: + actions.append({ + "order": len(actions) + 1, + "action": "apply_platform_backoff_and_do_not_treat_barrier_as_product_data", + "machine_runnable": True, + }) + actions.append({ + "order": len(actions) + 1, + "action": "keep_visual_fields_out_of_formal_price_tables", + "machine_runnable": True, + }) + return actions + + +def _manifest_preview(platform: str, url: str) -> dict[str, Any] | None: + keyword = _extract_keyword(url) + if not keyword: + return None + manifest = build_pixelrag_marketplace_search_manifest( + platform=platform, + keyword=keyword, + crawler="PixelRAGPlatformProbe.public_context_visual_fallback", + trigger_reason="platform_interstitial_or_blocked_page_probe", + evidence_intent="collect_public_marketplace_offer_cards_after_platform_probe", + ) + if manifest.get("success"): + manifest["public_browser_context"] = _public_browser_context(platform) + return manifest + + +def _vlm_receipt_candidates( + root: Path, + *, + platforms: tuple[str, ...], + limit: int, +) -> list[Path]: + if not root.exists(): + return [] + candidates: list[Path] = [] + if platforms: + for platform in platforms: + candidates.extend((root / platform).glob("*/vlm_replay_receipt.json")) + else: + candidates.extend(root.glob("*/*/vlm_replay_receipt.json")) + return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit] + + +def _candidate_from_capture(candidate: Mapping[str, Any]) -> dict[str, Any] | None: + next_action = str(candidate.get("next_machine_action") or "") + visual_barrier_reason = str(candidate.get("visual_barrier_reason") or "") + http_status = _safe_int(candidate.get("http_status")) + if ( + "platform_probe" not in next_action + and not visual_barrier_reason + and http_status < 400 + ): + return None + + platform = str(candidate.get("platform") or "unknown").strip().lower() + manifest_id = str(candidate.get("manifest_id") or "").strip() + url = str(candidate.get("url") or "").strip() + title = str(candidate.get("title") or "").strip() + barrier_type = _barrier_type_from_signals( + platform=platform, + http_status=http_status, + visual_barrier_reason=visual_barrier_reason, + url=url, + title=title, + ) + status = _probe_status(platform, barrier_type) + return { + "platform": platform, + "manifest_id": manifest_id, + "source_type": "capture_receipt", + "source_receipt_path": candidate.get("receipt_path"), + "generated_at": candidate.get("generated_at"), + "age_hours": candidate.get("age_hours"), + "url": url, + "title": title, + "http_status": http_status, + "barrier_type": barrier_type, + "visual_barrier_reason": visual_barrier_reason, + "probe_status": status, + "probe_ready": True, + "next_machine_action": _next_machine_action(platform, barrier_type, status), + "public_browser_context": _public_browser_context(platform), + "capture_manifest_preview": _manifest_preview(platform, url), + "structured_source_fallback": _structured_source_plan(platform), + "recommended_probe_actions": _recommended_actions(platform, barrier_type, status), + "source_next_machine_action": candidate.get("next_machine_action"), + "writes_database": False, + "primary_human_gate_count": 0, + } + + +def _candidate_from_vlm(path: Path, *, now: datetime, max_age_hours: int) -> dict[str, Any] | None: + receipt, error = _read_json(path) + if error: + return { + "platform": path.parent.parent.name, + "manifest_id": path.parent.name, + "source_type": "vlm_replay_receipt", + "source_receipt_path": str(path), + "probe_status": "invalid_vlm_receipt", + "probe_ready": False, + "barrier_type": "invalid_receipt", + "errors": [error], + "next_machine_action": "fix_invalid_pixelrag_vlm_receipt", + "writes_database": False, + "primary_human_gate_count": 0, + } + + validation = receipt.get("validation") if isinstance(receipt.get("validation"), Mapping) else {} + parsed_output = receipt.get("parsed_output") if isinstance(receipt.get("parsed_output"), Mapping) else {} + next_action = str(receipt.get("next_machine_action") or "") + if ( + "platform_probe" not in next_action + and not validation.get("non_product_or_interstitial_detected") + and not validation.get("blocked_page_detected") + ): + return None + + platform = str(receipt.get("platform") or path.parent.parent.name).strip().lower() + manifest_id = str(receipt.get("manifest_id") or path.parent.name).strip() + generated = _parse_iso_datetime(receipt.get("generated_at")) + age_hours = ((now - generated).total_seconds() / 3600) if generated else None + source_url = "" + source_title = "" + source_receipt = str(receipt.get("source_receipt_path") or "") + if source_receipt: + capture, _ = _read_json(Path(source_receipt)) + capture_target = capture.get("capture_target") or {} + page_metrics = capture.get("page_metrics") or {} + source_url = str(capture_target.get("url") or page_metrics.get("final_url") or "") + source_title = str(page_metrics.get("title") or "") + barrier_type = _barrier_type_from_signals( + platform=platform, + http_status=0, + visual_barrier_reason="", + url=source_url, + title=source_title, + validation=validation, + parsed_output=parsed_output, + ) + status = _probe_status(platform, barrier_type) + return { + "platform": platform, + "manifest_id": manifest_id, + "source_type": "vlm_replay_receipt", + "source_receipt_path": str(path), + "source_capture_receipt_path": source_receipt, + "generated_at": receipt.get("generated_at"), + "age_hours": round(age_hours, 3) if age_hours is not None else None, + "stale": age_hours is None or age_hours > max_age_hours, + "url": source_url, + "title": source_title, + "http_status": 0, + "barrier_type": barrier_type, + "probe_status": status, + "probe_ready": True, + "validation": { + "blocked_page_detected": bool(validation.get("blocked_page_detected")), + "non_product_or_interstitial_detected": bool( + validation.get("non_product_or_interstitial_detected") + ), + "interstitial_signal_detected": bool( + validation.get("interstitial_signal_detected") + ), + "generic_marketplace_title_detected": bool( + validation.get("generic_marketplace_title_detected") + ), + "present_field_count": _safe_int(validation.get("present_field_count")), + "missing_required_fields": list(validation.get("missing_required_fields") or []), + }, + "next_machine_action": _next_machine_action(platform, barrier_type, status), + "public_browser_context": _public_browser_context(platform), + "capture_manifest_preview": _manifest_preview(platform, source_url), + "structured_source_fallback": _structured_source_plan(platform), + "recommended_probe_actions": _recommended_actions(platform, barrier_type, status), + "source_next_machine_action": receipt.get("next_machine_action"), + "writes_database": False, + "primary_human_gate_count": 0, + } + + +def _dedupe_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]: + by_key: dict[tuple[str, str], dict[str, Any]] = {} + source_rank = {"vlm_replay_receipt": 2, "capture_receipt": 1} + for item in items: + key = (str(item.get("platform") or ""), str(item.get("manifest_id") or "")) + current = by_key.get(key) + if not current or source_rank.get(str(item.get("source_type")), 0) >= source_rank.get( + str(current.get("source_type")), 0 + ): + by_key[key] = item + return list(by_key.values()) + + +def build_pixelrag_platform_probe_readiness( + *, + artifact_root: str | Path | None = None, + vlm_receipt_root: str | Path | None = None, + platform: str | tuple[str, ...] | list[str] | None = None, + max_age_hours: int | None = None, + limit: int | None = None, +) -> dict[str, Any]: + """Build a no-write platform probe plan from PixelRAG barrier receipts.""" + capture_root = Path(artifact_root or DEFAULT_ARTIFACT_ROOT) + vlm_root = Path(vlm_receipt_root or DEFAULT_VLM_RECEIPT_ROOT) + platforms = _normalise_platforms(platform) + max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS)) + item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250)) + now = datetime.now(timezone.utc) + + capture_readback = build_pixelrag_rag_candidate_replay_readback( + artifact_root=capture_root, + platform=platforms, + max_age_hours=max_age, + limit=item_limit, + ) + capture_items = [ + item + for item in ( + _candidate_from_capture(candidate) + for candidate in list(capture_readback.get("candidates") or []) + ) + if item + ] + vlm_items = [ + item + for item in ( + _candidate_from_vlm(path, now=now, max_age_hours=max_age) + for path in _vlm_receipt_candidates(vlm_root, platforms=platforms, limit=item_limit) + ) + if item + ] + probe_items = _dedupe_items(capture_items + vlm_items) + invalid_count = sum(1 for item in probe_items if not item.get("probe_ready")) + ready_count = sum(1 for item in probe_items if item.get("probe_ready")) + shopee_public_context_count = sum( + 1 + for item in probe_items + if item.get("platform") == "shopee_tw" + and item.get("probe_status") == "ready_for_public_context_probe" + ) + access_denied_count = sum(1 for item in probe_items if item.get("barrier_type") == "access_denied") + traffic_count = sum( + 1 for item in probe_items if item.get("barrier_type") == "traffic_verification_interstitial" + ) + language_count = sum( + 1 for item in probe_items if item.get("barrier_type") == "language_or_region_interstitial" + ) + structured_fallback_count = sum( + 1 + for item in probe_items + if (item.get("structured_source_fallback") or {}).get("available") + ) + + if invalid_count and invalid_count == len(probe_items): + status = "critical" + elif probe_items and ready_count: + status = "ok" + else: + status = "warning" + + next_action = ( + "fix_invalid_pixelrag_platform_probe_receipts" + if invalid_count and not ready_count + else ( + "run_platform_probe_or_structured_source_fallback" + if ready_count + else "run_pixelrag_visual_capture_worker" + ) + ) + + return { + "success": status != "critical", + "policy": POLICY, + "status": status, + "generated_at": now.isoformat(), + "artifact_root": str(capture_root), + "vlm_receipt_root": str(vlm_root), + "platform_filter": list(platforms), + "max_age_hours": max_age, + "limit": item_limit, + "summary": { + "probe_candidate_count": len(probe_items), + "ready_for_probe_count": ready_count, + "invalid_count": invalid_count, + "capture_source_count": len(capture_items), + "vlm_source_count": len(vlm_items), + "shopee_public_context_probe_count": shopee_public_context_count, + "language_or_region_interstitial_count": language_count, + "traffic_verification_count": traffic_count, + "access_denied_count": access_denied_count, + "structured_source_fallback_count": structured_fallback_count, + "writes_database_count": 0, + "primary_human_gate_count": 0, + "platforms": sorted({str(item.get("platform") or "unknown") for item in probe_items}), + }, + "probe_items": probe_items, + "source_capture_replay": { + "policy": capture_readback.get("policy"), + "status": capture_readback.get("status"), + "summary": capture_readback.get("summary"), + "next_machine_action": capture_readback.get("next_machine_action"), + }, + "probe_contract": { + "automation_mode": "platform_probe_plan_no_write", + "network_call": False, + "db_write": False, + "writes_database": False, + "writes_ai_insights": False, + "writes_price_tables": False, + "secret_read": False, + "raw_cookie_or_session_read": False, + "credentialed_session_allowed": False, + "login_allowed": False, + "blocked_pages_are_not_product_data": True, + "visual_fields_are_candidate_evidence_only": True, + "primary_human_gate_count": 0, + }, + "controlled_apply": { + "network_call": False, + "db_write": False, + "writes_database": False, + "writes_database_count": 0, + "secret_read": False, + "raw_cookie_or_session_read": False, + "production_price_write": False, + "artifact_write": False, + "primary_human_gate_count": 0, + }, + "next_machine_action": next_action, + } + + +__all__ = [ + "POLICY", + "build_pixelrag_platform_probe_readiness", +] diff --git a/tests/test_ai_automation_smoke_service.py b/tests/test_ai_automation_smoke_service.py index 8f57ce3..a32752e 100644 --- a/tests/test_ai_automation_smoke_service.py +++ b/tests/test_ai_automation_smoke_service.py @@ -1436,11 +1436,12 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch): monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok")) + monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check("pixelrag platform probe", "ok", "ok")) result = smoke.collect_ai_automation_smoke(record_history=False) assert result["status"] == "critical" - assert result["summary"] == {"ok": 34, "warning": 1, "critical": 1, "total": 36} + assert result["summary"] == {"ok": 35, "warning": 1, "critical": 1, "total": 37} def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch): @@ -3972,6 +3973,7 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat monkeypatch.setattr(smoke, "_pixelrag_ocr_vlm_replay_check", lambda: smoke._check("pixelrag ocr vlm", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_vlm_route_readiness_check", lambda: smoke._check("pixelrag vlm route", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_vlm_replay_worker_check", lambda: smoke._check("pixelrag vlm worker", "ok", "ok")) + monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check("pixelrag platform probe", "ok", "ok")) first = smoke.collect_ai_automation_smoke(history_limit=5) second = smoke.collect_ai_automation_smoke(history_limit=5) @@ -4027,7 +4029,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( json.dumps({ "generated_at": datetime.now().isoformat(timespec="seconds"), "status": "ok", - "summary": {"ok": 36, "warning": 0, "critical": 0, "total": 36}, + "summary": {"ok": 37, "warning": 0, "critical": 0, "total": 37}, "checks": [ { "name": "PChome 受控落地 drift monitor", @@ -4192,6 +4194,24 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( "artifact_write_performed": False, "next_machine_action": "run_pixelrag_vlm_replay_worker_execute", }, + }, + { + "name": "PixelRAG platform probe readiness", + "status": "ok", + "summary": "PixelRAG platform probe candidates=1, ready=1, shopee_context=1, structured_fallback=1", + "details": { + "policy": "read_only_pixelrag_platform_probe_readiness_v1", + "probe_candidate_count": 1, + "ready_for_probe_count": 1, + "shopee_public_context_probe_count": 1, + "language_or_region_interstitial_count": 1, + "traffic_verification_count": 0, + "access_denied_count": 0, + "structured_source_fallback_count": 1, + "next_machine_action": "run_platform_probe_or_structured_source_fallback", + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, } ], }, ensure_ascii=False) + "\n", @@ -4210,7 +4230,7 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( ) assert summary["policy"] == "read_only_ai_automation_scheduled_health_summary" assert summary["status"] == "ok" - assert summary["summary"]["total"] == 33 + assert summary["summary"]["total"] == 34 assert summary["summary"]["primary_human_gate_count"] == 0 assert summary["summary"]["writes_database_count"] == 0 assert pchome_family["status"] == "ok" @@ -5910,6 +5930,13 @@ def test_scheduled_automation_health_summary_reads_history_without_side_effects( assert pixelrag_vlm_replay_worker_family["details"]["dry_run_count"] == 1 assert pixelrag_vlm_replay_worker_family["details"]["model_call_performed"] is False assert pixelrag_vlm_replay_worker_family["details"]["writes_database_count"] == 0 + pixelrag_platform_probe_family = next( + item for item in summary["families"] + if item["key"] == "pixelrag_platform_probe" + ) + assert pixelrag_platform_probe_family["status"] == "ok" + assert pixelrag_platform_probe_family["details"]["ready_for_probe_count"] == 1 + assert pixelrag_platform_probe_family["details"]["primary_human_gate_count"] == 0 assert summary["scheduled_outputs"]["telegram_send_in_preview"] is False assert summary["scheduled_outputs"]["writes_database"] is False assert summary["automation_policy"]["primary_flow"] == "ai_controlled" @@ -6541,6 +6568,11 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): "ok", "pixelrag vlm worker ok", )) + monkeypatch.setattr(smoke, "_pixelrag_platform_probe_check", lambda: smoke._check( + "PixelRAG platform probe readiness", + "ok", + "pixelrag platform probe ok", + )) result = smoke.collect_ai_automation_smoke(record_history=False) @@ -6556,7 +6588,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): item for item in result["checks"] if item["name"] == "Sitewide visual QA readback" ) - assert result["summary"]["total"] == 36 + assert result["summary"]["total"] == 37 assert surface_check["status"] == "ok" assert surface_check["details"]["checked_surface_count"] == 10 assert sitewide_check["status"] == "ok" diff --git a/tests/test_pixelrag_platform_probe_service.py b/tests/test_pixelrag_platform_probe_service.py new file mode 100644 index 0000000..8a98f08 --- /dev/null +++ b/tests/test_pixelrag_platform_probe_service.py @@ -0,0 +1,240 @@ +import json +import subprocess +import sys +from datetime import datetime, timezone + +from tests.test_pixelrag_rag_candidate_replay_service import _write_receipt + + +def _patch_receipt(path, **updates): + payload = json.loads(path.read_text(encoding="utf-8")) + for key, value in updates.items(): + if key == "page_metrics": + payload.setdefault("page_metrics", {}).update(value) + else: + payload[key] = value + path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") + return payload + + +def test_pixelrag_platform_probe_builds_machine_actions_for_shopee_and_coupang(tmp_path): + from services.pixelrag_platform_probe_service import ( + POLICY, + build_pixelrag_platform_probe_readiness, + ) + + shopee_receipt = _write_receipt( + tmp_path, + platform="shopee_tw", + manifest_id="shopee-traffic", + title="蝦皮購物 | 花得更少買得更好", + url="https://shopee.tw/search?keyword=sunscreen", + ) + _patch_receipt( + shopee_receipt, + page_metrics={ + "final_url": "https://shopee.tw/verify/traffic/error?home_url=https%3A%2F%2Fshopee.tw", + "title": "蝦皮購物 | 花得更少買得更好", + }, + ) + _write_receipt( + tmp_path, + platform="coupang_tw", + manifest_id="coupang-403", + title="Access Denied", + url="https://www.tw.coupang.com/search?q=iphone", + http_status=403, + ) + + payload = build_pixelrag_platform_probe_readiness(artifact_root=tmp_path) + by_platform = {item["platform"]: item for item in payload["probe_items"]} + + assert payload["policy"] == POLICY + assert payload["status"] == "ok" + assert payload["summary"]["probe_candidate_count"] == 2 + assert payload["summary"]["ready_for_probe_count"] == 2 + assert payload["summary"]["traffic_verification_count"] == 1 + assert payload["summary"]["access_denied_count"] == 1 + assert payload["summary"]["writes_database_count"] == 0 + assert payload["controlled_apply"]["primary_human_gate_count"] == 0 + assert by_platform["shopee_tw"]["probe_status"] == "ready_for_public_context_probe" + assert by_platform["shopee_tw"]["next_machine_action"] == ( + "run_shopee_public_context_probe_then_structured_source_fallback" + ) + assert by_platform["shopee_tw"]["public_browser_context"]["locale"] == "zh-TW" + assert by_platform["shopee_tw"]["public_browser_context"]["storage_state_allowed"] is False + assert by_platform["shopee_tw"]["capture_manifest_preview"]["success"] is True + assert by_platform["coupang_tw"]["probe_status"] == "structured_source_or_backoff_required" + assert by_platform["coupang_tw"]["next_machine_action"] == ( + "use_structured_source_or_platform_backoff_policy" + ) + assert by_platform["coupang_tw"]["structured_source_fallback"]["available"] is True + assert payload["probe_contract"]["raw_cookie_or_session_read"] is False + assert payload["probe_contract"]["blocked_pages_are_not_product_data"] is True + + +def test_pixelrag_platform_probe_reads_vlm_interstitial_receipts(tmp_path): + from services.pixelrag_platform_probe_service import build_pixelrag_platform_probe_readiness + + capture_path = _write_receipt( + tmp_path / "capture", + platform="shopee_tw", + manifest_id="shopee-language", + title="Shopee", + url="https://shopee.tw/search?keyword=sunscreen", + ) + vlm_dir = tmp_path / "vlm" / "shopee_tw" / "shopee-language" + vlm_dir.mkdir(parents=True) + (vlm_dir / "vlm_replay_receipt.json").write_text( + json.dumps({ + "generated_at": datetime.now(timezone.utc).isoformat(), + "platform": "shopee_tw", + "manifest_id": "shopee-language", + "source_receipt_path": str(capture_path), + "worker_status": "executed_warning", + "next_machine_action": "run_platform_probe_or_use_structured_api", + "parsed_output": { + "blocked_page_detected": False, + "fields": { + "title": { + "value": "蝦皮購物 | 花得更少買得更好", + "confidence": 0.95, + "evidence_refs": ["tile:1"], + }, + "price": {"value": None, "confidence": 0.0, "evidence_refs": []}, + }, + "notes": [{"note": "Language selection page"}], + }, + "validation": { + "blocked_page_detected": False, + "non_product_or_interstitial_detected": True, + "interstitial_signal_detected": True, + "generic_marketplace_title_detected": True, + "present_field_count": 1, + "missing_required_fields": ["price"], + }, + }, ensure_ascii=False), + encoding="utf-8", + ) + + payload = build_pixelrag_platform_probe_readiness( + artifact_root=tmp_path / "capture", + vlm_receipt_root=tmp_path / "vlm", + platform="shopee_tw", + ) + + assert payload["status"] == "ok" + assert payload["summary"]["probe_candidate_count"] == 1 + assert payload["summary"]["vlm_source_count"] == 1 + assert payload["summary"]["language_or_region_interstitial_count"] == 1 + item = payload["probe_items"][0] + assert item["source_type"] == "vlm_replay_receipt" + assert item["barrier_type"] == "language_or_region_interstitial" + assert item["probe_status"] == "ready_for_public_context_probe" + assert item["primary_human_gate_count"] == 0 + + +def test_pixelrag_platform_probe_cli_outputs_machine_readable_json(tmp_path): + _write_receipt( + tmp_path, + platform="coupang_tw", + manifest_id="coupang-403", + title="Access Denied", + url="https://www.tw.coupang.com/search?q=iphone", + http_status=403, + ) + + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/report_pixelrag_platform_probe.py", + "--artifact-root", + str(tmp_path), + "--platform", + "coupang_tw", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["summary"]["probe_candidate_count"] == 1 + assert payload["probe_items"][0]["platform"] == "coupang_tw" + assert payload["probe_contract"]["writes_database"] is False + + +def test_pixelrag_capture_worker_dry_run_uses_safe_public_context(tmp_path): + from services.pixelrag_crawler_integration_service import ( + build_pixelrag_marketplace_search_manifest, + ) + + manifest = build_pixelrag_marketplace_search_manifest( + platform="shopee_tw", + keyword="sunscreen", + crawler="PixelRAGPlatformProbe.public_context_visual_fallback", + trigger_reason="platform_interstitial_or_blocked_page_probe", + ) + manifest["public_browser_context"] = { + "locale": "zh-TW", + "timezone_id": "Asia/Taipei", + "extra_http_headers": { + "Accept-Language": "zh-TW,zh-Hant;q=0.95,zh;q=0.9,en;q=0.7", + "Cookie": "must_not_survive", + "Authorization": "must_not_survive", + }, + } + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/capture_pixelrag_visual_evidence.py", + "--manifest-json", + json.dumps(manifest, ensure_ascii=False), + "--output-dir", + str(tmp_path / "artifacts"), + "--dry-run", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + context = payload["public_browser_context"] + assert context["locale"] == "zh-TW" + assert context["timezone_id"] == "Asia/Taipei" + assert context["extra_http_headers"]["Accept-Language"].startswith("zh-TW") + assert "Cookie" not in context["extra_http_headers"] + assert "Authorization" not in context["extra_http_headers"] + assert context["raw_cookie_or_session_read_allowed"] is False + + +def test_pixelrag_platform_probe_route_returns_readback(tmp_path, monkeypatch): + from flask import Flask + from routes import system_public_routes as routes + from services import pixelrag_platform_probe_service as service + + _write_receipt( + tmp_path, + platform="coupang_tw", + manifest_id="coupang-403", + title="Access Denied", + url="https://www.tw.coupang.com/search?q=iphone", + http_status=403, + ) + monkeypatch.setattr(service, "DEFAULT_ARTIFACT_ROOT", str(tmp_path)) + + app = Flask(__name__) + with app.test_request_context( + "/api/ai-automation/pixelrag-platform-probe?platform=coupang_tw" + ): + response = routes.ai_automation_pixelrag_platform_probe_api.__wrapped__() + payload = response.get_json() + + assert payload["policy"] == "read_only_pixelrag_platform_probe_readiness_v1" + assert payload["summary"]["probe_candidate_count"] == 1 + assert payload["probe_items"][0]["next_machine_action"] == ( + "use_structured_source_or_platform_backoff_policy" + )