From c6350791b03a9d953350dc9be575f844b54299f6 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 21:15:22 +0800 Subject: [PATCH] fix(growth): verify committed offers by exact identity --- config.py | 2 +- .../pchome_growth_momo_backfill_service.py | 17 +- .../pchome_growth_same_item_reconciliation.py | 168 +++++++++++++++++- tests/test_pchome_revenue_growth_service.py | 18 ++ tests/test_pchome_same_item_reconciliation.py | 84 +++++++++ 5 files changed, 284 insertions(+), 5 deletions(-) diff --git a/config.py b/config.py index d0d72cb..c89107d 100644 --- a/config.py +++ b/config.py @@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.791" +SYSTEM_VERSION = "V10.792" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/services/pchome_growth_momo_backfill_service.py b/services/pchome_growth_momo_backfill_service.py index 227a7e8..3289041 100644 --- a/services/pchome_growth_momo_backfill_service.py +++ b/services/pchome_growth_momo_backfill_service.py @@ -15,6 +15,7 @@ from services.pchome_growth_same_item_reconciliation import ( ensure_evidence_receipt_table, new_reconciliation_identity, persist_reconciliation_receipts, + readback_external_offer_candidates, verify_same_item_candidates, ) @@ -101,6 +102,7 @@ def run_pchome_growth_momo_backfill( search_func: Callable[[list[dict[str, Any]], int], tuple[bool, str, list[dict[str, Any]]]] | None = None, sync_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None, sync_review_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None, + readback_func: Callable[[Any, list[dict[str, Any]]], dict[str, Any]] | None = None, ) -> dict[str, Any]: """Run a bounded search, independent verifier, apply, and readback loop.""" limit = max(1, min(int(limit or 12), 20)) @@ -108,6 +110,7 @@ def run_pchome_growth_momo_backfill( search_candidates = search_func or _default_search_candidates sync_candidates = sync_func or _default_sync_candidates sync_review_candidates = sync_review_func or _default_sync_review_candidates + readback_candidates = readback_func or readback_external_offer_candidates identity = new_reconciliation_identity() generated_at = datetime.now(timezone.utc).isoformat() @@ -166,6 +169,11 @@ def run_pchome_growth_momo_backfill( } if auto_candidates: external_offer_sync = sync_candidates(engine, auto_candidates) + external_offer_sync["independent_readback_required"] = True + external_offer_sync["independent_readback"] = readback_candidates( + engine, + auto_candidates, + ) review_candidate_sync = { "success": True, "status": "not_found", @@ -191,7 +199,14 @@ def run_pchome_growth_momo_backfill( terminal_status = "partial_post_verifier_failed" next_machine_action = "retry_bounded_batch_after_drift_diagnosis" elif written_count > 0: - terminal_status = "verified_with_coverage_gain" + terminal_status = ( + "verified_with_coverage_gain" + if max( + float(post_verifier.get("count_coverage_delta") or 0), + float(post_verifier.get("revenue_coverage_delta") or 0), + ) > 0 + else "verified_write_readback" + ) next_machine_action = ( "continue_next_revenue_weighted_batch" if unresolved_after diff --git a/services/pchome_growth_same_item_reconciliation.py b/services/pchome_growth_same_item_reconciliation.py index e5ea3c9..124e8db 100644 --- a/services/pchome_growth_same_item_reconciliation.py +++ b/services/pchome_growth_same_item_reconciliation.py @@ -335,6 +335,125 @@ def verify_same_item_candidates( } +def readback_external_offer_candidates( + engine, + candidates: list[dict[str, Any]], +) -> dict[str, Any]: + """Verify committed offers by exact identity on a fresh DB connection.""" + expected = list(candidates or []) + if not expected: + return { + "success": True, + "status": "no_write_verified", + "expected_count": 0, + "readback_pass_count": 0, + "readback": [], + } + if not inspect(engine).has_table("external_offers"): + return { + "success": False, + "status": "external_offers_schema_missing", + "expected_count": len(expected), + "readback_pass_count": 0, + "readback": [], + } + + query = text(""" + SELECT + pchome_product_id, + source_product_id, + match_status, + data_quality_status, + raw_payload_json, + observed_at + FROM external_offers + WHERE source_code = 'momo_reference' + AND ingestion_method = 'targeted_momo_search' + AND pchome_product_id = :pchome_product_id + AND source_product_id = :source_product_id + ORDER BY observed_at DESC, id DESC + LIMIT 1 + """) + readback: list[dict[str, Any]] = [] + try: + with engine.connect() as conn: + for candidate in expected: + target_id = str(candidate.get("target_pchome_product_id") or "") + source_id = _candidate_source_id(candidate) + expected_identity = candidate.get("same_item_reconciliation") or {} + expected_fingerprint = str( + candidate.get("same_item_candidate_fingerprint") or "" + ) + row = conn.execute(query, { + "pchome_product_id": target_id, + "source_product_id": source_id, + }).mappings().first() + raw_payload: dict[str, Any] = {} + if row: + raw_payload_value = row.get("raw_payload_json") + if isinstance(raw_payload_value, dict): + raw_payload = raw_payload_value + elif isinstance(raw_payload_value, str): + try: + parsed = json.loads(raw_payload_value) + raw_payload = parsed if isinstance(parsed, dict) else {} + except json.JSONDecodeError: + raw_payload = {} + observed_identity = raw_payload.get("same_item_reconciliation") or {} + observed_fingerprint = str( + raw_payload.get("same_item_candidate_fingerprint") or "" + ) + identity_keys = ("trace_id", "run_id", "work_item_id") + identity_passed = bool( + all(str(expected_identity.get(key) or "") for key in identity_keys) + and all( + str(observed_identity.get(key) or "") + == str(expected_identity.get(key) or "") + for key in identity_keys + ) + ) + passed = bool( + row + and str(row.get("pchome_product_id") or "") == target_id + and str(row.get("source_product_id") or "") == source_id + and row.get("match_status") == "verified" + and row.get("data_quality_status") == "verified" + and expected_fingerprint + and observed_fingerprint == expected_fingerprint + and identity_passed + ) + readback.append({ + "target_pchome_product_id": target_id, + "source_product_id": source_id, + "observed_source_product_id": ( + str(row.get("source_product_id") or "") if row else "" + ), + "candidate_fingerprint": expected_fingerprint, + "observed_candidate_fingerprint": observed_fingerprint, + "identity_passed": identity_passed, + "passed": passed, + }) + except Exception as exc: + return { + "success": False, + "status": "external_offer_readback_failed", + "expected_count": len(expected), + "readback_pass_count": sum(1 for item in readback if item["passed"]), + "readback": readback, + "error_type": type(exc).__name__, + } + + pass_count = sum(1 for item in readback if item["passed"]) + success = pass_count == len(expected) + return { + "success": success, + "status": "verified" if success else "readback_mismatch", + "expected_count": len(expected), + "readback_pass_count": pass_count, + "readback": readback, + } + + def build_coverage_snapshot(payload: dict[str, Any]) -> dict[str, Any]: opportunities = list((payload or {}).get("opportunities") or []) total_revenue = sum(max(0.0, _float(item.get("sales_7d"))) for item in opportunities) @@ -385,17 +504,53 @@ def build_coverage_post_verifier( str(item.get("pchome_product_id") or ""): item for item in list((after_payload or {}).get("opportunities") or []) } + independent_readback = (sync_result or {}).get("independent_readback") or {} + independent_required = bool( + (sync_result or {}).get("independent_readback_required") + ) + direct_by_pair = { + ( + str(item.get("target_pchome_product_id") or ""), + str(item.get("source_product_id") or ""), + ): item + for item in independent_readback.get("readback") or [] + } readback: list[dict[str, Any]] = [] for candidate in verified_candidates: target_id = str(candidate.get("target_pchome_product_id") or "") source_id = _candidate_source_id(candidate) - external_price = (after_by_id.get(target_id) or {}).get("external_price") or {} - observed_source_id = str(external_price.get("momo_sku") or "") - passed = bool(external_price and observed_source_id == source_id) + direct = direct_by_pair.get((target_id, source_id)) + if direct is not None: + observed_source_id = str(direct.get("observed_source_product_id") or "") + passed = bool(direct.get("passed")) + readback_source = "external_offers_exact_identity" + expected_fingerprint = direct.get("candidate_fingerprint") + observed_fingerprint = direct.get("observed_candidate_fingerprint") + identity_passed = bool(direct.get("identity_passed")) + else: + external_price = (after_by_id.get(target_id) or {}).get("external_price") or {} + observed_source_id = str(external_price.get("momo_sku") or "") + passed = bool( + not independent_required + and external_price + and observed_source_id == source_id + ) + readback_source = ( + "missing_required_external_offer_readback" + if independent_required + else "top50_payload_compatibility_fallback" + ) + expected_fingerprint = candidate.get("same_item_candidate_fingerprint") + observed_fingerprint = None + identity_passed = None readback.append({ "target_pchome_product_id": target_id, "source_product_id": source_id, "observed_source_product_id": observed_source_id, + "readback_source": readback_source, + "candidate_fingerprint": expected_fingerprint, + "observed_candidate_fingerprint": observed_fingerprint, + "identity_passed": identity_passed, "passed": passed, }) @@ -407,6 +562,10 @@ def build_coverage_post_verifier( selected_count > 0 and written_count == selected_count and readback_pass_count == selected_count + and ( + not independent_required + or bool(independent_readback.get("success")) + ) ) no_regression = ( after["comparison_ready_count"] >= before["comparison_ready_count"] @@ -426,6 +585,8 @@ def build_coverage_post_verifier( "selected_candidate_count": selected_count, "written_count": written_count, "readback_pass_count": readback_pass_count, + "independent_readback_required": independent_required, + "independent_readback_status": independent_readback.get("status"), "count_coverage_delta": round( after["count_coverage_rate"] - before["count_coverage_rate"], 3 ), @@ -713,6 +874,7 @@ __all__ = ( "ensure_evidence_receipt_table", "new_reconciliation_identity", "persist_reconciliation_receipts", + "readback_external_offer_candidates", "read_latest_reconciliation_receipts", "verify_same_item_candidates", ) diff --git a/tests/test_pchome_revenue_growth_service.py b/tests/test_pchome_revenue_growth_service.py index 49428d7..8cb7bd2 100644 --- a/tests/test_pchome_revenue_growth_service.py +++ b/tests/test_pchome_revenue_growth_service.py @@ -562,6 +562,23 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items() "written_count": len(candidates), } + def fake_readback(engine, candidates): + return { + "success": True, + "status": "verified", + "expected_count": len(candidates), + "readback_pass_count": len(candidates), + "readback": [ + { + "target_pchome_product_id": item["target_pchome_product_id"], + "source_product_id": item["product_id"], + "observed_source_product_id": item["product_id"], + "passed": True, + } + for item in candidates + ], + } + payload = run_pchome_growth_momo_backfill( engine, limit=2, @@ -569,6 +586,7 @@ def test_pchome_growth_momo_backfill_service_targets_unmapped_high_sales_items() search_func=fake_search, sync_func=fake_sync, sync_review_func=fake_sync_review, + readback_func=fake_readback, ) assert payload["success"] is True diff --git a/tests/test_pchome_same_item_reconciliation.py b/tests/test_pchome_same_item_reconciliation.py index 9f807d0..bedf925 100644 --- a/tests/test_pchome_same_item_reconciliation.py +++ b/tests/test_pchome_same_item_reconciliation.py @@ -153,6 +153,90 @@ def test_coverage_post_verifier_requires_exact_source_readback_and_reports_reven assert result["revenue_coverage_delta"] == 80 +def test_exact_offer_readback_survives_target_leaving_dynamic_top50(): + from services.pchome_growth_same_item_reconciliation import ( + build_coverage_post_verifier, + readback_external_offer_candidates, + ) + + engine = create_engine("sqlite:///:memory:") + identity = { + "trace_id": "trace-db-readback", + "run_id": "run-db-readback", + "work_item_id": "GROWTH-P0-001-B", + } + candidate = { + "target_pchome_product_id": "P-LEAVES-TOP50", + "product_id": "M-EXACT", + "same_item_candidate_fingerprint": "fingerprint-exact", + "same_item_reconciliation": identity, + } + with engine.begin() as conn: + conn.execute(text(""" + CREATE TABLE external_offers ( + id INTEGER PRIMARY KEY, + source_code TEXT NOT NULL, + ingestion_method TEXT NOT NULL, + pchome_product_id TEXT NOT NULL, + source_product_id TEXT NOT NULL, + match_status TEXT NOT NULL, + data_quality_status TEXT NOT NULL, + raw_payload_json TEXT NOT NULL, + observed_at TEXT NOT NULL + ) + """)) + conn.execute(text(""" + INSERT INTO external_offers ( + id, source_code, ingestion_method, pchome_product_id, + source_product_id, match_status, data_quality_status, + raw_payload_json, observed_at + ) VALUES ( + 1, 'momo_reference', 'targeted_momo_search', + 'P-LEAVES-TOP50', 'M-EXACT', 'verified', 'verified', + :raw_payload, '2026-07-14T13:00:00' + ) + """), { + "raw_payload": json.dumps({ + "same_item_reconciliation": identity, + "same_item_candidate_fingerprint": "fingerprint-exact", + }) + }) + + exact_readback = readback_external_offer_candidates(engine, [candidate]) + result = build_coverage_post_verifier( + before_payload={ + "stats": {"mapping_rate": 0, "mapped_count": 0}, + "opportunities": [{ + "pchome_product_id": "P-LEAVES-TOP50", + "sales_7d": 10000, + "external_price": None, + }], + }, + after_payload={ + "stats": {"mapping_rate": 0, "mapped_count": 0}, + "opportunities": [{ + "pchome_product_id": "P-NEW-TOP50-ROW", + "sales_7d": 10000, + "external_price": None, + }], + }, + verified_candidates=[candidate], + sync_result={ + "written_count": 1, + "independent_readback_required": True, + "independent_readback": exact_readback, + }, + ) + + assert exact_readback["status"] == "verified" + assert exact_readback["readback_pass_count"] == 1 + assert result["status"] == "verified" + assert result["all_checks_passed"] is True + assert result["readback"][0]["readback_source"] == "external_offers_exact_identity" + assert result["readback"][0]["identity_passed"] is True + assert result["readback"][0]["observed_candidate_fingerprint"] == "fingerprint-exact" + + def test_receipt_schema_apply_persistence_and_public_safe_readback(): from services.pchome_growth_same_item_reconciliation import ( ensure_evidence_receipt_table,