From b120230cae0cf77befaa190e5a03c4250555b564 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 15:42:00 +0800 Subject: [PATCH] refactor(pchome): extract evidence enrichment --- config.py | 2 +- docs/AI_INTELLIGENCE_MODULE_SOT.md | 2 +- .../code_modularization_inventory_20260430.md | 2 +- services/pchome_mapping_backlog/enrichment.py | 388 ++++++++++++++++++ services/pchome_mapping_backlog_service.py | 318 +------------- ...t_pchome_mapping_backlog_modularization.py | 48 ++- 6 files changed, 445 insertions(+), 315 deletions(-) create mode 100644 services/pchome_mapping_backlog/enrichment.py diff --git a/config.py b/config.py index de8e0f9..bc1d4fb 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.784" +SYSTEM_VERSION = "V10.785" 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 390409c..88367c9 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -2,7 +2,7 @@ > **最後更新**: 2026-07-11 (台北時間) > **狀態**: 🟠 Partial。四 Agent、PixelRAG、MCP/RAG registry、PChome controlled-preview families 與多項 smoke/readback 已建立；但 access control、database identity/RBAC、full asset reconciliation、software supply chain、same-run controlled apply、restore drill、internal RAG canary 與 MCP/RAG runtime closure 尚未全部完成。任何局部 receipt 不得代表整體閉環完成。 -> **適用版本**: V10.784 +> **適用版本**: V10.785 --- diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index f5c3ea6..0b64af3 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -129,7 +129,7 @@ | 行數 | 檔案 | 分類 | 拆分方向 | |---:|---|---|---| -| 43183 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier、receipt replay 與 reporter 已移到 `services/pchome_mapping_backlog/`；下一步拆 evidence enrichment orchestration | +| 42881 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier、receipt replay、reporter 與 evidence enrichment/source preview 已移到 `services/pchome_mapping_backlog/`；下一步拆 fetch/merge orchestration | | 14289 | `services/ai_automation_smoke_service.py` | P0 smoke 巨型 service | 拆 family registry、collectors、health projection、metrics adapter | | 9383 | `routes/openclaw_bot_routes.py` | P0 巨型 Blueprint、拆分進行中 | scheduler hook 已移到 `services/openclaw_bot/scheduled_jobs.py`；下一步拆 report、command | | 7641 | `routes/ai_routes.py` | P0 巨型 Blueprint | PChome growth、AI automation、recommendation route extension 分離 | diff --git a/services/pchome_mapping_backlog/enrichment.py b/services/pchome_mapping_backlog/enrichment.py new file mode 100644 index 0000000..114d875 --- /dev/null +++ b/services/pchome_mapping_backlog/enrichment.py @@ -0,0 +1,388 @@ +"""Read-only evidence enrichment and source-wiring previews.""" + +from __future__ import annotations + +from typing import Any + +from services.ai_exception_contract import LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY +from services.pchome_mapping_backlog.contracts import ( + _ai_exception_compatibility_fields, + _evidence_requires_ai_exception, +) +from services.pchome_mapping_backlog.policies import ( + EVIDENCE_ENRICHMENT_PREVIEW_POLICY, + EVIDENCE_SOURCE_PREVIEW_POLICY, + PCHOME_FETCH_MAX_BATCH_SIZE, + PRODUCT_PAGE_EVIDENCE_PARSER_POLICY, +) +from services.pchome_mapping_backlog.reporter import ( + build_pchome_mapping_operator_preview, +) + + +def _field_enrichment_sources(field: str) -> list[dict[str, Any]]: + source_map = { + "image": [ + { + "source": "PChome product page structured data", + "mode": "future_read_only_fetch", + "writes_database": False, + "expected_output": "primary product image URL", + }, + { + "source": "existing marketplace catalog payload", + "mode": "reuse_if_present", + "writes_database": False, + "expected_output": "cached image_url", + }, + ], + "availability": [ + { + "source": "PChome product page offer availability", + "mode": "future_read_only_fetch", + "writes_database": False, + "expected_output": "in_stock / out_of_stock / unknown", + }, + { + "source": "merchant listing structured data", + "mode": "future_read_only_parse", + "writes_database": False, + "expected_output": "schema.org Offer availability", + }, + ], + "unit_price_or_package_basis": [ + { + "source": "deterministic product title parser", + "mode": "local_parse_preview", + "writes_database": False, + "expected_output": "size, count, unit label, package basis", + }, + { + "source": "external_price.price_basis", + "mode": "reuse_if_present", + "writes_database": False, + "expected_output": "total_price / unit_price", + }, + ], + "price": [ + { + "source": "growth opportunity payload", + "mode": "reuse_if_present", + "writes_database": False, + "expected_output": "PChome listed price", + }, + ], + } + return source_map.get(field, []) + + +def _source_plan_for_field(field: str, missing_count: int) -> dict[str, Any]: + plans = { + "image": { + "payload_keys_checked": ["image_url", "image", "product_image_url"], + "preferred_current_source": "existing marketplace catalog payload", + "future_read_only_fetch_gate": { + "method": "GET", + "allowed_domain": "24h.pchome.com.tw", + "product_url_required": True, + "parse_targets": [ + "schema.org Product.image", + "og:image", + "primary product image", + ], + "check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY, + "fetches_external_sites_in_preview": False, + "writes_database": False, + }, + }, + "availability": { + "payload_keys_checked": [ + "availability", + "stock_status", + "is_available", + ], + "preferred_current_source": "existing marketplace catalog payload", + "future_read_only_fetch_gate": { + "method": "GET", + "allowed_domain": "24h.pchome.com.tw", + "product_url_required": True, + "parse_targets": [ + "schema.org Offer.availability", + "merchant listing offer availability", + ], + "check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY, + "fetches_external_sites_in_preview": False, + "writes_database": False, + }, + }, + "price": { + "payload_keys_checked": [ + "pchome_price", + "external_price.pchome_price", + "review_candidate.pchome_price", + ], + "preferred_current_source": "growth opportunity payload", + "future_read_only_fetch_gate": None, + "payload_mapping_probe": { + "goal": "Confirm whether missing price is a source payload gap or summary field mapping gap.", + "fetches_external_sites_in_preview": False, + "writes_database": False, + }, + }, + } + plan = dict(plans.get(field, {})) + if not plan: + return {} + plan["field"] = field + plan["status"] = ( + "missing_in_current_payload" + if missing_count + else "covered_by_current_payload" + ) + plan["missing_count"] = missing_count + return plan + + +def _build_fetch_gate_candidates( + tasks: list[dict[str, Any]], +) -> list[dict[str, Any]]: + candidates = [] + for task in tasks: + missing_fields = set(task.get("missing_fields") or []) + fetch_fields = [ + field + for field in ("image", "availability") + if field in missing_fields + ] + if not fetch_fields: + continue + candidates.append( + { + "pchome_product_id": task.get("pchome_product_id") or "", + "product_name": task.get("product_name") or "", + "product_url": task.get("product_url"), + "fields": fetch_fields, + "method": "GET", + "allowed_domain": "24h.pchome.com.tw", + "executes_fetch_in_preview": False, + "writes_database": False, + } + ) + return candidates[:PCHOME_FETCH_MAX_BATCH_SIZE] + + +def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]: + evidence = target.get("evidence_completeness") or {} + missing_fields = list(evidence.get("missing_fields") or []) + blocking_missing_fields = list(evidence.get("blocking_missing_fields") or []) + enrichment_steps = [ + { + "field": field, + "blocking": field in blocking_missing_fields, + "sources": _field_enrichment_sources(field), + } + for field in missing_fields + ] + ai_exception_required = _evidence_requires_ai_exception(evidence) + + return { + "lane": lane, + "pchome_product_id": target.get("pchome_product_id") or "", + "product_name": target.get("product_name") or "", + "product_url": target.get("product_url") or evidence.get("product_url"), + "sales_7d": target.get("sales_7d"), + "priority_score": target.get("priority_score"), + "action_code": target.get("action_code"), + "evidence_score": evidence.get("score"), + "present_fields": list(evidence.get("present_fields") or []), + "missing_fields": missing_fields, + "blocking_missing_fields": blocking_missing_fields, + "auto_accept_ready": bool(evidence.get("auto_accept_ready")), + **_ai_exception_compatibility_fields(ai_exception_required), + "unit_package_basis": evidence.get("unit_package_basis"), + "enrichment_steps": enrichment_steps, + } + + +def build_pchome_evidence_enrichment_preview( + payload: dict[str, Any], + batch_size: int = 5, +) -> dict[str, Any]: + """Build a read-only evidence enrichment package for mapping targets.""" + operator_preview = build_pchome_mapping_operator_preview( + payload, + batch_size=batch_size, + ) + operator_batch = operator_preview.get("operator_batch") or {} + direct_targets = list(operator_batch.get("direct_mapping_targets") or []) + review_targets = list(operator_batch.get("review_candidate_targets") or []) + evidence_tasks = [ + *[ + _build_evidence_task(target, "direct_mapping") + for target in direct_targets + ], + *[ + _build_evidence_task(target, "review_candidate") + for target in review_targets + ], + ] + tasks_with_blockers = [ + task for task in evidence_tasks if task.get("blocking_missing_fields") + ] + missing_field_counts: dict[str, int] = {} + for task in evidence_tasks: + for field in task.get("missing_fields") or []: + missing_field_counts[field] = missing_field_counts.get(field, 0) + 1 + + if tasks_with_blockers: + result = "NEEDS_EVIDENCE_ENRICHMENT" + elif evidence_tasks: + result = "EVIDENCE_PREVIEW_READY" + else: + result = "NO_TARGETS" + + return { + "policy": EVIDENCE_ENRICHMENT_PREVIEW_POLICY, + "result": result, + "success": bool(operator_preview.get("success")), + "generated_at": operator_preview.get("generated_at"), + "source_policy": operator_preview.get("policy"), + "stats": operator_preview.get("stats") or {}, + "summary": { + "task_count": len(evidence_tasks), + "tasks_with_blockers": len(tasks_with_blockers), + "missing_field_counts": missing_field_counts, + "auto_accept_ready_count": sum( + 1 for task in evidence_tasks if task.get("auto_accept_ready") + ), + LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY: 0, + "primary_human_gate_count": 0, + "ai_exception_required_count": sum( + 1 + for task in evidence_tasks + if task.get("ai_exception_required") + ), + }, + "evidence_tasks": evidence_tasks, + "external_benchmark_alignment": ( + operator_preview.get("external_benchmark_alignment") or {} + ), + "ai_automation_plan": { + "policy": "ollama_first_read_only_evidence_assist", + "llm_calls_in_preview": False, + "gemini_allowed": False, + "can_execute_write": False, + "recommended_next_ai_task": "Generate deterministic identity anchors and evidence gap summaries after image and availability sources are wired.", + "blocked_until": [ + "image evidence source is wired", + "availability evidence source is wired", + "unit/package parser preview is compared against production title samples", + ], + }, + "safety": { + "read_only_preview": True, + "fetches_external_sites": False, + "writes_database": False, + "executes_search": False, + "dispatches_telegram": False, + "llm_calls_in_preview": False, + "requires_operator_write_approval": True, + }, + "next_actions": [ + "Wire read-only image and availability enrichment before expanding auto-accept.", + "Validate deterministic unit/package basis parsing for bundle-sensitive items.", + "Keep DB writes behind the existing /api/ai/pchome-growth/backfill-momo-candidates write gate.", + ], + } + + +def build_pchome_evidence_source_preview( + payload: dict[str, Any], + batch_size: int = 5, +) -> dict[str, Any]: + """Build a read-only source wiring preview for missing product evidence fields.""" + enrichment_preview = build_pchome_evidence_enrichment_preview( + payload, + batch_size=batch_size, + ) + tasks = list(enrichment_preview.get("evidence_tasks") or []) + fields = ["image", "availability", "price", "unit_price_or_package_basis"] + field_counts = {} + source_plans = {} + for field in fields: + missing_tasks = [ + task + for task in tasks + if field in (task.get("missing_fields") or []) + ] + field_counts[field] = { + "missing_count": len(missing_tasks), + "present_count": max(len(tasks) - len(missing_tasks), 0), + "sample_missing_targets": [ + { + "pchome_product_id": task.get("pchome_product_id") or "", + "product_name": task.get("product_name") or "", + "product_url": task.get("product_url"), + "lane": task.get("lane"), + } + for task in missing_tasks[:3] + ], + } + plan = _source_plan_for_field(field, len(missing_tasks)) + if plan: + source_plans[field] = plan + + fetch_gate_candidates = _build_fetch_gate_candidates(tasks) + if ( + field_counts["image"]["missing_count"] + or field_counts["availability"]["missing_count"] + ): + result = "NEEDS_SOURCE_WIRING" + elif field_counts["price"]["missing_count"]: + result = "NEEDS_PAYLOAD_MAPPING_PROBE" + elif tasks: + result = "SOURCE_PREVIEW_READY" + else: + result = "NO_TARGETS" + + return { + "policy": EVIDENCE_SOURCE_PREVIEW_POLICY, + "result": result, + "success": bool(enrichment_preview.get("success")), + "generated_at": enrichment_preview.get("generated_at"), + "source_policy": enrichment_preview.get("policy"), + "stats": enrichment_preview.get("stats") or {}, + "summary": { + "task_count": len(tasks), + "field_counts": field_counts, + "fetch_gate_candidate_count": len(fetch_gate_candidates), + }, + "source_plans": source_plans, + "fetch_gate_candidates": fetch_gate_candidates, + "external_benchmark_alignment": ( + enrichment_preview.get("external_benchmark_alignment") or {} + ), + "ai_automation_plan": { + "policy": "ollama_first_read_only_source_wiring_assist", + "llm_calls_in_preview": False, + "gemini_allowed": False, + "can_execute_fetch": False, + "can_execute_write": False, + "recommended_next_ai_task": "Generate schema-aware parsers for image and availability after fetch-gate tests are accepted.", + }, + "safety": { + "read_only_preview": True, + "fetches_external_sites": False, + "writes_database": False, + "executes_search": False, + "dispatches_telegram": False, + "llm_calls_in_preview": False, + "requires_production_version_truth": True, + "requires_fetch_gate_implementation_before_external_get": True, + }, + "next_actions": [ + "Wire image source preview from existing payload keys before adding a controlled product-page fetch gate.", + "Wire availability source preview from existing payload keys before adding a controlled Offer availability parser.", + "Probe missing price rows through current payload mapping before any external fetch or write.", + ], + } diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 116953e..29528da 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -162,6 +162,14 @@ from services.pchome_mapping_backlog.evidence import ( parse_pchome_product_page_evidence_html, parse_unit_package_basis, ) +from services.pchome_mapping_backlog.enrichment import ( + _build_evidence_task, + _build_fetch_gate_candidates, + _field_enrichment_sources, + _source_plan_for_field, + build_pchome_evidence_enrichment_preview, + build_pchome_evidence_source_preview, +) from services.pchome_mapping_backlog.reporter import ( _build_ai_automation_plan, _build_external_benchmark_alignment, @@ -172,165 +180,6 @@ from services.pchome_mapping_backlog.reporter import ( ) -def _field_enrichment_sources(field: str) -> list[dict[str, Any]]: - source_map = { - "image": [ - { - "source": "PChome product page structured data", - "mode": "future_read_only_fetch", - "writes_database": False, - "expected_output": "primary product image URL", - }, - { - "source": "existing marketplace catalog payload", - "mode": "reuse_if_present", - "writes_database": False, - "expected_output": "cached image_url", - }, - ], - "availability": [ - { - "source": "PChome product page offer availability", - "mode": "future_read_only_fetch", - "writes_database": False, - "expected_output": "in_stock / out_of_stock / unknown", - }, - { - "source": "merchant listing structured data", - "mode": "future_read_only_parse", - "writes_database": False, - "expected_output": "schema.org Offer availability", - }, - ], - "unit_price_or_package_basis": [ - { - "source": "deterministic product title parser", - "mode": "local_parse_preview", - "writes_database": False, - "expected_output": "size, count, unit label, package basis", - }, - { - "source": "external_price.price_basis", - "mode": "reuse_if_present", - "writes_database": False, - "expected_output": "total_price / unit_price", - }, - ], - "price": [ - { - "source": "growth opportunity payload", - "mode": "reuse_if_present", - "writes_database": False, - "expected_output": "PChome listed price", - }, - ], - } - return source_map.get(field, []) - - -def _source_plan_for_field(field: str, missing_count: int) -> dict[str, Any]: - plans = { - "image": { - "payload_keys_checked": ["image_url", "image", "product_image_url"], - "preferred_current_source": "existing marketplace catalog payload", - "future_read_only_fetch_gate": { - "method": "GET", - "allowed_domain": "24h.pchome.com.tw", - "product_url_required": True, - "parse_targets": ["schema.org Product.image", "og:image", "primary product image"], - "check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY, - "fetches_external_sites_in_preview": False, - "writes_database": False, - }, - }, - "availability": { - "payload_keys_checked": ["availability", "stock_status", "is_available"], - "preferred_current_source": "existing marketplace catalog payload", - "future_read_only_fetch_gate": { - "method": "GET", - "allowed_domain": "24h.pchome.com.tw", - "product_url_required": True, - "parse_targets": ["schema.org Offer.availability", "merchant listing offer availability"], - "check_mode_parser": PRODUCT_PAGE_EVIDENCE_PARSER_POLICY, - "fetches_external_sites_in_preview": False, - "writes_database": False, - }, - }, - "price": { - "payload_keys_checked": ["pchome_price", "external_price.pchome_price", "review_candidate.pchome_price"], - "preferred_current_source": "growth opportunity payload", - "future_read_only_fetch_gate": None, - "payload_mapping_probe": { - "goal": "Confirm whether missing price is a source payload gap or summary field mapping gap.", - "fetches_external_sites_in_preview": False, - "writes_database": False, - }, - }, - } - plan = dict(plans.get(field, {})) - if not plan: - return {} - plan["field"] = field - plan["status"] = "missing_in_current_payload" if missing_count else "covered_by_current_payload" - plan["missing_count"] = missing_count - return plan - - -def _build_fetch_gate_candidates(tasks: list[dict[str, Any]]) -> list[dict[str, Any]]: - candidates = [] - for task in tasks: - missing_fields = set(task.get("missing_fields") or []) - fetch_fields = [field for field in ("image", "availability") if field in missing_fields] - if not fetch_fields: - continue - candidates.append( - { - "pchome_product_id": task.get("pchome_product_id") or "", - "product_name": task.get("product_name") or "", - "product_url": task.get("product_url"), - "fields": fetch_fields, - "method": "GET", - "allowed_domain": "24h.pchome.com.tw", - "executes_fetch_in_preview": False, - "writes_database": False, - } - ) - return candidates[:PCHOME_FETCH_MAX_BATCH_SIZE] - - -def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]: - evidence = target.get("evidence_completeness") or {} - missing_fields = list(evidence.get("missing_fields") or []) - blocking_missing_fields = list(evidence.get("blocking_missing_fields") or []) - enrichment_steps = [ - { - "field": field, - "blocking": field in blocking_missing_fields, - "sources": _field_enrichment_sources(field), - } - for field in missing_fields - ] - ai_exception_required = _evidence_requires_ai_exception(evidence) - - return { - "lane": lane, - "pchome_product_id": target.get("pchome_product_id") or "", - "product_name": target.get("product_name") or "", - "product_url": target.get("product_url") or evidence.get("product_url"), - "sales_7d": target.get("sales_7d"), - "priority_score": target.get("priority_score"), - "action_code": target.get("action_code"), - "evidence_score": evidence.get("score"), - "present_fields": list(evidence.get("present_fields") or []), - "missing_fields": missing_fields, - "blocking_missing_fields": blocking_missing_fields, - "auto_accept_ready": bool(evidence.get("auto_accept_ready")), - **_ai_exception_compatibility_fields(ai_exception_required), - "unit_package_basis": evidence.get("unit_package_basis"), - "enrichment_steps": enrichment_steps, - } - - def _build_direct_mapping_search_terms(product_name: str, max_terms: int) -> list[str]: try: from services.momo_crawler import build_targeted_momo_search_terms @@ -4486,157 +4335,6 @@ def build_pchome_direct_mapping_retry_candidate_exception_controlled_apply_rollb } -def build_pchome_evidence_enrichment_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]: - """Build a read-only evidence enrichment package for mapping targets.""" - operator_preview = build_pchome_mapping_operator_preview(payload, batch_size=batch_size) - operator_batch = operator_preview.get("operator_batch") or {} - direct_targets = list(operator_batch.get("direct_mapping_targets") or []) - review_targets = list(operator_batch.get("review_candidate_targets") or []) - evidence_tasks = [ - *[_build_evidence_task(target, "direct_mapping") for target in direct_targets], - *[_build_evidence_task(target, "review_candidate") for target in review_targets], - ] - tasks_with_blockers = [ - task - for task in evidence_tasks - if task.get("blocking_missing_fields") - ] - missing_field_counts: dict[str, int] = {} - for task in evidence_tasks: - for field in task.get("missing_fields") or []: - missing_field_counts[field] = missing_field_counts.get(field, 0) + 1 - - if tasks_with_blockers: - result = "NEEDS_EVIDENCE_ENRICHMENT" - elif evidence_tasks: - result = "EVIDENCE_PREVIEW_READY" - else: - result = "NO_TARGETS" - - return { - "policy": EVIDENCE_ENRICHMENT_PREVIEW_POLICY, - "result": result, - "success": bool(operator_preview.get("success")), - "generated_at": operator_preview.get("generated_at"), - "source_policy": operator_preview.get("policy"), - "stats": operator_preview.get("stats") or {}, - "summary": { - "task_count": len(evidence_tasks), - "tasks_with_blockers": len(tasks_with_blockers), - "missing_field_counts": missing_field_counts, - "auto_accept_ready_count": sum(1 for task in evidence_tasks if task.get("auto_accept_ready")), - LEGACY_HUMAN_REVIEW_REQUIRED_COUNT_KEY: 0, - "primary_human_gate_count": 0, - "ai_exception_required_count": sum(1 for task in evidence_tasks if task.get("ai_exception_required")), - }, - "evidence_tasks": evidence_tasks, - "external_benchmark_alignment": operator_preview.get("external_benchmark_alignment") or {}, - "ai_automation_plan": { - "policy": "ollama_first_read_only_evidence_assist", - "llm_calls_in_preview": False, - "gemini_allowed": False, - "can_execute_write": False, - "recommended_next_ai_task": "Generate deterministic identity anchors and evidence gap summaries after image and availability sources are wired.", - "blocked_until": [ - "image evidence source is wired", - "availability evidence source is wired", - "unit/package parser preview is compared against production title samples", - ], - }, - "safety": { - "read_only_preview": True, - "fetches_external_sites": False, - "writes_database": False, - "executes_search": False, - "dispatches_telegram": False, - "llm_calls_in_preview": False, - "requires_operator_write_approval": True, - }, - "next_actions": [ - "Wire read-only image and availability enrichment before expanding auto-accept.", - "Validate deterministic unit/package basis parsing for bundle-sensitive items.", - "Keep DB writes behind the existing /api/ai/pchome-growth/backfill-momo-candidates write gate.", - ], - } - - -def build_pchome_evidence_source_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]: - """Build a read-only source wiring preview for missing product evidence fields.""" - enrichment_preview = build_pchome_evidence_enrichment_preview(payload, batch_size=batch_size) - tasks = list(enrichment_preview.get("evidence_tasks") or []) - fields = ["image", "availability", "price", "unit_price_or_package_basis"] - field_counts = {} - source_plans = {} - for field in fields: - missing_tasks = [task for task in tasks if field in (task.get("missing_fields") or [])] - field_counts[field] = { - "missing_count": len(missing_tasks), - "present_count": max(len(tasks) - len(missing_tasks), 0), - "sample_missing_targets": [ - { - "pchome_product_id": task.get("pchome_product_id") or "", - "product_name": task.get("product_name") or "", - "product_url": task.get("product_url"), - "lane": task.get("lane"), - } - for task in missing_tasks[:3] - ], - } - plan = _source_plan_for_field(field, len(missing_tasks)) - if plan: - source_plans[field] = plan - - fetch_gate_candidates = _build_fetch_gate_candidates(tasks) - if field_counts["image"]["missing_count"] or field_counts["availability"]["missing_count"]: - result = "NEEDS_SOURCE_WIRING" - elif field_counts["price"]["missing_count"]: - result = "NEEDS_PAYLOAD_MAPPING_PROBE" - elif tasks: - result = "SOURCE_PREVIEW_READY" - else: - result = "NO_TARGETS" - - return { - "policy": EVIDENCE_SOURCE_PREVIEW_POLICY, - "result": result, - "success": bool(enrichment_preview.get("success")), - "generated_at": enrichment_preview.get("generated_at"), - "source_policy": enrichment_preview.get("policy"), - "stats": enrichment_preview.get("stats") or {}, - "summary": { - "task_count": len(tasks), - "field_counts": field_counts, - "fetch_gate_candidate_count": len(fetch_gate_candidates), - }, - "source_plans": source_plans, - "fetch_gate_candidates": fetch_gate_candidates, - "external_benchmark_alignment": enrichment_preview.get("external_benchmark_alignment") or {}, - "ai_automation_plan": { - "policy": "ollama_first_read_only_source_wiring_assist", - "llm_calls_in_preview": False, - "gemini_allowed": False, - "can_execute_fetch": False, - "can_execute_write": False, - "recommended_next_ai_task": "Generate schema-aware parsers for image and availability after fetch-gate tests are accepted.", - }, - "safety": { - "read_only_preview": True, - "fetches_external_sites": False, - "writes_database": False, - "executes_search": False, - "dispatches_telegram": False, - "llm_calls_in_preview": False, - "requires_production_version_truth": True, - "requires_fetch_gate_implementation_before_external_get": True, - }, - "next_actions": [ - "Wire image source preview from existing payload keys before adding a controlled product-page fetch gate.", - "Wire availability source preview from existing payload keys before adding a controlled Offer availability parser.", - "Probe missing price rows through current payload mapping before any external fetch or write.", - ], - } - - def build_pchome_evidence_fetch_gate( payload: dict[str, Any], batch_size: int = 3, diff --git a/tests/test_pchome_mapping_backlog_modularization.py b/tests/test_pchome_mapping_backlog_modularization.py index 3671041..6bb6b61 100644 --- a/tests/test_pchome_mapping_backlog_modularization.py +++ b/tests/test_pchome_mapping_backlog_modularization.py @@ -7,6 +7,7 @@ from services.pchome_mapping_backlog import ( controlled_apply_executor, controlled_apply_receipt, controlled_apply_verifier, + enrichment, evidence, policies, reporter, @@ -201,7 +202,50 @@ def test_reporter_keeps_legacy_identity_and_read_only_preview(): assert preview["safety"]["writes_database"] is False -def test_legacy_facade_is_smaller_after_fifth_p0_extraction(): +def test_enrichment_keeps_legacy_identity_and_no_write_preview(): + assert ( + compatibility_facade.build_pchome_evidence_enrichment_preview + is enrichment.build_pchome_evidence_enrichment_preview + ) + assert ( + compatibility_facade.build_pchome_evidence_source_preview + is enrichment.build_pchome_evidence_source_preview + ) + assert ( + compatibility_facade._build_evidence_task + is enrichment._build_evidence_task + ) + assert ( + compatibility_facade._build_fetch_gate_candidates + is enrichment._build_fetch_gate_candidates + ) + assert ( + compatibility_facade._field_enrichment_sources + is enrichment._field_enrichment_sources + ) + assert ( + compatibility_facade._source_plan_for_field + is enrichment._source_plan_for_field + ) + + payload = { + "success": True, + "generated_at": "2026-07-11T15:30:00+08:00", + "stats": {}, + "opportunities": [], + } + preview = enrichment.build_pchome_evidence_enrichment_preview(payload) + source_preview = enrichment.build_pchome_evidence_source_preview(payload) + + assert preview["result"] == "NO_TARGETS" + assert source_preview["result"] == "NO_TARGETS" + assert preview["safety"]["fetches_external_sites"] is False + assert preview["safety"]["writes_database"] is False + assert source_preview["safety"]["fetches_external_sites"] is False + assert source_preview["safety"]["writes_database"] is False + + +def test_legacy_facade_is_smaller_after_sixth_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" - assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_200 + assert sum(1 for _ in facade.open(encoding="utf-8")) < 42_900