feat(growth): automate exact pack reconciliation
Some checks failed
CD Pipeline / deploy (push) Has been cancelled

This commit is contained in:
ogt
2026-07-15 09:37:54 +08:00
parent d7be63f71e
commit 05c7cd8421
6 changed files with 448 additions and 6 deletions

View File

@@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '')
# ==========================================
# 系統版本與路徑
# ==========================================
SYSTEM_VERSION = "V10.796"
SYSTEM_VERSION = "V10.797"
LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log')
public_url = PUBLIC_URL # 用於模板顯示

View File

@@ -127,6 +127,8 @@ GENERIC_TOKENS = {
"",
"",
"",
"",
"張入",
"",
"",
"",
@@ -854,8 +856,8 @@ PRODUCT_TYPES = {
}
COUNT_UNITS = {"", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "", "刀把", "刀片", "刀頭", ""}
COUNT_UNIT_PATTERN = r"(?:刀把|刀片|刀頭|入|組|瓶|支|條|盒|包|袋|片|顆|粒|錠|枚|件|罐|杯|本|蕊)"
PIECE_UNITS = {"", "", "", "", "", "", ""}
COUNT_UNIT_PATTERN = r"(?:刀把|刀片|刀頭|入|組|瓶|支|條|盒|包|袋|片|張|顆|粒|錠|枚|件|罐|杯|本|蕊)"
PIECE_UNITS = {"", "", "", "", "", "", "", ""}
CONTAINER_UNITS = {"", "", "", "", "", "", "", "", "", ""}
COUNT_UNIT_FAMILIES = {
"刀片": "blade",
@@ -3327,7 +3329,13 @@ def _search_spec_terms(identity: ProductIdentity) -> list[str]:
dosage = identity.dosages_mg[0]
specs.append(f"{dosage:g}mg")
if identity.total_piece_count:
specs.append(f"{identity.total_piece_count}")
piece_units = [
unit
for _, unit in identity.counts
if unit in PIECE_UNITS
]
unit = piece_units[0] if piece_units else ""
specs.append(f"{identity.total_piece_count}{unit}")
return specs

View File

@@ -0,0 +1,178 @@
"""Public-safe evidence shaping for same-item reconciliation receipts."""
from __future__ import annotations
from typing import Any
def _float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def _candidate_source_id(candidate: dict[str, Any]) -> str:
return str(
candidate.get("product_id")
or candidate.get("goodsCode")
or candidate.get("id")
or candidate.get("source_product_id")
or ""
).strip()
def _candidate_name(candidate: dict[str, Any]) -> str:
return str(
candidate.get("name")
or candidate.get("title")
or candidate.get("source_name")
or ""
).strip()
def safe_candidate_evidence(candidate: dict[str, Any]) -> dict[str, Any]:
reconciliation = candidate.get("same_item_reconciliation") or {}
return {
"target_pchome_product_id": str(
candidate.get("target_pchome_product_id") or ""
),
"source_product_id": _candidate_source_id(candidate),
"source_name": _candidate_name(candidate),
"source_price": _float(
candidate.get("price"),
_float(candidate.get("source_price")),
),
"search_term": str(
candidate.get("target_search_term")
or candidate.get("search_term")
or ""
),
"match_score": _float(
candidate.get("target_match_score"),
_float(candidate.get("match_score")),
),
"comparison_mode": str(
candidate.get("target_comparison_mode")
or candidate.get("comparison_mode")
or ""
),
"match_type": str(
candidate.get("target_match_type")
or candidate.get("match_type")
or ""
),
"price_basis": str(
candidate.get("target_price_basis")
or candidate.get("price_basis")
or ""
),
"alert_tier": str(
candidate.get("target_alert_tier")
or candidate.get("alert_tier")
or ""
),
"hard_veto": bool(
candidate.get("target_hard_veto")
if "target_hard_veto" in candidate
else candidate.get("hard_veto")
),
"reason_code": str(candidate.get("reason_code") or ""),
"reasons": list(
candidate.get("target_match_reasons")
or candidate.get("reasons")
or []
)[:8],
"auto_compare_type": str(candidate.get("auto_compare_type") or ""),
"candidate_fingerprint": str(
candidate.get("same_item_candidate_fingerprint")
or reconciliation.get("candidate_fingerprint")
or ""
),
"independent_verifier_passed": bool(
reconciliation.get("independent_verifier_passed")
),
}
def build_target_candidate_evidence(
target_id: str,
run_receipt: dict[str, Any],
) -> dict[str, Any]:
verification = run_receipt.get("candidate_verification") or {}
def matching(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [
safe_candidate_evidence(item)
for item in items
if str(item.get("target_pchome_product_id") or "") == target_id
]
verified = matching(list(verification.get("verified_candidates") or []))
review = matching(list(verification.get("review_candidates") or []))
blocked = matching(list(verification.get("blocked_candidates") or []))
reason_codes = sorted({
item.get("reason_code")
for item in blocked
if item.get("reason_code")
})
verifier_reasons = sorted({
reason
for item in [*verified, *review, *blocked]
for reason in item.get("reasons") or []
if reason
})
return {
"verified_candidates": verified,
"review_candidates": review,
"blocked_candidates": blocked,
"verified_candidate_count": len(verified),
"review_candidate_count": len(review),
"blocked_candidate_count": len(blocked),
"reason_codes": reason_codes,
"verifier_reasons": verifier_reasons,
}
def select_next_machine_action(
decision: str,
candidate_evidence: dict[str, Any],
) -> str:
if decision == "promoted_verified":
return "continue_next_revenue_weighted_batch"
if decision == "apply_not_verified":
return "retry_exact_identity_readback_before_next_apply"
reason_codes = set(candidate_evidence.get("reason_codes") or [])
verifier_reasons = set(candidate_evidence.get("verifier_reasons") or [])
if decision == "no_candidate":
return "expand_search_terms_with_brand_pack_and_spec_anchors"
if "source_identity_missing" in reason_codes:
return "refresh_source_identity_then_retry"
if "positive_price_evidence_missing" in reason_codes:
return "refresh_positive_price_evidence_then_retry"
if verifier_reasons & {
"count_conflict",
"bundle_offer_conflict",
"multi_component_conflict",
"component_count_conflict",
"pack_quantity_difference",
}:
return "retry_search_with_exact_pack_quantity"
if verifier_reasons & {
"variant_descriptor_conflict",
"variant_option_conflict",
"variant_selection_review",
"makeup_catalog_selection_gap",
}:
return "retry_search_with_exact_variant_evidence"
if verifier_reasons & {"unit_comparable", "unit_price_review"}:
return "replay_unit_basis_before_promotion"
return "refresh_source_candidates_with_evidence_delta"
__all__ = (
"build_target_candidate_evidence",
"safe_candidate_evidence",
"select_next_machine_action",
)

View File

@@ -10,6 +10,11 @@ from typing import Any, Callable
from sqlalchemy import bindparam, inspect, text
from services.pchome_growth_same_item_receipt_evidence import (
build_target_candidate_evidence,
select_next_machine_action,
)
CONTRACT_VERSION = "pchome_same_item_reconciliation_v1"
WORK_ITEM_ID = "GROWTH-P0-001-B"
@@ -169,6 +174,9 @@ def verify_same_item_candidates(
blocked.append({
"target_pchome_product_id": target_id,
"source_product_id": source_id,
"source_name": source_name,
"source_price": _float(candidate.get("price")),
"search_term": str(candidate.get("target_search_term") or ""),
"reason_code": "target_not_in_bounded_batch",
})
continue
@@ -176,6 +184,9 @@ def verify_same_item_candidates(
blocked.append({
"target_pchome_product_id": target_id,
"source_product_id": source_id,
"source_name": source_name,
"source_price": _float(candidate.get("price")),
"search_term": str(candidate.get("target_search_term") or ""),
"reason_code": "source_identity_missing",
})
continue
@@ -186,6 +197,9 @@ def verify_same_item_candidates(
blocked.append({
"target_pchome_product_id": target_id,
"source_product_id": source_id,
"source_name": source_name,
"source_price": source_price,
"search_term": str(candidate.get("target_search_term") or ""),
"reason_code": "positive_price_evidence_missing",
})
continue
@@ -235,9 +249,18 @@ def verify_same_item_candidates(
blocked.append({
"target_pchome_product_id": target_id,
"source_product_id": source_id,
"source_name": source_name,
"source_price": source_price,
"search_term": str(candidate.get("target_search_term") or ""),
"reason_code": "same_item_verifier_rejected",
"match_score": _float(getattr(diagnostics, "score", 0.0)),
"hard_veto": bool(getattr(diagnostics, "hard_veto", False)),
"comparison_mode": str(
getattr(diagnostics, "comparison_mode", "") or ""
),
"match_type": str(getattr(diagnostics, "match_type", "") or ""),
"price_basis": str(getattr(diagnostics, "price_basis", "") or ""),
"alert_tier": str(getattr(diagnostics, "alert_tier", "") or ""),
"reasons": list(getattr(diagnostics, "reasons", ()) or ())[:8],
})
continue
@@ -720,12 +743,21 @@ def persist_reconciliation_receipts(
for target in receipt_targets:
target_id = str(target.get("product_id") or "__run__")
decision, apply_status = _target_receipt_decision(target, run_receipt)
candidate_evidence = build_target_candidate_evidence(target_id, run_receipt)
next_machine_action = select_next_machine_action(
decision,
candidate_evidence,
)
evidence = {
"contract_version": CONTRACT_VERSION,
"identity": identity,
"generated_at": generated_at,
"target": {
"pchome_product_id": target_id,
"name": target.get("name"),
"price": target.get("price"),
"sales_7d": target.get("sales_7d"),
"revenue_at_risk": target.get("revenue_at_risk"),
"target_state_before": target.get("target_state"),
"execution_rank": target.get("execution_rank"),
"execution_priority_score": target.get("execution_priority_score"),
@@ -734,6 +766,8 @@ def persist_reconciliation_receipts(
"source_receipt": run_receipt.get("source_receipt") or {},
"source_of_truth_diff": run_receipt.get("coverage_diff") or {},
"ai_decision": decision,
"candidate_evidence": candidate_evidence,
"next_machine_action": next_machine_action,
"risk_policy": run_receipt.get("risk_policy") or {},
"check_mode": run_receipt.get("check_mode") or {},
"execution": run_receipt.get("execution") or {},
@@ -840,6 +874,7 @@ def read_latest_reconciliation_receipts(engine, *, limit: int = 20) -> dict[str,
except json.JSONDecodeError:
evidence = {}
evidence = evidence if isinstance(evidence, dict) else {}
candidate_evidence = evidence.get("candidate_evidence") or {}
receipts.append({
"receipt_id": row.get("receipt_id"),
"pchome_product_id": row.get("pchome_product_id"),
@@ -852,6 +887,22 @@ def read_latest_reconciliation_receipts(engine, *, limit: int = 20) -> dict[str,
"run_id": (evidence.get("identity") or {}).get("run_id"),
"work_item_id": (evidence.get("identity") or {}).get("work_item_id"),
"post_verifier_status": (evidence.get("post_verifier") or {}).get("status"),
"next_machine_action": evidence.get("next_machine_action"),
"candidate_evidence_summary": {
"verified_candidate_count": int(
candidate_evidence.get("verified_candidate_count") or 0
),
"review_candidate_count": int(
candidate_evidence.get("review_candidate_count") or 0
),
"blocked_candidate_count": int(
candidate_evidence.get("blocked_candidate_count") or 0
),
"reason_codes": list(candidate_evidence.get("reason_codes") or []),
"verifier_reasons": list(
candidate_evidence.get("verifier_reasons") or []
)[:8],
},
})
return {
"contract_version": CONTRACT_VERSION,

View File

@@ -3441,6 +3441,36 @@ def test_marketplace_search_terms_prefer_readable_brand_core_spec():
assert not any(term.endswith(" l") for term in terms)
def test_marketplace_sheet_count_is_structured_for_search_and_identity():
from services.marketplace_product_matcher import (
build_search_terms,
parse_product_identity,
score_marketplace_match,
)
pchome_name = "GATSBY 潔面濕紙巾超值包42張入"
same_pack = "【GATSBY】潔面濕紙巾超值包42張入(2款任選)"
different_pack = "【GATSBY】潔面濕紙巾15張入(3款任選)"
identity = parse_product_identity(pchome_name)
terms = build_search_terms(pchome_name, max_terms=5)
exact = score_marketplace_match(same_pack, pchome_name)
rejected = score_marketplace_match(different_pack, pchome_name)
assert (42, "") in identity.counts
assert identity.total_piece_count == 42
assert terms[0] == "gatsby 潔面濕紙巾 42張"
assert not any(term in {"gatsby 張入", "張入"} for term in terms)
assert exact.hard_veto is False
assert exact.match_type == "exact"
assert exact.price_basis == "total_price"
assert exact.alert_tier == "price_alert_exact"
assert exact.spec_score == 1.0
assert "spec_name_alignment" in exact.reasons
assert rejected.hard_veto is True
assert "count_conflict" in rejected.reasons
def test_marketplace_search_terms_prioritize_identity_phrase_over_ambiguous_copy():
from services.marketplace_product_matcher import build_search_terms
@@ -3556,7 +3586,7 @@ def test_marketplace_search_terms_prefer_exact_identity_for_nail_foam_and_foot_m
assert "小銀蓋" not in " ".join(opi_terms[:3])
assert arau_terms[0] == "愛樂寶 溫和洗手慕斯 300ml"
assert "溫和不乾澀" not in " ".join(arau_terms[:3])
assert kameria_terms[0] == "凱蜜菈 足足稱奇足膜 17ml 2"
assert kameria_terms[0] == "凱蜜菈 足足稱奇足膜 17ml 2"
assert "枚入" not in " ".join(kameria_terms[:3])

View File

@@ -90,6 +90,7 @@ def test_independent_verifier_promotes_exact_and_unit_candidates_but_rejects_con
"name": "蘭蔻 玫瑰霜 60ml",
"price": 5000,
"target_pchome_product_id": "P-BLOCK",
"target_search_term": "巴黎萊雅 膠原輕盈乳霜 60ml",
"auto_compare_type": "total_price",
},
]
@@ -110,6 +111,63 @@ def test_independent_verifier_promotes_exact_and_unit_candidates_but_rejects_con
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_promotes_exact_sheet_pack_and_blocks_wrong_count():
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"
rejected = next(
item
for item in result["blocked_candidates"]
if item["source_product_id"] == "M-GATSBY-15"
)
assert rejected["hard_veto"] is True
assert rejected["search_term"] == "gatsby 潔面濕紙巾 42張"
assert "count_conflict" in rejected["reasons"]
def test_coverage_post_verifier_requires_exact_source_readback_and_reports_revenue_delta():
@@ -247,6 +305,7 @@ def test_receipt_schema_apply_persistence_and_public_safe_readback():
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": {
@@ -256,7 +315,24 @@ def test_receipt_schema_apply_persistence_and_public_safe_readback():
},
"source_receipt": {"candidate_count": 1},
"candidate_verification": {
"verified_candidates": [{"target_pchome_product_id": "P-1"}],
"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},
@@ -276,6 +352,11 @@ def test_receipt_schema_apply_persistence_and_public_safe_readback():
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"
@@ -284,6 +365,100 @@ def test_receipt_schema_apply_persistence_and_public_safe_readback():
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():