diff --git a/config.py b/config.py index 9270bbe..de8e0f9 100644 --- a/config.py +++ b/config.py @@ -402,7 +402,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.783" +SYSTEM_VERSION = "V10.784" 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 d0cbc18..390409c 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.783 +> **適用版本**: V10.784 --- diff --git a/docs/memory/code_modularization_inventory_20260430.md b/docs/memory/code_modularization_inventory_20260430.md index 4baa340..f5c3ea6 100644 --- a/docs/memory/code_modularization_inventory_20260430.md +++ b/docs/memory/code_modularization_inventory_20260430.md @@ -129,7 +129,7 @@ | 行數 | 檔案 | 分類 | 拆分方向 | |---:|---|---|---| -| 43543 | `services/pchome_mapping_backlog_service.py` | P0 gate 巨型 service、拆分進行中 | policy、AI exception contract、evidence parser、artifact guard、bounded executor、drift verifier 與 receipt replay 已移到 `services/pchome_mapping_backlog/`；下一步拆 reporter | +| 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 | | 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/reporter.py b/services/pchome_mapping_backlog/reporter.py new file mode 100644 index 0000000..49d7291 --- /dev/null +++ b/services/pchome_mapping_backlog/reporter.py @@ -0,0 +1,451 @@ +"""Read-only PChome backlog reporting and operator preview projections.""" + +from __future__ import annotations + +from typing import Any + +from services.pchome_mapping_backlog.contracts import _ai_exception_compatibility_fields +from services.pchome_mapping_backlog.evidence import ( + _action_code, + _action_label, + _first_present, + _pchome_product_url, + _to_float, + parse_unit_package_basis, +) +from services.pchome_mapping_backlog.policies import ( + BACKLOG_POLICY, + EXTERNAL_BENCHMARK_REFERENCES, + OPERATOR_PREVIEW_POLICY, +) + + +def _evidence_completeness( + item: dict[str, Any], + review_candidate: dict[str, Any], + external_price: dict[str, Any], +) -> dict[str, Any]: + product_id = str(item.get("pchome_product_id") or "").strip() + product_name = str(item.get("product_name") or "").strip() + product_url = _first_present( + item.get("product_url"), + item.get("pchome_url"), + _pchome_product_url(product_id), + ) + pchome_price = _first_present( + item.get("pchome_price"), + external_price.get("pchome_price"), + review_candidate.get("pchome_price"), + ) + image_url = _first_present( + item.get("image_url"), + item.get("image"), + item.get("product_image_url"), + ) + availability = _first_present( + item.get("availability"), + item.get("stock_status"), + item.get("is_available"), + ) + unit_package_basis = parse_unit_package_basis(product_name) + parsed_unit_basis = ( + unit_package_basis + if unit_package_basis.get("package_basis") != "insufficient" + else None + ) + unit_basis = _first_present( + external_price.get("price_basis"), + item.get("price_basis"), + item.get("unit_label"), + parsed_unit_basis, + ) + unit_review_required = bool(unit_package_basis.get("risk_signals")) + + checks = [ + ("stable_product_id", bool(product_id), "required"), + ("product_name", bool(product_name), "required"), + ("product_url", bool(product_url), "required"), + ("price", pchome_price is not None, "required"), + ("image", bool(image_url), "strongly_recommended"), + ("availability", availability is not None, "strongly_recommended"), + ( + "unit_price_or_package_basis", + bool(unit_basis), + "required_when_bundle_or_unit_sensitive", + ), + ] + present = [field for field, ok, _requirement in checks if ok] + missing = [field for field, ok, _requirement in checks if not ok] + blocking_missing = [ + field + for field, ok, requirement in checks + if not ok and requirement in {"required", "strongly_recommended"} + ] + score = round(len(present) / max(len(checks), 1) * 100, 1) + + ai_exception_required = ( + bool(blocking_missing) + or bool(review_candidate) + or not external_price + or unit_review_required + ) + + return { + "score": score, + "present_fields": present, + "missing_fields": missing, + "blocking_missing_fields": blocking_missing, + "auto_accept_ready": ( + not blocking_missing and bool(external_price) and not unit_review_required + ), + **_ai_exception_compatibility_fields(ai_exception_required), + "product_url": product_url, + "image_url": image_url, + "availability": availability, + "unit_package_basis": unit_package_basis, + } + + +def compact_mapping_item(item: dict[str, Any]) -> dict[str, Any]: + review_candidate = item.get("review_candidate") or {} + external_price = item.get("external_price") or {} + product_id = str(item.get("pchome_product_id") or "") + product_url = _first_present( + item.get("product_url"), + item.get("pchome_url"), + _pchome_product_url(product_id), + ) + return { + "pchome_product_id": product_id, + "product_url": product_url, + "product_name": item.get("product_name") or "", + "sales_7d": round(_to_float(item.get("sales_7d")), 2), + "sales_delta_pct": item.get("sales_delta_pct"), + "priority_score": item.get("priority_score"), + "pchome_price": item.get("pchome_price"), + "action_code": _action_code(item), + "action_label": _action_label(item), + "review_candidate": { + "id": review_candidate.get("id"), + "momo_sku": review_candidate.get("momo_sku"), + "momo_name": review_candidate.get("momo_name"), + "quality_score": review_candidate.get("quality_score"), + "gap_pct": review_candidate.get("gap_pct"), + } + if review_candidate + else None, + "external_price": { + "momo_sku": external_price.get("momo_sku"), + "momo_name": external_price.get("momo_name"), + "price_basis": external_price.get("price_basis"), + "gap_pct": external_price.get("gap_pct"), + "data_source_label": external_price.get("data_source_label"), + "updated_at": external_price.get("updated_at"), + } + if external_price + else None, + "evidence_completeness": _evidence_completeness( + item, + review_candidate, + external_price, + ), + "reason_lines": list(item.get("reason_lines") or [])[:3], + } + + +def _build_external_benchmark_alignment() -> dict[str, Any]: + return { + "references": EXTERNAL_BENCHMARK_REFERENCES, + "required_evidence_fields": [ + { + "field": "stable_product_id", + "current_payload": "pchome_product_id", + "status": "present", + "why": "Stable IDs preserve mapping history and make post-run readback comparable.", + }, + { + "field": "product_name", + "current_payload": "product_name", + "status": "present", + "why": "Exact title/name matching is the first identity anchor for operator review.", + }, + { + "field": "product_url", + "current_payload": "derived_from_pchome_product_id", + "status": "present_for_pchome", + "why": "Operators need a direct product page path for visual confirmation.", + }, + { + "field": "price", + "current_payload": "pchome_price/external_price", + "status": "partial", + "why": "Offer price and currency are required before a candidate can become decision-ready.", + }, + { + "field": "image", + "current_payload": None, + "status": "missing_in_current_growth_payload", + "why": "Image evidence should be added before high-volume auto-accept expansion.", + }, + { + "field": "availability", + "current_payload": None, + "status": "missing_in_current_growth_payload", + "why": "Availability prevents matching stale or non-purchasable offers.", + }, + { + "field": "unit_price_or_package_basis", + "current_payload": ( + "external_price.price_basis or deterministic title parser preview" + ), + "status": "parser_preview_available", + "why": "Unit price and package basis protect bundles, variants, and volume-size comparisons.", + }, + ], + "operator_review_principles": [ + "Separate direct mapping, review candidate, and already comparable items.", + "Do not auto-accept variants, colors, bundles, or catalog offers without explicit evidence.", + "Keep search/query support exact-title friendly so copied product names and model terms remain useful.", + ], + } + + +def _build_ai_automation_plan( + selected_direct: list[dict[str, Any]], + selected_review: list[dict[str, Any]], +) -> dict[str, Any]: + return { + "policy": "ollama_first_read_only_ai_assist", + "llm_calls_in_preview": False, + "gemini_allowed": False, + "provider_order": [ + "GCP-A 34.87.90.216:11434", + "GCP-B 34.21.145.224:11434", + "111 192.168.0.111:11434", + ], + "automation_readiness": { + "direct_mapping_targets": len(selected_direct), + "review_candidate_targets": len(selected_review), + "can_generate_operator_summary": bool(selected_direct or selected_review), + "can_execute_write": False, + }, + "steps": [ + { + "name": "identity_anchor_extraction", + "mode": "deterministic_first_ollama_assist_later", + "writes_database": False, + "output": "brand/product_line/spec/package/variant anchors for each selected target", + }, + { + "name": "candidate_search_plan", + "mode": "rule_based_query_pack", + "writes_database": False, + "output": "exact title, brand plus product line, and spec-preserving search terms", + }, + { + "name": "operator_decision_summary", + "mode": "ollama_first_after_write_gate_only", + "writes_database": False, + "output": "plain-language review reason, evidence gaps, and post-write readback checklist", + }, + { + "name": "post_write_readback", + "mode": "deterministic_metrics", + "writes_database": False, + "output": "mapping_rate, direct_mapping_count, review_candidate_count, mapped_count delta", + }, + ], + "ai_exception_required_for": [ + "missing image or availability evidence", + "variant/color/fragrance/shade/package ambiguity", + "unit-price or bundle-sensitive comparisons", + "any candidate not meeting exact identity evidence", + ], + } + + +def summarize_pchome_mapping_backlog(payload: dict[str, Any]) -> dict[str, Any]: + stats = payload.get("stats") or {} + opportunities = [ + item + for item in payload.get("opportunities") or [] + if isinstance(item, dict) + ] + needs_mapping = [item for item in opportunities if not item.get("external_price")] + review_candidates = [ + item for item in needs_mapping if item.get("review_candidate") + ] + direct_mapping = [ + item + for item in needs_mapping + if _action_code(item) == "map_external_product" + and not item.get("review_candidate") + ] + mapped = [item for item in opportunities if item.get("external_price")] + + action_counts: dict[str, int] = {} + sales_by_action: dict[str, float] = {} + for item in opportunities: + label = _action_label(item) + action_counts[label] = action_counts.get(label, 0) + 1 + sales_by_action[label] = round( + sales_by_action.get(label, 0.0) + _to_float(item.get("sales_7d")), + 2, + ) + + candidate_count = int(stats.get("candidate_count") or len(opportunities)) + mapped_count = int(stats.get("mapped_count") or len(mapped)) + needs_mapping_count = int(stats.get("needs_mapping_count") or len(needs_mapping)) + mapping_rate = stats.get("mapping_rate") + if mapping_rate is None: + mapping_rate = round(mapped_count / max(candidate_count, 1) * 100, 1) + + top_needs_mapping = sorted( + needs_mapping, + key=lambda item: ( + _to_float(item.get("sales_7d")), + _to_float(item.get("priority_score")), + ), + reverse=True, + )[:10] + top_review_candidates = sorted( + review_candidates, + key=lambda item: _to_float( + (item.get("review_candidate") or {}).get("quality_score") + ), + reverse=True, + )[:10] + + if not payload.get("success", False): + result = "BLOCKED" + elif needs_mapping_count > 0: + result = "NEEDS_MAPPING" + else: + result = "PASS" + + return { + "policy": BACKLOG_POLICY, + "result": result, + "success": bool(payload.get("success")), + "generated_at": payload.get("generated_at"), + "cache_state": payload.get("cache_state"), + "system_name": payload.get("system_name"), + "message": payload.get("message"), + "stats": { + "candidate_count": candidate_count, + "mapped_count": mapped_count, + "mapping_rate": mapping_rate, + "needs_mapping_count": needs_mapping_count, + "review_candidate_count": int( + stats.get("review_candidate_count") or len(review_candidates) + ), + "latest_sales_date": stats.get("latest_sales_date"), + "overall_latest_sales_date": stats.get("overall_latest_sales_date"), + "overall_sales_7d": stats.get("overall_sales_7d"), + "opportunity_sales_7d": stats.get("opportunity_sales_7d"), + "action_counts": dict(stats.get("action_counts") or action_counts), + "action_code_counts": dict(stats.get("action_code_counts") or {}), + "external_data_source_counts": dict( + stats.get("external_data_source_counts") or {} + ), + }, + "backlog": { + "direct_mapping_count": len(direct_mapping), + "review_candidate_count": len(review_candidates), + "mapped_opportunity_count": len(mapped), + "sales_by_action": sales_by_action, + "top_needs_mapping": [ + compact_mapping_item(item) for item in top_needs_mapping + ], + "top_review_candidates": [ + compact_mapping_item(item) for item in top_review_candidates + ], + }, + "next_actions": [ + "Run the production version truth guard before changing or deploying.", + "Handle direct mapping items first; they have no verified external price yet.", + "Review candidate items next; they already have MOMO candidates but need same-item confirmation.", + "Keep this report read-only until an explicit DB-write operator run is approved.", + ], + } + + +def build_pchome_mapping_operator_preview( + payload: dict[str, Any], + batch_size: int = 5, +) -> dict[str, Any]: + """Build a read-only operator run package for the direct mapping backlog.""" + summary = summarize_pchome_mapping_backlog(payload) + backlog = summary.get("backlog") or {} + direct_items = [ + item + for item in backlog.get("top_needs_mapping") or [] + if item.get("action_code") == "map_external_product" + ] + review_items = list(backlog.get("top_review_candidates") or []) + batch_size = max(1, min(int(batch_size or 5), 8)) + selected_direct = direct_items[:batch_size] + selected_review = review_items[:batch_size] + + if selected_direct: + result = "READY_FOR_OPERATOR_PREVIEW" + elif selected_review: + result = "REVIEW_CANDIDATES_ONLY" + else: + result = "NO_DIRECT_MAPPING_TARGETS" + + return { + "policy": OPERATOR_PREVIEW_POLICY, + "result": result, + "success": bool(summary.get("success")), + "generated_at": summary.get("generated_at"), + "stats": summary.get("stats") or {}, + "backlog": { + "direct_mapping_count": int(backlog.get("direct_mapping_count") or 0), + "review_candidate_count": int( + backlog.get("review_candidate_count") or 0 + ), + "mapped_opportunity_count": int( + backlog.get("mapped_opportunity_count") or 0 + ), + }, + "operator_batch": { + "batch_size": batch_size, + "selected_direct_mapping_count": len(selected_direct), + "selected_review_candidate_count": len(selected_review), + "direct_mapping_targets": selected_direct, + "review_candidate_targets": selected_review, + }, + "command_preview": { + "method": "POST", + "endpoint": "/api/ai/pchome-growth/backfill-momo-candidates", + "payload": {"limit": min(batch_size, 8)}, + "executes_search": True, + "writes_database": True, + "write_gate_required": True, + }, + "external_benchmark_alignment": _build_external_benchmark_alignment(), + "ai_automation_plan": _build_ai_automation_plan( + selected_direct, + selected_review, + ), + "safety": { + "read_only_preview": True, + "executes_search": False, + "writes_database": False, + "dispatches_telegram": False, + "requires_production_version_truth": True, + "requires_operator_write_approval": True, + }, + "required_before_execute": [ + "Run production version truth guard and keep production /health as latest truth.", + "Confirm the selected direct mapping targets are the intended PChome products.", + "Confirm DB-write authorization for /api/ai/pchome-growth/backfill-momo-candidates.", + "Run post-write mapping backlog readback and compare direct_mapping_count / mapped_count.", + ], + "acceptance_criteria": [ + "direct_mapping_count decreases, or review_candidate_count increases with named MOMO candidates.", + "mapped_count or mapping_rate increases only when a verified external price is written.", + "No Gemini, Telegram dispatch, scheduler mutation, or unrelated DB write is part of this run.", + ], + } diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index b80dc4a..116953e 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -162,220 +162,14 @@ from services.pchome_mapping_backlog.evidence import ( parse_pchome_product_page_evidence_html, parse_unit_package_basis, ) - - - -def _evidence_completeness(item: dict[str, Any], review_candidate: dict[str, Any], external_price: dict[str, Any]) -> dict[str, Any]: - product_id = str(item.get("pchome_product_id") or "").strip() - product_name = str(item.get("product_name") or "").strip() - product_url = _first_present(item.get("product_url"), item.get("pchome_url"), _pchome_product_url(product_id)) - pchome_price = _first_present( - item.get("pchome_price"), - external_price.get("pchome_price"), - review_candidate.get("pchome_price"), - ) - image_url = _first_present(item.get("image_url"), item.get("image"), item.get("product_image_url")) - availability = _first_present(item.get("availability"), item.get("stock_status"), item.get("is_available")) - unit_package_basis = parse_unit_package_basis(product_name) - parsed_unit_basis = ( - unit_package_basis - if unit_package_basis.get("package_basis") != "insufficient" - else None - ) - unit_basis = _first_present( - external_price.get("price_basis"), - item.get("price_basis"), - item.get("unit_label"), - parsed_unit_basis, - ) - unit_review_required = bool(unit_package_basis.get("risk_signals")) - - checks = [ - ("stable_product_id", bool(product_id), "required"), - ("product_name", bool(product_name), "required"), - ("product_url", bool(product_url), "required"), - ("price", pchome_price is not None, "required"), - ("image", bool(image_url), "strongly_recommended"), - ("availability", availability is not None, "strongly_recommended"), - ( - "unit_price_or_package_basis", - bool(unit_basis), - "required_when_bundle_or_unit_sensitive", - ), - ] - present = [field for field, ok, _requirement in checks if ok] - missing = [field for field, ok, _requirement in checks if not ok] - blocking_missing = [ - field - for field, ok, requirement in checks - if not ok and requirement in {"required", "strongly_recommended"} - ] - score = round(len(present) / max(len(checks), 1) * 100, 1) - - ai_exception_required = ( - bool(blocking_missing) - or bool(review_candidate) - or not external_price - or unit_review_required - ) - - return { - "score": score, - "present_fields": present, - "missing_fields": missing, - "blocking_missing_fields": blocking_missing, - "auto_accept_ready": not blocking_missing and bool(external_price) and not unit_review_required, - **_ai_exception_compatibility_fields(ai_exception_required), - "product_url": product_url, - "image_url": image_url, - "availability": availability, - "unit_package_basis": unit_package_basis, - } - - -def compact_mapping_item(item: dict[str, Any]) -> dict[str, Any]: - review_candidate = item.get("review_candidate") or {} - external_price = item.get("external_price") or {} - product_id = str(item.get("pchome_product_id") or "") - product_url = _first_present(item.get("product_url"), item.get("pchome_url"), _pchome_product_url(product_id)) - return { - "pchome_product_id": product_id, - "product_url": product_url, - "product_name": item.get("product_name") or "", - "sales_7d": round(_to_float(item.get("sales_7d")), 2), - "sales_delta_pct": item.get("sales_delta_pct"), - "priority_score": item.get("priority_score"), - "pchome_price": item.get("pchome_price"), - "action_code": _action_code(item), - "action_label": _action_label(item), - "review_candidate": { - "id": review_candidate.get("id"), - "momo_sku": review_candidate.get("momo_sku"), - "momo_name": review_candidate.get("momo_name"), - "quality_score": review_candidate.get("quality_score"), - "gap_pct": review_candidate.get("gap_pct"), - } - if review_candidate - else None, - "external_price": { - "momo_sku": external_price.get("momo_sku"), - "momo_name": external_price.get("momo_name"), - "price_basis": external_price.get("price_basis"), - "gap_pct": external_price.get("gap_pct"), - "data_source_label": external_price.get("data_source_label"), - "updated_at": external_price.get("updated_at"), - } - if external_price - else None, - "evidence_completeness": _evidence_completeness(item, review_candidate, external_price), - "reason_lines": list(item.get("reason_lines") or [])[:3], - } - - -def _build_external_benchmark_alignment() -> dict[str, Any]: - return { - "references": EXTERNAL_BENCHMARK_REFERENCES, - "required_evidence_fields": [ - { - "field": "stable_product_id", - "current_payload": "pchome_product_id", - "status": "present", - "why": "Stable IDs preserve mapping history and make post-run readback comparable.", - }, - { - "field": "product_name", - "current_payload": "product_name", - "status": "present", - "why": "Exact title/name matching is the first identity anchor for operator review.", - }, - { - "field": "product_url", - "current_payload": "derived_from_pchome_product_id", - "status": "present_for_pchome", - "why": "Operators need a direct product page path for visual confirmation.", - }, - { - "field": "price", - "current_payload": "pchome_price/external_price", - "status": "partial", - "why": "Offer price and currency are required before a candidate can become decision-ready.", - }, - { - "field": "image", - "current_payload": None, - "status": "missing_in_current_growth_payload", - "why": "Image evidence should be added before high-volume auto-accept expansion.", - }, - { - "field": "availability", - "current_payload": None, - "status": "missing_in_current_growth_payload", - "why": "Availability prevents matching stale or non-purchasable offers.", - }, - { - "field": "unit_price_or_package_basis", - "current_payload": "external_price.price_basis or deterministic title parser preview", - "status": "parser_preview_available", - "why": "Unit price and package basis protect bundles, variants, and volume-size comparisons.", - }, - ], - "operator_review_principles": [ - "Separate direct mapping, review candidate, and already comparable items.", - "Do not auto-accept variants, colors, bundles, or catalog offers without explicit evidence.", - "Keep search/query support exact-title friendly so copied product names and model terms remain useful.", - ], - } - - -def _build_ai_automation_plan(selected_direct: list[dict[str, Any]], selected_review: list[dict[str, Any]]) -> dict[str, Any]: - return { - "policy": "ollama_first_read_only_ai_assist", - "llm_calls_in_preview": False, - "gemini_allowed": False, - "provider_order": [ - "GCP-A 34.87.90.216:11434", - "GCP-B 34.21.145.224:11434", - "111 192.168.0.111:11434", - ], - "automation_readiness": { - "direct_mapping_targets": len(selected_direct), - "review_candidate_targets": len(selected_review), - "can_generate_operator_summary": bool(selected_direct or selected_review), - "can_execute_write": False, - }, - "steps": [ - { - "name": "identity_anchor_extraction", - "mode": "deterministic_first_ollama_assist_later", - "writes_database": False, - "output": "brand/product_line/spec/package/variant anchors for each selected target", - }, - { - "name": "candidate_search_plan", - "mode": "rule_based_query_pack", - "writes_database": False, - "output": "exact title, brand plus product line, and spec-preserving search terms", - }, - { - "name": "operator_decision_summary", - "mode": "ollama_first_after_write_gate_only", - "writes_database": False, - "output": "plain-language review reason, evidence gaps, and post-write readback checklist", - }, - { - "name": "post_write_readback", - "mode": "deterministic_metrics", - "writes_database": False, - "output": "mapping_rate, direct_mapping_count, review_candidate_count, mapped_count delta", - }, - ], - "ai_exception_required_for": [ - "missing image or availability evidence", - "variant/color/fragrance/shade/package ambiguity", - "unit-price or bundle-sensitive comparisons", - "any candidate not meeting exact identity evidence", - ], - } +from services.pchome_mapping_backlog.reporter import ( + _build_ai_automation_plan, + _build_external_benchmark_alignment, + _evidence_completeness, + build_pchome_mapping_operator_preview, + compact_mapping_item, + summarize_pchome_mapping_backlog, +) def _field_enrichment_sources(field: str) -> list[dict[str, Any]]: @@ -537,160 +331,6 @@ def _build_evidence_task(target: dict[str, Any], lane: str) -> dict[str, Any]: } -def summarize_pchome_mapping_backlog(payload: dict[str, Any]) -> dict[str, Any]: - stats = payload.get("stats") or {} - opportunities = [item for item in payload.get("opportunities") or [] if isinstance(item, dict)] - needs_mapping = [item for item in opportunities if not item.get("external_price")] - review_candidates = [item for item in needs_mapping if item.get("review_candidate")] - direct_mapping = [ - item - for item in needs_mapping - if _action_code(item) == "map_external_product" and not item.get("review_candidate") - ] - mapped = [item for item in opportunities if item.get("external_price")] - - action_counts: dict[str, int] = {} - sales_by_action: dict[str, float] = {} - for item in opportunities: - label = _action_label(item) - action_counts[label] = action_counts.get(label, 0) + 1 - sales_by_action[label] = round(sales_by_action.get(label, 0.0) + _to_float(item.get("sales_7d")), 2) - - candidate_count = int(stats.get("candidate_count") or len(opportunities)) - mapped_count = int(stats.get("mapped_count") or len(mapped)) - needs_mapping_count = int(stats.get("needs_mapping_count") or len(needs_mapping)) - mapping_rate = stats.get("mapping_rate") - if mapping_rate is None: - mapping_rate = round(mapped_count / max(candidate_count, 1) * 100, 1) - - top_needs_mapping = sorted( - needs_mapping, - key=lambda item: (_to_float(item.get("sales_7d")), _to_float(item.get("priority_score"))), - reverse=True, - )[:10] - top_review_candidates = sorted( - review_candidates, - key=lambda item: _to_float((item.get("review_candidate") or {}).get("quality_score")), - reverse=True, - )[:10] - - if not payload.get("success", False): - result = "BLOCKED" - elif needs_mapping_count > 0: - result = "NEEDS_MAPPING" - else: - result = "PASS" - - return { - "policy": BACKLOG_POLICY, - "result": result, - "success": bool(payload.get("success")), - "generated_at": payload.get("generated_at"), - "cache_state": payload.get("cache_state"), - "system_name": payload.get("system_name"), - "message": payload.get("message"), - "stats": { - "candidate_count": candidate_count, - "mapped_count": mapped_count, - "mapping_rate": mapping_rate, - "needs_mapping_count": needs_mapping_count, - "review_candidate_count": int(stats.get("review_candidate_count") or len(review_candidates)), - "latest_sales_date": stats.get("latest_sales_date"), - "overall_latest_sales_date": stats.get("overall_latest_sales_date"), - "overall_sales_7d": stats.get("overall_sales_7d"), - "opportunity_sales_7d": stats.get("opportunity_sales_7d"), - "action_counts": dict(stats.get("action_counts") or action_counts), - "action_code_counts": dict(stats.get("action_code_counts") or {}), - "external_data_source_counts": dict(stats.get("external_data_source_counts") or {}), - }, - "backlog": { - "direct_mapping_count": len(direct_mapping), - "review_candidate_count": len(review_candidates), - "mapped_opportunity_count": len(mapped), - "sales_by_action": sales_by_action, - "top_needs_mapping": [compact_mapping_item(item) for item in top_needs_mapping], - "top_review_candidates": [compact_mapping_item(item) for item in top_review_candidates], - }, - "next_actions": [ - "Run the production version truth guard before changing or deploying.", - "Handle direct mapping items first; they have no verified external price yet.", - "Review candidate items next; they already have MOMO candidates but need same-item confirmation.", - "Keep this report read-only until an explicit DB-write operator run is approved.", - ], - } - - -def build_pchome_mapping_operator_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]: - """Build a read-only operator run package for the direct mapping backlog.""" - summary = summarize_pchome_mapping_backlog(payload) - backlog = summary.get("backlog") or {} - direct_items = [ - item - for item in backlog.get("top_needs_mapping") or [] - if item.get("action_code") == "map_external_product" - ] - review_items = list(backlog.get("top_review_candidates") or []) - batch_size = max(1, min(int(batch_size or 5), 8)) - selected_direct = direct_items[:batch_size] - selected_review = review_items[:batch_size] - - if selected_direct: - result = "READY_FOR_OPERATOR_PREVIEW" - elif selected_review: - result = "REVIEW_CANDIDATES_ONLY" - else: - result = "NO_DIRECT_MAPPING_TARGETS" - - return { - "policy": OPERATOR_PREVIEW_POLICY, - "result": result, - "success": bool(summary.get("success")), - "generated_at": summary.get("generated_at"), - "stats": summary.get("stats") or {}, - "backlog": { - "direct_mapping_count": int(backlog.get("direct_mapping_count") or 0), - "review_candidate_count": int(backlog.get("review_candidate_count") or 0), - "mapped_opportunity_count": int(backlog.get("mapped_opportunity_count") or 0), - }, - "operator_batch": { - "batch_size": batch_size, - "selected_direct_mapping_count": len(selected_direct), - "selected_review_candidate_count": len(selected_review), - "direct_mapping_targets": selected_direct, - "review_candidate_targets": selected_review, - }, - "command_preview": { - "method": "POST", - "endpoint": "/api/ai/pchome-growth/backfill-momo-candidates", - "payload": {"limit": min(batch_size, 8)}, - "executes_search": True, - "writes_database": True, - "write_gate_required": True, - }, - "external_benchmark_alignment": _build_external_benchmark_alignment(), - "ai_automation_plan": _build_ai_automation_plan(selected_direct, selected_review), - "safety": { - "read_only_preview": True, - "executes_search": False, - "writes_database": False, - "dispatches_telegram": False, - "requires_production_version_truth": True, - "requires_operator_write_approval": True, - }, - "required_before_execute": [ - "Run production version truth guard and keep production /health as latest truth.", - "Confirm the selected direct mapping targets are the intended PChome products.", - "Confirm DB-write authorization for /api/ai/pchome-growth/backfill-momo-candidates.", - "Run post-write mapping backlog readback and compare direct_mapping_count / mapped_count.", - ], - "acceptance_criteria": [ - "direct_mapping_count decreases, or review_candidate_count increases with named MOMO candidates.", - "mapped_count or mapping_rate increases only when a verified external price is written.", - "No Gemini, Telegram dispatch, scheduler mutation, or unrelated DB write is part of this run.", - ], - } - - 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 diff --git a/tests/test_pchome_mapping_backlog_modularization.py b/tests/test_pchome_mapping_backlog_modularization.py index 6d7b4d2..3671041 100644 --- a/tests/test_pchome_mapping_backlog_modularization.py +++ b/tests/test_pchome_mapping_backlog_modularization.py @@ -9,6 +9,7 @@ from services.pchome_mapping_backlog import ( controlled_apply_verifier, evidence, policies, + reporter, ) @@ -161,7 +162,46 @@ def test_controlled_apply_receipt_replay_keeps_identity_and_no_write_default(tmp assert len(result["executor_receipt_artifact"]["payload_sha256"]) == 64 -def test_legacy_facade_is_smaller_after_fourth_p0_extraction(): +def test_reporter_keeps_legacy_identity_and_read_only_preview(): + assert ( + compatibility_facade.summarize_pchome_mapping_backlog + is reporter.summarize_pchome_mapping_backlog + ) + assert ( + compatibility_facade.build_pchome_mapping_operator_preview + is reporter.build_pchome_mapping_operator_preview + ) + assert compatibility_facade.compact_mapping_item is reporter.compact_mapping_item + assert ( + compatibility_facade._evidence_completeness + is reporter._evidence_completeness + ) + assert ( + compatibility_facade._build_external_benchmark_alignment + is reporter._build_external_benchmark_alignment + ) + assert ( + compatibility_facade._build_ai_automation_plan + is reporter._build_ai_automation_plan + ) + + payload = { + "success": True, + "generated_at": "2026-07-11T14:00:00+08:00", + "stats": {}, + "opportunities": [], + } + summary = reporter.summarize_pchome_mapping_backlog(payload) + preview = reporter.build_pchome_mapping_operator_preview(payload) + + assert summary["result"] == "PASS" + assert preview["result"] == "NO_DIRECT_MAPPING_TARGETS" + assert preview["safety"]["read_only_preview"] is True + assert preview["safety"]["executes_search"] is False + assert preview["safety"]["writes_database"] is False + + +def test_legacy_facade_is_smaller_after_fifth_p0_extraction(): facade = ROOT / "services/pchome_mapping_backlog_service.py" - assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_600 + assert sum(1 for _ in facade.open(encoding="utf-8")) < 43_200