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

修正 PChome AI 候選決策就緒狀態
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-01 18:33:29 +08:00
parent 08d9e3fe7d
commit 56d167f15a
2 changed files with 86 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
services/pchome_mapping_backlog_service.py
View File

@@ -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: