wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

補齊 PChome 例外解法收斂包
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-01 20:57:23 +08:00
parent 5f57dfd29c
commit 12e03ef64e
3 changed files with 469 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

319
services/pchome_mapping_backlog_service.py
View File

@@ -41,6 +41,9 @@ DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = (
DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout"
)
AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
@@ -1646,6 +1649,197 @@ def _summarize_exception_auto_resolution_artifacts(artifacts: list[dict[str, Any
}
def _expanded_retry_search_terms_from_artifact(artifact: dict[str, Any]) -> list[str]:
terms: list[str] = []
resolvers = artifact.get("resolvers") or {}
for key in ("unit_basis_search_expansion", "brand_spec_search_expansion"):
resolver = resolvers.get(key) or {}
terms.extend(str(term).strip() for term in resolver.get("expanded_search_terms") or [])
return list(dict.fromkeys(term for term in terms if term))
def _retry_search_targets_from_artifact(
artifact: dict[str, Any],
*,
max_terms_per_product: int,
) -> list[dict[str, Any]]:
subject = artifact.get("subject") or {}
terms = _expanded_retry_search_terms_from_artifact(artifact)[:max_terms_per_product]
return [
{
"product_id": subject.get("target_pchome_product_id") or "",
"name": term,
"price": subject.get("pchome_price"),
"source_artifact_id": artifact.get("artifact_id"),
"source_resolution_receipt_id": artifact.get("source_receipt_id"),
}
for term in terms
if term
]
def _run_retry_search_for_artifact(
artifact: dict[str, Any],
*,
execute_retry_search: bool,
limit_per_product: int,
max_terms_per_product: int,
min_score: float,
search_func: Any = None,
) -> dict[str, Any]:
targets = _retry_search_targets_from_artifact(
artifact,
max_terms_per_product=max_terms_per_product,
)
if not execute_retry_search or not targets:
return {
"executed": False,
"search_success": None,
"search_message": "retry_search_not_executed" if targets else "no_retry_search_terms",
"targets": targets,
"candidate_count": 0,
"candidates": [],
}
if search_func is None:
from services.momo_crawler import search_momo_products_for_pchome_products
search_func = search_momo_products_for_pchome_products
search_success, search_message, candidates = search_func(
targets,
limit_per_product=limit_per_product,
max_products=len(targets),
max_terms_per_product=1,
min_score=min_score,
)
candidates = list(candidates or [])
for candidate in candidates:
candidate["source_resolution_artifact_id"] = artifact.get("artifact_id")
candidate["source_resolution_receipt_id"] = artifact.get("source_receipt_id")
candidate["retry_search_source"] = "candidate_exception_resolution_closeout"
return {
"executed": True,
"search_success": bool(search_success),
"search_message": search_message,
"targets": targets,
"candidate_count": len(candidates),
"candidates": candidates[:20],
}
def _build_candidate_exception_resolution_closeout_receipt(
artifact: dict[str, Any],
retry_search_result: dict[str, Any],
) -> dict[str, Any]:
resolvers = artifact.get("resolvers") or {}
evidence_delta = resolvers.get("named_candidate_evidence_delta") or {}
retry_search_ready = bool(_expanded_retry_search_terms_from_artifact(artifact))
closeout_basis = {
"artifact_id": artifact.get("artifact_id"),
"resolver_keys": sorted(resolvers.keys()),
"retry_search_ready": retry_search_ready,
"retry_candidate_count": retry_search_result.get("candidate_count"),
}
closeout_hash = hashlib.sha256(
json.dumps(closeout_basis, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
).hexdigest()
retry_candidate_count = int(retry_search_result.get("candidate_count") or 0)
return {
"receipt_id": f"pchome-direct-mapping-exception-closeout-{closeout_hash[:16]}",
"source_artifact_id": artifact.get("artifact_id"),
"source_receipt_id": artifact.get("source_receipt_id"),
"source_decision_id": artifact.get("source_decision_id"),
"stage": "P2_machine_verifiable_exception_resolution_closeout",
"resolution_status": "AUTO_RESOLUTION_CLOSEOUT_READY",
"completed_resolvers": sorted(resolvers.keys()),
"evidence_delta": {
"ready": bool(evidence_delta),
"missing_evidence_keys": list(evidence_delta.get("missing_evidence_keys") or []),
"resolution": evidence_delta.get("resolution") or "not_applicable",
},
"retry_search": {
"ready": retry_search_ready,
"executed": bool(retry_search_result.get("executed")),
"search_success": retry_search_result.get("search_success"),
"search_message": retry_search_result.get("search_message"),
"target_count": len(retry_search_result.get("targets") or []),
"candidate_count": retry_candidate_count,
"targets": retry_search_result.get("targets") or [],
"candidates": retry_search_result.get("candidates") or [],
},
"next_package": (
"direct_mapping_candidate_decision_package_after_retry"
if retry_search_ready
else "no_write_receipt_verifier_when_identity_delta_clears"
),
"guardrails": {
"machine_actionable": True,
"can_auto_execute_read_only": retry_search_ready,
"writes_database": False,
"persists_candidate": False,
"requires_no_write_receipt": True,
"requires_verifier_before_persistence": True,
},
}
def _build_candidate_exception_resolution_closeout_receipts(
artifacts: list[dict[str, Any]],
*,
execute_retry_search: bool = False,
limit_per_product: int = 8,
max_terms_per_product: int = 5,
min_score: float = 0.45,
search_func: Any = None,
) -> list[dict[str, Any]]:
receipts = []
for artifact in artifacts:
retry_search_result = _run_retry_search_for_artifact(
artifact,
execute_retry_search=execute_retry_search,
limit_per_product=limit_per_product,
max_terms_per_product=max_terms_per_product,
min_score=min_score,
search_func=search_func,
)
receipts.append(_build_candidate_exception_resolution_closeout_receipt(artifact, retry_search_result))
return receipts
def _summarize_exception_resolution_closeout_receipts(receipts: list[dict[str, Any]]) -> dict[str, int]:
retry_search_ready_count = 0
retry_search_executed_count = 0
retry_candidate_count = 0
evidence_delta_closeout_count = 0
ready_for_next_decision_count = 0
for receipt in receipts:
retry_search = receipt.get("retry_search") or {}
evidence_delta = receipt.get("evidence_delta") or {}
retry_ready = bool(retry_search.get("ready"))
retry_executed = bool(retry_search.get("executed"))
retry_candidates = int(retry_search.get("candidate_count") or 0)
if retry_ready:
retry_search_ready_count += 1
if retry_executed:
retry_search_executed_count += 1
if evidence_delta.get("ready"):
evidence_delta_closeout_count += 1
retry_candidate_count += retry_candidates
if retry_candidates:
ready_for_next_decision_count += 1
return {
"exception_resolution_closeout_receipt_count": len(receipts),
"retry_search_ready_count": retry_search_ready_count,
"retry_search_executed_count": retry_search_executed_count,
"retry_candidate_count": retry_candidate_count,
"evidence_delta_closeout_count": evidence_delta_closeout_count,
"ready_for_next_candidate_decision_count": ready_for_next_decision_count,
"writes_database_count": 0,
}
def build_pchome_direct_mapping_auto_search_package(
payload: dict[str, Any],
batch_size: int = 5,
@@ -1987,6 +2181,96 @@ def build_pchome_direct_mapping_candidate_exception_auto_resolution_package(
}
def build_pchome_direct_mapping_candidate_exception_resolution_closeout_package(
payload: dict[str, Any],
batch_size: int = 5,
*,
execute_search: bool = False,
execute_retry_search: bool = False,
limit_per_product: int = 8,
max_terms_per_product: int = 5,
min_score: float = 0.45,
search_func: Any = None,
) -> dict[str, Any]:
"""Close out candidate exception auto-resolution artifacts into retry-ready receipts."""
auto_resolution = build_pchome_direct_mapping_candidate_exception_auto_resolution_package(
payload,
batch_size=batch_size,
execute_search=execute_search,
limit_per_product=limit_per_product,
max_terms_per_product=max_terms_per_product,
min_score=min_score,
search_func=search_func,
)
package = auto_resolution.get("auto_resolution_package") or {}
artifacts = list(package.get("auto_resolution_artifacts") or [])
closeout_receipts = _build_candidate_exception_resolution_closeout_receipts(
artifacts,
execute_retry_search=execute_retry_search,
limit_per_product=limit_per_product,
max_terms_per_product=max_terms_per_product,
min_score=min_score,
search_func=search_func,
)
closeout_summary = _summarize_exception_resolution_closeout_receipts(closeout_receipts)
selected_direct_count = int((auto_resolution.get("summary") or {}).get("selected_direct_mapping_count") or 0)
artifact_count = int((auto_resolution.get("summary") or {}).get("exception_auto_resolution_artifact_count") or 0)
if not selected_direct_count:
result = "NO_DIRECT_MAPPING_TARGETS"
elif closeout_receipts:
result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_READY"
elif artifact_count:
result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY"
else:
result = "WAITING_FOR_DIRECT_MAPPING_CANDIDATES"
return {
"policy": DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY,
"result": result,
"success": bool(auto_resolution.get("success")),
"generated_at": auto_resolution.get("generated_at"),
"source_policy": auto_resolution.get("policy"),
"stats": auto_resolution.get("stats") or {},
"backlog": auto_resolution.get("backlog") or {},
"summary": {
"direct_mapping_count": int((auto_resolution.get("summary") or {}).get("direct_mapping_count") or 0),
"selected_direct_mapping_count": selected_direct_count,
"candidate_decision_count": int((auto_resolution.get("summary") or {}).get("candidate_decision_count") or 0),
"machine_review_exception_receipt_count": int(
(auto_resolution.get("summary") or {}).get("machine_review_exception_receipt_count") or 0
),
"exception_auto_resolution_artifact_count": artifact_count,
**closeout_summary,
},
"closeout_package": {
"stage": "P2_machine_verifiable_exception_resolution_closeout",
"execute_search": bool(execute_search),
"execute_retry_search": bool(execute_retry_search),
"closeout_receipts": closeout_receipts,
"resolution_mode": "ai_controlled_read_only",
},
"upstream_auto_resolution_summary": auto_resolution.get("summary") or {},
"next_actions": [
"Feed retry_search candidates back into the candidate decision package when retry_candidate_count is nonzero.",
"Feed evidence_delta closeouts into no-write verifier receipts before any persistence.",
"Keep database writes behind controlled apply, rollback, and production readback.",
],
"safety": {
"read_only_preview": True,
"executes_search": bool(execute_search),
"executes_retry_search": bool(execute_retry_search),
"writes_database": False,
"persists_candidate": False,
"syncs_external_offers": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"gemini_allowed": False,
"requires_production_version_truth": True,
},
}
def build_pchome_evidence_enrichment_preview(payload: dict[str, Any], batch_size: int = 5) -> dict[str, Any]:
"""Build a read-only evidence enrichment package for mapping targets."""
operator_preview = build_pchome_mapping_operator_preview(payload, batch_size=batch_size)
@@ -2599,6 +2883,19 @@ def build_pchome_growth_ai_automation_readiness(
search_summary = decision_package.get("upstream_search_summary") or {}
decision_summary = decision_package.get("summary") or {}
receipt_summary = receipt_gate.get("summary") or {}
exception_artifacts = list(
(decision_package.get("decision_package") or {}).get(
"machine_review_exception_auto_resolution_artifacts"
)
or []
)
exception_closeout_receipts = _build_candidate_exception_resolution_closeout_receipts(
exception_artifacts,
execute_retry_search=False,
)
exception_closeout_summary = _summarize_exception_resolution_closeout_receipts(
exception_closeout_receipts
)
direct_mapping_count = int(backlog.get("direct_mapping_count") or 0)
selected_search_targets = int(search_summary.get("selected_direct_mapping_count") or 0)
@@ -2610,6 +2907,9 @@ def build_pchome_growth_ai_automation_readiness(
exception_auto_resolution_artifact_count = int(
decision_summary.get("exception_auto_resolution_artifact_count") or 0
)
exception_resolution_closeout_receipt_count = int(
exception_closeout_summary.get("exception_resolution_closeout_receipt_count") or 0
)
retry_search_action_count = int(decision_summary.get("retry_search_action_count") or 0)
waiting_candidate_count = selected_search_targets if not candidate_decision_count else 0
receipt_count = int(receipt_summary.get("receipt_count") or 0)
@@ -2623,6 +2923,7 @@ def build_pchome_growth_ai_automation_readiness(
"ai_exception_count": exception_count,
"exception_receipt_count": exception_receipt_count,
"exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count,
"exception_resolution_closeout_receipt_count": exception_resolution_closeout_receipt_count,
"retry_search_action_count": retry_search_action_count,
"routes": [
{
@@ -2641,6 +2942,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 exception_resolution_closeout_receipt_count:
result = "AI_AUTOMATION_EXCEPTION_RESOLUTION_CLOSEOUT_READY"
elif exception_auto_resolution_artifact_count:
result = "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY"
elif candidate_decision_count:
@@ -2685,6 +2988,14 @@ def build_pchome_growth_ai_automation_readiness(
f"{retry_search_action_count} 組 retry search 動作",
"執行變體/組合判別、命名證據差分與單位基準擴搜尋",
),
_automation_lane(
"candidate_exception_resolution_closeout",
"候選例外解法收斂",
"ready" if exception_resolution_closeout_receipt_count else "waiting",
exception_resolution_closeout_receipt_count,
f"{exception_closeout_summary.get('retry_search_ready_count', 0)} 筆可進 retry search",
"把 resolver artifacts 收斂成 retry search / verifier receipts",
),
_automation_lane(
"evidence_receipts",
"證據收據",
@@ -2718,7 +3029,15 @@ def build_pchome_growth_ai_automation_readiness(
"machine_review_decision_count": int(decision_summary.get("machine_review_decision_count") or 0),
"machine_review_exception_receipt_count": exception_receipt_count,
"exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count,
"exception_resolution_closeout_receipt_count": exception_resolution_closeout_receipt_count,
"retry_search_action_count": retry_search_action_count,
"retry_search_ready_count": int(exception_closeout_summary.get("retry_search_ready_count") or 0),
"retry_search_executed_count": int(exception_closeout_summary.get("retry_search_executed_count") or 0),
"retry_candidate_count": int(exception_closeout_summary.get("retry_candidate_count") or 0),
"evidence_delta_closeout_count": int(exception_closeout_summary.get("evidence_delta_closeout_count") or 0),
"ready_for_next_candidate_decision_count": int(
exception_closeout_summary.get("ready_for_next_candidate_decision_count") or 0
),
"variant_bundle_discriminator_count": int(
decision_summary.get("variant_bundle_discriminator_count") or 0
),