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

補齊 PChome retry 候選決策包
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-01 21:21:21 +08:00
parent 12e03ef64e
commit a9c2d4b628
3 changed files with 354 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
routes/ai_routes.py
View File

@@ -1972,6 +1972,61 @@ def api_pchome_growth_direct_mapping_candidate_exception_resolution_closeout_pac
}), 500
@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-decision-package')
@login_required
def api_pchome_growth_direct_mapping_retry_candidate_decision_package():
"""P2 no-write retry-candidate decision and verifier input package."""
try:
from config import DATABASE_PATH
from services.pchome_revenue_growth_service import build_pchome_growth_opportunities
from services.pchome_mapping_backlog_service import (
build_pchome_direct_mapping_retry_candidate_decision_package,
)
force_refresh = str(request.args.get('refresh') or '').strip().lower() in {'1', 'true', 'yes'}
execute_search = str(request.args.get('execute_search') or '').strip().lower() in {'1', 'true', 'yes'}
execute_retry_search = str(request.args.get('execute_retry_search') or '').strip().lower() in {'1', 'true', 'yes'}
limit = request.args.get('limit', 20, type=int)
batch_size = request.args.get('batch_size', 5, type=int)
limit_per_product = request.args.get('limit_per_product', 8, type=int)
max_terms_per_product = request.args.get('max_terms_per_product', 5, type=int)
min_score = request.args.get('min_score', 0.45, type=float)
limit = max(5, min(limit, 50))
payload = None
if not force_refresh:
payload = _get_cached_pchome_growth_payload()
if payload is None:
engine = _create_icaim_dashboard_engine(DATABASE_PATH)
try:
payload = build_pchome_growth_opportunities(engine, limit=limit)
finally:
engine.dispose()
payload["cache_state"] = "fresh"
_set_pchome_growth_cache(payload)
package = build_pchome_direct_mapping_retry_candidate_decision_package(
payload,
batch_size=batch_size,
execute_search=execute_search,
execute_retry_search=execute_retry_search,
limit_per_product=limit_per_product,
max_terms_per_product=max_terms_per_product,
min_score=min_score,
)
package["source_endpoint"] = (
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-resolution-closeout-package"
)
return jsonify(package)
except Exception as exc:
logger.error("[PChomeGrowth] direct mapping retry candidate decision package 讀取失敗: %s", exc, exc_info=True)
return jsonify({
"success": False,
"error": "PChome 商品對應 retry 候選決策包暫時無法讀取，請稍後再試。",
}), 500
@ai_bp.route('/api/ai/pchome-growth/ai-automation-readiness')
@login_required
def api_pchome_growth_ai_automation_readiness():

199
services/pchome_mapping_backlog_service.py
View File

@@ -44,6 +44,9 @@ DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
DIRECT_MAPPING_CANDIDATE_EXCEPTION_RESOLUTION_CLOSEOUT_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_resolution_closeout"
)
DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_retry_candidate_decision_package"
)
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"
@@ -1840,6 +1843,83 @@ def _summarize_exception_resolution_closeout_receipts(receipts: list[dict[str, A
}
def _retry_candidates_from_closeout_receipts(receipts: list[dict[str, Any]]) -> list[dict[str, Any]]:
retry_candidates: list[dict[str, Any]] = []
seen: set[str] = set()
for receipt in receipts:
retry_search = receipt.get("retry_search") or {}
for candidate in retry_search.get("candidates") or []:
candidate = dict(candidate)
candidate.setdefault("source_resolution_closeout_receipt_id", receipt.get("receipt_id"))
candidate.setdefault("source_resolution_artifact_id", receipt.get("source_artifact_id"))
dedupe_key = "|".join([
str(candidate.get("target_pchome_product_id") or ""),
str(candidate.get("product_id") or ""),
str(candidate.get("source_resolution_artifact_id") or ""),
])
if dedupe_key in seen:
continue
seen.add(dedupe_key)
retry_candidates.append(candidate)
return retry_candidates
def _build_no_write_auto_compare_verifier_receipt(decision: dict[str, Any]) -> dict[str, Any]:
subject = decision.get("subject") or {}
receipt_basis = {
"decision_id": decision.get("decision_id"),
"subject": subject,
"confidence": decision.get("confidence"),
"decision": decision.get("decision"),
}
receipt_hash = hashlib.sha256(
json.dumps(receipt_basis, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
).hexdigest()
checks = [
{
"check": "routes_to_no_write_auto_compare_receipt",
"passed": decision.get("decision") == "route_to_no_write_auto_compare_receipt",
},
{
"check": "target_pchome_product_id_present",
"passed": bool(subject.get("target_pchome_product_id")),
},
{
"check": "momo_product_id_present",
"passed": bool(subject.get("momo_product_id")),
},
{
"check": "target_match_score_present",
"passed": decision.get("confidence") not in (None, ""),
},
{
"check": "database_write_locked",
"passed": True,
},
]
ready = all(check["passed"] for check in checks)
return {
"receipt_id": f"pchome-direct-mapping-no-write-verifier-{receipt_hash[:16]}",
"source_decision_id": decision.get("decision_id"),
"stage": "P2_retry_candidate_no_write_verifier_input",
"receipt_status": "NO_WRITE_VERIFIER_INPUT_READY" if ready else "NO_WRITE_VERIFIER_INPUT_BLOCKED",
"subject": subject,
"confidence": decision.get("confidence"),
"verification_checks": checks,
"ready_for_no_write_verifier": ready,
"ready_for_controlled_apply": False,
"next_package": "auto_policy_db_apply_verifier_artifact_preview_after_no_write_receipt",
"guardrails": {
"machine_actionable": True,
"writes_database": False,
"persists_candidate": False,
"requires_no_write_receipt": True,
"requires_verifier_before_persistence": True,
"requires_rollback_and_readback": True,
},
}
def build_pchome_direct_mapping_auto_search_package(
payload: dict[str, Any],
batch_size: int = 5,
@@ -2271,6 +2351,125 @@ def build_pchome_direct_mapping_candidate_exception_resolution_closeout_package(
}
def build_pchome_direct_mapping_retry_candidate_decision_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]:
"""Route retry-search candidates back into machine decisions and no-write verifier receipts."""
closeout = build_pchome_direct_mapping_candidate_exception_resolution_closeout_package(
payload,
batch_size=batch_size,
execute_search=execute_search,
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_package = closeout.get("closeout_package") or {}
closeout_receipts = list(closeout_package.get("closeout_receipts") or [])
retry_candidates = _retry_candidates_from_closeout_receipts(closeout_receipts)
try:
effective_min_score = max(0.35, min(float(min_score), 0.95))
except (TypeError, ValueError):
effective_min_score = 0.45
retry_decisions = [
_build_candidate_decision_envelope(candidate, min_score=effective_min_score)
for candidate in retry_candidates
]
no_write_decisions = [
decision
for decision in retry_decisions
if decision.get("decision") == "route_to_no_write_auto_compare_receipt"
]
machine_review_decisions = [
decision
for decision in retry_decisions
if decision.get("decision") == "route_to_machine_review_decision"
]
no_write_verifier_receipts = [
_build_no_write_auto_compare_verifier_receipt(decision)
for decision in no_write_decisions
]
machine_review_exception_receipts = [
_build_candidate_exception_receipt(decision)
for decision in machine_review_decisions
]
selected_direct_count = int((closeout.get("summary") or {}).get("selected_direct_mapping_count") or 0)
if not selected_direct_count:
result = "NO_DIRECT_MAPPING_TARGETS"
elif retry_decisions:
result = "DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_READY"
elif int((closeout.get("summary") or {}).get("retry_search_ready_count") or 0):
result = "WAITING_FOR_RETRY_CANDIDATES"
else:
result = "WAITING_FOR_EXCEPTION_RESOLUTION_CLOSEOUT"
return {
"policy": DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_POLICY,
"result": result,
"success": bool(closeout.get("success")),
"generated_at": closeout.get("generated_at"),
"source_policy": closeout.get("policy"),
"stats": closeout.get("stats") or {},
"backlog": closeout.get("backlog") or {},
"summary": {
"direct_mapping_count": int((closeout.get("summary") or {}).get("direct_mapping_count") or 0),
"selected_direct_mapping_count": selected_direct_count,
"exception_resolution_closeout_receipt_count": int(
(closeout.get("summary") or {}).get("exception_resolution_closeout_receipt_count") or 0
),
"retry_candidate_count": len(retry_candidates),
"retry_candidate_decision_count": len(retry_decisions),
"retry_no_write_verifier_input_count": len(no_write_verifier_receipts),
"retry_machine_review_exception_count": len(machine_review_exception_receipts),
"ready_for_no_write_verifier_count": sum(
1 for receipt in no_write_verifier_receipts if receipt.get("ready_for_no_write_verifier")
),
"ready_for_controlled_apply_count": 0,
"writes_database_count": 0,
"persists_candidate_count": 0,
},
"retry_candidate_decision_package": {
"stage": "P2_retry_candidate_machine_decision",
"execute_search": bool(execute_search),
"execute_retry_search": bool(execute_retry_search),
"retry_candidates": retry_candidates,
"retry_candidate_decisions": retry_decisions,
"no_write_verifier_receipts": no_write_verifier_receipts,
"machine_review_exception_receipts": machine_review_exception_receipts,
"manual_review_mode": "exception_only",
},
"upstream_closeout_summary": closeout.get("summary") or {},
"next_actions": [
"Send ready no-write verifier receipts into verifier artifact preview before any persistence.",
"Route retry machine-review exceptions back through exception auto-resolution.",
"Only controlled apply can reduce direct_mapping_count after verifier, rollback, and production readback pass.",
],
"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)

100
tests/test_pchome_mapping_backlog_report.py
View File

@@ -74,6 +74,7 @@ from services.pchome_mapping_backlog_service import (
build_pchome_direct_mapping_candidate_decision_package,
build_pchome_direct_mapping_candidate_exception_auto_resolution_package,
build_pchome_direct_mapping_candidate_exception_resolution_closeout_package,
build_pchome_direct_mapping_retry_candidate_decision_package,
build_pchome_growth_ai_automation_readiness,
build_pchome_mapping_operator_preview,
parse_pchome_product_page_evidence_html,
@@ -524,6 +525,74 @@ def test_direct_mapping_candidate_exception_resolution_closeout_executes_retry_s
assert call_count["search"] == 2
def test_direct_mapping_retry_candidate_decision_package_routes_retry_candidates_to_verifier_inputs():
call_count = {"search": 0}
def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score):
call_count["search"] += 1
if targets[0].get("source_artifact_id"):
return True, "retry_found", [
{
"product_id": "MOMO-READY",
"name": "Direct mapping product 40ml 單入",
"price": 520,
"target_pchome_product_id": "PCH-2",
"target_pchome_name": "Direct mapping product 40ml x2",
"target_match_score": 0.88,
"auto_compare_type": "unit_price",
"target_hard_veto": False,
},
{
"product_id": "MOMO-REVIEW",
"name": "Direct mapping product 40ml 多款任選",
"price": 499,
"target_pchome_product_id": "PCH-2",
"target_pchome_name": "Direct mapping product 40ml x2",
"target_match_score": 0.72,
"auto_compare_type": "manual_review",
"target_hard_veto": False,
},
]
return True, "found", [
{
"product_id": "MOMO-UNIT",
"name": "Direct mapping product 40ml",
"price": 499,
"target_pchome_product_id": "PCH-2",
"target_pchome_name": "Direct mapping product 40ml x2",
"target_match_score": 0.91,
"auto_compare_type": "unit_price",
"target_hard_veto": True,
}
]
package = build_pchome_direct_mapping_retry_candidate_decision_package(
_payload(),
batch_size=1,
execute_search=True,
execute_retry_search=True,
max_terms_per_product=6,
search_func=fake_search,
)
verifier_receipts = package["retry_candidate_decision_package"]["no_write_verifier_receipts"]
exception_receipts = package["retry_candidate_decision_package"]["machine_review_exception_receipts"]
assert package["policy"] == "read_only_pchome_growth_direct_mapping_retry_candidate_decision_package"
assert package["result"] == "DIRECT_MAPPING_RETRY_CANDIDATE_DECISION_PACKAGE_READY"
assert package["summary"]["retry_candidate_count"] == 2
assert package["summary"]["retry_candidate_decision_count"] == 2
assert package["summary"]["retry_no_write_verifier_input_count"] == 1
assert package["summary"]["retry_machine_review_exception_count"] == 1
assert package["summary"]["ready_for_no_write_verifier_count"] == 1
assert verifier_receipts[0]["receipt_status"] == "NO_WRITE_VERIFIER_INPUT_READY"
assert verifier_receipts[0]["ready_for_no_write_verifier"] is True
assert verifier_receipts[0]["guardrails"]["writes_database"] is False
assert exception_receipts[0]["failure_reasons"] == ["auto_compare_type_not_receipt_ready"]
assert package["summary"]["writes_database_count"] == 0
assert package["safety"]["writes_database"] is False
assert call_count["search"] == 2
def test_ai_automation_readiness_makes_automation_visible_without_manual_primary_flow():
readiness = build_pchome_growth_ai_automation_readiness(_payload(), batch_size=1)
@@ -14822,6 +14891,37 @@ def test_direct_mapping_candidate_exception_resolution_closeout_route_uses_cache
assert payload["safety"]["writes_database"] is False
def test_direct_mapping_retry_candidate_decision_route_uses_cached_payload(monkeypatch):
from flask import Flask
from routes import ai_routes as routes
monkeypatch.setattr(routes, "_get_cached_pchome_growth_payload", lambda: _payload())
def fail_engine(database_path):
raise AssertionError("cached retry candidate decision package should not open a DB engine")
monkeypatch.setattr(routes, "_create_icaim_dashboard_engine", fail_engine)
app = Flask(__name__)
with app.test_request_context(
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-retry-candidate-decision-package?batch_size=1"
):
response = routes.api_pchome_growth_direct_mapping_retry_candidate_decision_package.__wrapped__()
payload = response.get_json()
assert payload["success"] is True
assert payload["policy"] == "read_only_pchome_growth_direct_mapping_retry_candidate_decision_package"
assert payload["source_endpoint"] == (
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-resolution-closeout-package"
)
assert payload["result"] == "WAITING_FOR_EXCEPTION_RESOLUTION_CLOSEOUT"
assert payload["summary"]["retry_candidate_decision_count"] == 0
assert payload["retry_candidate_decision_package"]["manual_review_mode"] == "exception_only"
assert payload["safety"]["executes_search"] is False
assert payload["safety"]["executes_retry_search"] is False
assert payload["safety"]["writes_database"] is False
def test_ai_automation_readiness_route_defaults_to_no_search_and_uses_cached_payload(monkeypatch):
from flask import Flask
from routes import ai_routes as routes