diff --git a/services/pchome_mapping_backlog_service.py b/services/pchome_mapping_backlog_service.py index 714f6b5..9f976ba 100644 --- a/services/pchome_mapping_backlog_service.py +++ b/services/pchome_mapping_backlog_service.py @@ -1324,19 +1324,25 @@ def _build_candidate_decision_id(candidate: dict[str, Any]) -> str: return f"pchome-direct-mapping-candidate-{digest[:16]}" +def _candidate_ready_for_no_write_receipt(candidate: dict[str, Any], min_score: float) -> bool: + confidence = _to_float(candidate.get("target_match_score")) + auto_compare_type = str(candidate.get("auto_compare_type") or "").strip() + return ( + bool(str(candidate.get("target_pchome_product_id") or "").strip()) + and bool(str(candidate.get("product_id") or "").strip()) + and confidence >= min_score + and auto_compare_type in {"total_price", "unit_price"} + and not _is_truthy_flag(candidate.get("target_hard_veto")) + ) + + def _build_candidate_decision_envelope(candidate: dict[str, Any], min_score: float) -> dict[str, Any]: confidence = _to_float(candidate.get("target_match_score")) auto_compare_type = str(candidate.get("auto_compare_type") or "").strip() hard_veto = _is_truthy_flag(candidate.get("target_hard_veto")) target_id = str(candidate.get("target_pchome_product_id") or "").strip() momo_product_id = str(candidate.get("product_id") or "").strip() - can_route_to_receipt = ( - bool(target_id) - and bool(momo_product_id) - and confidence >= min_score - and auto_compare_type in {"total_price", "unit_price"} - and not hard_veto - ) + can_route_to_receipt = _candidate_ready_for_no_write_receipt(candidate, min_score) failure_reasons = [] if not target_id: failure_reasons.append("missing_target_pchome_product_id") @@ -1453,12 +1459,12 @@ def build_pchome_direct_mapping_auto_search_package( auto_candidates = [ candidate for candidate in candidates - if candidate.get("auto_compare_type") in {"total_price", "unit_price"} + if _candidate_ready_for_no_write_receipt(candidate, min_score) ] review_candidates = [ candidate for candidate in candidates - if candidate.get("auto_compare_type") not in {"total_price", "unit_price"} + if not _candidate_ready_for_no_write_receipt(candidate, min_score) ] grouped_candidates = _search_candidates_by_target(candidates) for target in search_targets: @@ -2238,18 +2244,15 @@ def build_pchome_growth_ai_automation_readiness( *, execute_search: bool = False, execute_fetch: bool = False, + search_func: Any = None, ) -> dict[str, Any]: """Build a single read-only product-facing AI automation readiness view.""" mapping_summary = summarize_pchome_mapping_backlog(payload) - search_package = build_pchome_direct_mapping_auto_search_package( - payload, - batch_size=batch_size, - execute_search=execute_search, - ) decision_package = build_pchome_direct_mapping_candidate_decision_package( payload, batch_size=batch_size, execute_search=execute_search, + search_func=search_func, ) receipt_gate = build_pchome_auto_policy_receipt_gate( payload, @@ -2257,7 +2260,7 @@ def build_pchome_growth_ai_automation_readiness( execute_fetch=execute_fetch, ) backlog = mapping_summary.get("backlog") or {} - search_summary = search_package.get("summary") or {} + search_summary = decision_package.get("upstream_search_summary") or {} decision_summary = decision_package.get("summary") or {} receipt_summary = receipt_gate.get("summary") or {} @@ -2292,6 +2295,8 @@ def build_pchome_growth_ai_automation_readiness( if not direct_mapping_count and ready_receipt_count: result = "AI_AUTOMATION_READY_FOR_CONTROLLED_APPLY" + elif candidate_decision_count: + result = "AI_AUTOMATION_CANDIDATE_DECISIONS_READY" elif direct_mapping_count and selected_search_targets: result = "AI_AUTOMATION_ACTIVE_WAITING_FOR_CANDIDATES" elif receipt_count: diff --git a/tests/test_pchome_mapping_backlog_report.py b/tests/test_pchome_mapping_backlog_report.py index 8780eb8..bd87a53 100644 --- a/tests/test_pchome_mapping_backlog_report.py +++ b/tests/test_pchome_mapping_backlog_report.py @@ -290,6 +290,35 @@ def test_direct_mapping_auto_search_package_executes_fake_search_without_db_writ assert package["safety"]["persists_candidate"] is False +def test_direct_mapping_auto_search_package_does_not_count_hard_veto_as_auto_candidate(): + def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score): + return True, "found", [ + { + "product_id": "MOMO-UNIT", + "name": "Unit candidate with hard veto", + "price": 999, + "target_pchome_product_id": "PCH-2", + "target_match_score": 0.92, + "auto_compare_type": "unit_price", + "target_hard_veto": True, + } + ] + + package = build_pchome_direct_mapping_auto_search_package( + _payload(), + batch_size=1, + execute_search=True, + search_func=fake_search, + ) + + assert package["summary"]["candidates_found_count"] == 1 + assert package["summary"]["auto_compare_candidate_count"] == 0 + assert package["summary"]["review_candidate_count"] == 1 + assert package["candidate_preview"][0]["auto_compare_type"] == "unit_price" + assert package["candidate_preview"][0]["target_hard_veto"] is True + assert package["safety"]["writes_database"] is False + + def test_direct_mapping_candidate_decision_package_waits_for_search_candidates_without_db_write(): package = build_pchome_direct_mapping_candidate_decision_package(_payload(), batch_size=1) @@ -398,6 +427,43 @@ def test_ai_automation_readiness_makes_automation_visible_without_manual_primary assert readiness["safety"]["llm_calls_in_preview"] is False +def test_ai_automation_readiness_reports_candidate_decisions_after_controlled_search(): + call_count = {"search": 0} + + def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score): + call_count["search"] += 1 + return True, "found", [ + { + "product_id": "MOMO-1", + "name": "Direct mapping product 40ml x2", + "price": 999, + "target_pchome_product_id": "PCH-2", + "target_match_score": 0.92, + "auto_compare_type": "total_price", + "target_hard_veto": False, + } + ] + + readiness = build_pchome_growth_ai_automation_readiness( + _payload(), + batch_size=1, + execute_search=True, + search_func=fake_search, + ) + + lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]} + assert readiness["result"] == "AI_AUTOMATION_CANDIDATE_DECISIONS_READY" + assert readiness["summary"]["candidate_decision_count"] == 1 + assert readiness["summary"]["waiting_candidate_count"] == 0 + assert readiness["summary"]["auto_compare_decision_count"] == 1 + assert readiness["summary"]["machine_review_decision_count"] == 0 + assert readiness["summary"]["external_network_execute_count"] == 1 + assert lanes["candidate_decision_package"]["status"] == "ready" + assert readiness["safety"]["executes_search"] is True + assert readiness["safety"]["writes_database"] is False + assert call_count["search"] == 1 + + def test_unit_package_basis_parser_extracts_quantity_count_and_risk_signals(): single = parse_unit_package_basis("雅詩蘭黛 粉持久完美持妝粉底 30ml") assert single["package_basis"] == "single_unit_quantity_candidate"