import json from sqlalchemy import create_engine, text def _target(product_id, name, price, rank): return { "product_id": product_id, "name": name, "price": price, "execution_rank": rank, "execution_priority_score": 90 - rank, "revenue_share_pct": 20, "target_state": "unmatched", } def test_revenue_weighted_targets_include_candidate_validation_and_sort_by_revenue(): from services.pchome_growth_same_item_reconciliation import ( build_revenue_weighted_targets, ) payload = { "opportunities": [ { "pchome_product_id": "LOW", "product_name": "低業績未配對", "sales_7d": 100, "sales_prev_7d": 100, "priority_score": 95, "external_price": None, "recommended_action": {"code": "map_external_product"}, }, { "pchome_product_id": "HIGH", "product_name": "高業績候選", "sales_7d": 10000, "sales_prev_7d": 9000, "priority_score": 40, "external_price": None, "review_candidate": {"momo_sku": "M-1"}, "recommended_action": {"code": "review_external_candidate"}, }, { "pchome_product_id": "READY", "product_name": "已有正式比價", "sales_7d": 50000, "priority_score": 100, "external_price": {"momo_sku": "M-2"}, "recommended_action": {"code": "monitor"}, }, ] } targets = build_revenue_weighted_targets(payload, 10) assert [item["product_id"] for item in targets] == ["HIGH", "LOW"] assert targets[0]["target_state"] == "candidate_validation" assert targets[0]["execution_rank"] == 1 assert targets[0]["execution_priority_score"] > targets[1]["execution_priority_score"] def test_independent_verifier_promotes_exact_and_unit_candidates_but_rejects_conflict(): from services.pchome_growth_same_item_reconciliation import ( verify_same_item_candidates, ) targets = [ _target("P-EXACT", "理膚寶水 全面修復霜 B5 40ml", 679, 1), _target("P-UNIT", "理膚寶水 全面修復霜 B5 40ml", 679, 2), _target("P-BLOCK", "巴黎萊雅 膠原輕盈乳霜 60ml", 1249, 3), ] candidates = [ { "product_id": "M-EXACT", "name": "理膚寶水 B5 修復霜 40ml", "price": 699, "target_pchome_product_id": "P-EXACT", "auto_compare_type": "manual_review", }, { "product_id": "M-UNIT", "name": "理膚寶水 B5 全面修復霜 40ml x2 超值組", "price": 1199, "target_pchome_product_id": "P-UNIT", "auto_compare_type": "unit_price", }, { "product_id": "M-BLOCK", "name": "蘭蔻 玫瑰霜 60ml", "price": 5000, "target_pchome_product_id": "P-BLOCK", "target_search_term": "巴黎萊雅 膠原輕盈乳霜 60ml", "auto_compare_type": "total_price", }, ] identity = { "trace_id": "trace-1", "run_id": "run-1", "work_item_id": "GROWTH-P0-001-B", } result = verify_same_item_candidates(targets, candidates, identity=identity) selected = {item["target_pchome_product_id"]: item for item in result["verified_candidates"]} assert result["selected_candidate_count"] == 2 assert result["blocked_candidate_count"] == 1 assert result["claim_drift_count"] == 1 assert selected["P-EXACT"]["auto_compare_type"] == "total_price" assert selected["P-UNIT"]["auto_compare_type"] == "unit_price" assert selected["P-UNIT"]["target_unit_price_comparison"]["comparable"] is True assert selected["P-EXACT"]["same_item_candidate_fingerprint"] assert result["blocked_candidates"][0]["reason_code"] == "same_item_verifier_rejected" assert result["blocked_candidates"][0]["source_name"] == "蘭蔻 玫瑰霜 60ml" assert result["blocked_candidates"][0]["search_term"] == "巴黎萊雅 膠原輕盈乳霜 60ml" assert result["blocked_candidates"][0]["comparison_mode"] assert result["blocked_candidates"][0]["reasons"] def test_independent_verifier_prefers_exact_sheet_pack_over_unit_candidate(): from services.pchome_growth_same_item_reconciliation import ( verify_same_item_candidates, ) target = _target( "P-GATSBY-42", "GATSBY 潔面濕紙巾超值包42張入", 199, 1, ) candidates = [ { "product_id": "M-GATSBY-42", "name": "【GATSBY】潔面濕紙巾超值包42張入(2款任選)", "price": 179, "target_pchome_product_id": "P-GATSBY-42", "target_search_term": "gatsby 潔面濕紙巾 42張", "auto_compare_type": "manual_review", }, { "product_id": "M-GATSBY-15", "name": "【GATSBY】潔面濕紙巾15張入(3款任選)", "price": 89, "target_pchome_product_id": "P-GATSBY-42", "target_search_term": "gatsby 潔面濕紙巾 42張", "auto_compare_type": "manual_review", }, ] result = verify_same_item_candidates( [target], candidates, identity={ "trace_id": "trace-gatsby-pack", "run_id": "run-gatsby-pack", "work_item_id": "GROWTH-P0-001-B", }, ) assert result["selected_candidate_count"] == 1 assert result["verified_candidates"][0]["product_id"] == "M-GATSBY-42" assert result["verified_candidates"][0]["auto_compare_type"] == "total_price" assert result["verified_candidate_count"] == 2 assert result["blocked_candidate_count"] == 0 unit_result = verify_same_item_candidates( [target], [candidates[1]], identity={ "trace_id": "trace-gatsby-unit", "run_id": "run-gatsby-unit", "work_item_id": "GROWTH-P0-001-B", }, ) assert unit_result["selected_candidate_count"] == 1 unit_candidate = unit_result["verified_candidates"][0] assert unit_candidate["auto_compare_type"] == "unit_price" assert unit_candidate["target_price_basis"] == "unit_price" assert unit_candidate["target_unit_price_comparison"]["comparable"] is True assert unit_candidate["target_unit_price_comparison"]["unit_label"] == "張" def test_independent_verifier_never_treats_outer_sheet_pack_as_exact_total_price(): from services.pchome_growth_same_item_reconciliation import ( verify_same_item_candidates, ) target = _target( "P-GATSBY-42", "GATSBY 潔面濕紙巾超值包42張入", 199, 1, ) result = verify_same_item_candidates( [target], [{ "product_id": "M-GATSBY-42-X4", "name": "【GATSBY】潔面濕紙巾超值包42張入*4包(4款任選)", "price": 495, "target_pchome_product_id": "P-GATSBY-42", "target_search_term": "gatsby 潔面濕紙巾 42張", "auto_compare_type": "total_price", }], identity={ "trace_id": "trace-gatsby-outer-pack", "run_id": "run-gatsby-outer-pack", "work_item_id": "GROWTH-P0-001-B", }, ) assert result["selected_candidate_count"] == 1 candidate = result["verified_candidates"][0] assert candidate["auto_compare_type"] == "unit_price" assert candidate["same_item_reconciliation"]["candidate_claim_drift"] is True assert candidate["target_match_type"] == "same_product_different_pack" assert candidate["target_price_basis"] == "unit_price" assert candidate["target_unit_price_comparison"]["unit_label"] == "張" assert candidate["target_unit_price_comparison"]["momo_total_quantity"] == 168 assert candidate["target_unit_price_comparison"]["competitor_total_quantity"] == 42 def test_coverage_post_verifier_requires_exact_source_readback_and_reports_revenue_delta(): from services.pchome_growth_same_item_reconciliation import ( build_coverage_post_verifier, ) before = { "stats": {"mapping_rate": 0, "mapped_count": 0}, "opportunities": [ {"pchome_product_id": "P-1", "sales_7d": 8000, "external_price": None}, {"pchome_product_id": "P-2", "sales_7d": 2000, "external_price": None}, ], } after = { "stats": {"mapping_rate": 50, "mapped_count": 1}, "opportunities": [ { "pchome_product_id": "P-1", "sales_7d": 8000, "external_price": {"momo_sku": "M-1"}, }, {"pchome_product_id": "P-2", "sales_7d": 2000, "external_price": None}, ], } result = build_coverage_post_verifier( before_payload=before, after_payload=after, verified_candidates=[{ "target_pchome_product_id": "P-1", "product_id": "M-1", }], sync_result={"written_count": 1}, ) assert result["status"] == "verified" assert result["all_checks_passed"] is True assert result["comparison_ready_delta"] == 1 assert result["count_coverage_delta"] == 50 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, persist_reconciliation_receipts, read_latest_reconciliation_receipts, ) engine = create_engine("sqlite:///:memory:") schema = ensure_evidence_receipt_table(engine) target = _target("P-1", "商品", 100, 1) target.update({"sales_7d": 8000, "revenue_at_risk": 8000}) run_receipt = { "generated_at": "2026-07-14T12:00:00+00:00", "identity": { "trace_id": "trace-1", "run_id": "run-1", "work_item_id": "GROWTH-P0-001-B", }, "source_receipt": {"candidate_count": 1}, "candidate_verification": { "verified_candidates": [{ "target_pchome_product_id": "P-1", "product_id": "M-1", "name": "商品 100ml", "price": 90, "target_search_term": "商品 100ml", "target_match_score": 0.99, "target_comparison_mode": "total_price", "target_match_type": "exact", "target_price_basis": "total_price", "target_alert_tier": "price_alert_exact", "auto_compare_type": "total_price", "same_item_candidate_fingerprint": "fingerprint-1", "same_item_reconciliation": { "independent_verifier_passed": True, }, }], "review_candidates": [], "blocked_candidates": [], }, "coverage_diff": {"comparison_ready_delta": 1}, "risk_policy": {"risk": "medium"}, "check_mode": {"verified_candidate_count": 1}, "execution": {"written_count": 1}, "post_verifier": { "status": "verified", "readback": [{"target_pchome_product_id": "P-1", "passed": True}], }, "terminal": {"status": "verified_with_coverage_gain"}, } persistence = persist_reconciliation_receipts( engine, targets=[target], run_receipt=run_receipt, ) readback = read_latest_reconciliation_receipts(engine) with engine.connect() as conn: evidence = conn.execute(text( "SELECT evidence_delta_json FROM external_offer_evidence_receipts" )).scalar_one() evidence = json.loads(evidence) assert schema["status"] == "created_and_verified" assert persistence["status"] == "persisted_and_verified" assert persistence["written_count"] == 1 assert readback["status"] == "ready" assert readback["verified_count"] == 1 assert readback["latest_run_id"] == "run-1" assert readback["receipts"][0]["work_item_id"] == "GROWTH-P0-001-B" assert evidence["target"]["sales_7d"] == 8000 assert evidence["candidate_evidence"]["verified_candidate_count"] == 1 assert evidence["candidate_evidence"]["verified_candidates"][0]["source_product_id"] == "M-1" assert evidence["candidate_evidence"]["verified_candidates"][0]["search_term"] == "商品 100ml" assert evidence["next_machine_action"] == "continue_next_revenue_weighted_batch" assert readback["receipts"][0]["next_machine_action"] == ( "continue_next_revenue_weighted_batch" ) assert readback["receipts"][0]["candidate_evidence_summary"] == { "verified_candidate_count": 1, "review_candidate_count": 0, "blocked_candidate_count": 0, "reason_codes": [], "verifier_reasons": [], } def test_rejected_pack_candidate_persists_autonomous_retry_action(): from services.pchome_growth_same_item_reconciliation import ( ensure_evidence_receipt_table, persist_reconciliation_receipts, read_latest_reconciliation_receipts, ) engine = create_engine("sqlite:///:memory:") ensure_evidence_receipt_table(engine) target = _target( "P-GATSBY-42", "GATSBY 潔面濕紙巾超值包42張入", 199, 1, ) run_receipt = { "generated_at": "2026-07-15T00:30:00+00:00", "identity": { "trace_id": "trace-pack-retry", "run_id": "run-pack-retry", "work_item_id": "GROWTH-P0-001-B", }, "candidate_verification": { "verified_candidates": [], "review_candidates": [], "blocked_candidates": [{ "target_pchome_product_id": "P-GATSBY-42", "source_product_id": "M-GATSBY-15", "source_name": "GATSBY 潔面濕紙巾15張入", "source_price": 89, "search_term": "gatsby 潔面濕紙巾 42張", "reason_code": "same_item_verifier_rejected", "match_score": 0.92, "hard_veto": True, "comparison_mode": "blocked", "match_type": "mismatch", "price_basis": "none", "alert_tier": "none", "reasons": ["count_conflict"], }], }, "post_verifier": {"status": "no_write_verified", "readback": []}, "terminal": {"status": "verified_no_write"}, } persistence = persist_reconciliation_receipts( engine, targets=[target], run_receipt=run_receipt, ) readback = read_latest_reconciliation_receipts(engine) with engine.connect() as conn: evidence = conn.execute(text( "SELECT evidence_delta_json FROM external_offer_evidence_receipts" )).scalar_one() evidence = json.loads(evidence) assert persistence["status"] == "persisted_and_verified" assert evidence["ai_decision"] == "candidate_rejected" assert evidence["candidate_evidence"]["reason_codes"] == [ "same_item_verifier_rejected" ] assert evidence["candidate_evidence"]["verifier_reasons"] == [ "count_conflict" ] assert evidence["next_machine_action"] == ( "retry_search_with_exact_pack_quantity" ) receipt = readback["receipts"][0] assert receipt["apply_status"] == "no_write" assert receipt["next_machine_action"] == ( "retry_search_with_exact_pack_quantity" ) assert receipt["candidate_evidence_summary"]["blocked_candidate_count"] == 1 assert receipt["candidate_evidence_summary"]["verifier_reasons"] == [ "count_conflict" ] def test_targeted_momo_search_keeps_same_source_product_per_pchome_target(): from services.momo_crawler import search_momo_products_for_pchome_products class FakeCrawler: def search_products(self, term, limit): return True, "ok", [{ "product_id": "MOMO-SAME", "name": "理膚寶水 B5 全面修復霜 40ml", "price": 699, }] success, _, candidates = search_momo_products_for_pchome_products( [ {"product_id": "P-1", "name": "理膚寶水 全面修復霜 B5 40ml", "price": 679}, {"product_id": "P-2", "name": "理膚寶水 全面修復霜 B5 40ml", "price": 679}, ], limit_per_product=3, max_products=2, max_terms_per_product=2, crawler=FakeCrawler(), ) assert success is True assert len(candidates) == 2 assert {item["target_pchome_product_id"] for item in candidates} == {"P-1", "P-2"} def test_dashboard_exposes_count_and_revenue_coverage_without_a_text_wall(): from pathlib import Path template = Path("templates/dashboard_v2.html").read_text(encoding="utf-8") css = Path("web/static/css/page-dashboard-v2.css").read_text(encoding="utf-8") assert "業績比價覆蓋" in template assert "growth.comparison_revenue_coverage_rate | default(0)" in template assert "growth.comparison_revenue_coverage_width | default(0)" in template assert "growth.comparison_ready_sales_7d" in template assert "growth.comparison_total_sales_7d" in template assert "growth.mapped_count" in template assert ".growth-mapping-backlog {" in css assert ".growth-mapping-backlog-grid {" in css assert ".growth-automation-panel {" in css assert ".growth-automation-steps {" in css assert ".growth-automation-step {" in css def test_same_item_reconciliation_route_returns_durable_readback(monkeypatch): from flask import Flask from routes import ai_routes as routes from services.pchome_growth_same_item_reconciliation import ( ensure_evidence_receipt_table, ) engine = create_engine("sqlite:///:memory:") ensure_evidence_receipt_table(engine) with engine.begin() as conn: conn.execute(text(""" INSERT INTO external_offer_evidence_receipts ( receipt_id, pchome_product_id, automation_decision, payload_hash, evidence_delta_json, apply_status ) VALUES ( 'sameitem-route', 'P-1', 'candidate_rejected', 'hash-1', :evidence, 'no_write' ) """), { "evidence": json.dumps({ "identity": { "trace_id": "trace-route", "run_id": "run-route", "work_item_id": "GROWTH-P0-001-B", }, "post_verifier": {"status": "no_write_verified"}, }) }) monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", lambda _: engine) app = Flask(__name__) with app.test_request_context( "/api/ai/pchome-growth/same-item-reconciliation?limit=5" ): response = routes.api_pchome_growth_same_item_reconciliation.__wrapped__() payload = response.get_json() assert payload["status"] == "ready" assert payload["receipt_count"] == 1 assert payload["latest_run_id"] == "run-route"