wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions 1 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:38:18 +08:00
parent b421e2b26d
commit 36d1d5c78d
3 changed files with 467 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
routes/ai_routes.py
View File

@@ -1864,6 +1864,59 @@ def api_pchome_growth_direct_mapping_candidate_decision_package():
}), 500
@ai_bp.route('/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-exception-auto-resolution-package')
@login_required
def api_pchome_growth_direct_mapping_candidate_exception_auto_resolution_package():
"""P2 no-write auto-resolution package for direct mapping candidate exceptions."""
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_candidate_exception_auto_resolution_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'}
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_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,
)
package["source_endpoint"] = (
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-package"
)
return jsonify(package)
except Exception as exc:
logger.error("[PChomeGrowth] direct mapping candidate exception auto-resolution package 讀取失敗: %s", exc, exc_info=True)
return jsonify({
"success": False,
"error": "PChome 商品對應候選例外自動解法暫時無法讀取，請稍後再試。",
}), 500
@ai_bp.route('/api/ai/pchome-growth/ai-automation-readiness')
@login_required
def api_pchome_growth_ai_automation_readiness():

296
services/pchome_mapping_backlog_service.py
View File

@@ -38,6 +38,9 @@ DIRECT_MAPPING_AUTO_SEARCH_PACKAGE_POLICY = (
DIRECT_MAPPING_CANDIDATE_DECISION_PACKAGE_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_decision_package"
)
DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY = (
"read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
)
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"
@@ -1455,6 +1458,182 @@ def _build_candidate_exception_receipt(decision: dict[str, Any]) -> dict[str, An
}
def _unit_basis_search_terms_from_subject(subject: dict[str, Any], max_terms: int = 8) -> list[str]:
pchome_name = str(subject.get("pchome_product_name") or "").strip()
momo_name = str(subject.get("momo_product_name") or "").strip()
source_name = pchome_name or momo_name
terms = _build_direct_mapping_search_terms(source_name, max_terms=3)
basis = parse_unit_package_basis(source_name) if source_name else {}
unit_measure = basis.get("unit_pricing_measure") or {}
base_measure = basis.get("unit_pricing_base_measure") or {}
quantity_terms: list[str] = []
if unit_measure.get("value") and unit_measure.get("unit"):
quantity_terms.append(f"{unit_measure.get('value'):g}{unit_measure.get('unit')}")
if base_measure.get("value") and base_measure.get("unit"):
quantity_terms.append(f"每{base_measure.get('value'):g}{base_measure.get('unit')}")
if basis.get("estimated_total_quantity") and basis.get("unit_label"):
quantity_terms.append(f"{basis.get('estimated_total_quantity'):g}{basis.get('unit_label')}")
if basis.get("multipliers"):
quantity_terms.append("x".join(str(item) for item in basis.get("multipliers") or []))
for quantity_term in quantity_terms:
if source_name:
terms.append(f"{source_name} {quantity_term}")
if source_name and ("bundle_or_promo" in (basis.get("risk_signals") or [])):
terms.append(f"{source_name} 單入")
if source_name and quantity_terms:
terms.append(f"{source_name} 單位價")
return list(dict.fromkeys(term for term in terms if term))[:max_terms]
def _build_variant_bundle_discriminator(subject: dict[str, Any], failure_reasons: list[str]) -> dict[str, Any]:
pchome_name = str(subject.get("pchome_product_name") or "").strip()
momo_name = str(subject.get("momo_product_name") or "").strip()
target_basis = parse_unit_package_basis(pchome_name) if pchome_name else {}
momo_basis = parse_unit_package_basis(momo_name) if momo_name else {}
signals = set(target_basis.get("risk_signals") or [])
signals.update(momo_basis.get("risk_signals") or [])
if "auto_compare_type_not_receipt_ready" in failure_reasons:
signals.add("needs_auto_compare_type_resolution")
if "target_hard_veto_true" in failure_reasons:
signals.add("target_identity_veto_blocks_receipt")
return {
"resolver": "variant_bundle_discriminator",
"decision": "blocks_no_write_receipt_until_resolved" if signals else "identity_delta_only",
"risk_signals": sorted(signals),
"target_unit_package_basis": target_basis,
"momo_unit_package_basis": momo_basis,
"checks": [
"same_brand_or_named_line",
"same_quantity_or_convertible_unit_basis",
"same_variant_or_color_scope",
"same_bundle_or_single_item_scope",
],
"writes_database": False,
}
def _build_named_candidate_evidence_delta(subject: dict[str, Any], failure_reasons: list[str]) -> dict[str, Any]:
evidence_keys = [
"target_pchome_product_id",
"pchome_product_name",
"momo_product_id",
"momo_product_name",
"confidence",
"failure_reasons",
]
missing_keys = [
key
for key in ("target_pchome_product_id", "pchome_product_name", "momo_product_id", "momo_product_name")
if not subject.get(key)
]
return {
"resolver": "named_candidate_evidence_delta",
"named_evidence_keys": evidence_keys,
"missing_evidence_keys": missing_keys,
"failure_reasons": failure_reasons,
"resolution": "ready_for_retry_search" if not missing_keys else "drop_incomplete_candidate_and_retry_search",
"writes_database": False,
}
def _build_candidate_exception_auto_resolution_artifact(receipt: dict[str, Any]) -> dict[str, Any]:
subject = receipt.get("subject") or {}
failure_reasons = list(receipt.get("failure_reasons") or [])
next_actions = list(receipt.get("next_machine_actions") or [])
artifact_basis = {
"receipt_id": receipt.get("receipt_id"),
"failure_reasons": failure_reasons,
"next_actions": next_actions,
"subject": subject,
}
artifact_hash = hashlib.sha256(
json.dumps(artifact_basis, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
).hexdigest()
resolvers: dict[str, Any] = {}
if "run_variant_bundle_discriminator" in next_actions:
resolvers["variant_bundle_discriminator"] = _build_variant_bundle_discriminator(
subject,
failure_reasons,
)
if "build_named_candidate_evidence_delta" in next_actions:
resolvers["named_candidate_evidence_delta"] = _build_named_candidate_evidence_delta(
subject,
failure_reasons,
)
if "expand_search_terms_with_unit_basis" in next_actions:
resolvers["unit_basis_search_expansion"] = {
"resolver": "unit_basis_search_expansion",
"expanded_search_terms": _unit_basis_search_terms_from_subject(subject),
"retry_package": "direct_mapping_auto_search_package",
"writes_database": False,
}
if "expand_search_terms_with_brand_spec_anchors" in next_actions:
resolvers["brand_spec_search_expansion"] = {
"resolver": "brand_spec_search_expansion",
"expanded_search_terms": _build_direct_mapping_search_terms(
str(subject.get("pchome_product_name") or subject.get("momo_product_name") or ""),
max_terms=6,
),
"retry_package": "direct_mapping_auto_search_package",
"writes_database": False,
}
return {
"artifact_id": f"pchome-direct-mapping-exception-resolution-{artifact_hash[:16]}",
"source_receipt_id": receipt.get("receipt_id"),
"source_decision_id": receipt.get("source_decision_id"),
"stage": "P2_machine_verifiable_exception_auto_resolution",
"subject": subject,
"failure_reasons": failure_reasons,
"machine_actions": next_actions,
"resolvers": resolvers,
"resolution_status": "AUTO_RESOLUTION_PLANNED",
"next_package": "direct_mapping_candidate_decision_package_after_retry",
"guardrails": {
"machine_actionable": True,
"can_auto_execute_read_only": True,
"writes_database": False,
"persists_candidate": False,
"requires_verifier_before_persistence": True,
},
}
def _build_candidate_exception_auto_resolution_artifacts(
exception_receipts: list[dict[str, Any]],
) -> list[dict[str, Any]]:
return [
_build_candidate_exception_auto_resolution_artifact(receipt)
for receipt in exception_receipts
]
def _summarize_exception_auto_resolution_artifacts(artifacts: list[dict[str, Any]]) -> dict[str, int]:
resolver_counts = {
"variant_bundle_discriminator_count": 0,
"named_candidate_evidence_delta_count": 0,
"unit_basis_search_expansion_count": 0,
"brand_spec_search_expansion_count": 0,
}
retry_search_action_count = 0
for artifact in artifacts:
resolvers = artifact.get("resolvers") or {}
for key in resolver_counts:
resolver_name = key.removesuffix("_count")
if resolver_name in resolvers:
resolver_counts[key] += 1
if any(key.endswith("search_expansion") for key in resolvers):
retry_search_action_count += 1
return {
"exception_auto_resolution_artifact_count": len(artifacts),
"retry_search_action_count": retry_search_action_count,
**resolver_counts,
"writes_database_count": 0,
}
def build_pchome_direct_mapping_auto_search_package(
payload: dict[str, Any],
batch_size: int = 5,
@@ -1643,6 +1822,12 @@ def build_pchome_direct_mapping_candidate_decision_package(
_build_candidate_exception_receipt(envelope)
for envelope in machine_review_decisions
]
exception_auto_resolution_artifacts = _build_candidate_exception_auto_resolution_artifacts(
machine_review_exception_receipts
)
exception_auto_resolution_summary = _summarize_exception_auto_resolution_artifacts(
exception_auto_resolution_artifacts
)
if not int((search_package.get("summary") or {}).get("selected_direct_mapping_count") or 0):
result = "NO_DIRECT_MAPPING_TARGETS"
@@ -1669,6 +1854,7 @@ def build_pchome_direct_mapping_candidate_decision_package(
"auto_compare_decision_count": len(auto_compare_decisions),
"machine_review_decision_count": len(machine_review_decisions),
"machine_review_exception_receipt_count": len(machine_review_exception_receipts),
**exception_auto_resolution_summary,
"can_auto_persist_now_count": 0,
"writes_database_count": 0,
"persists_candidate_count": 0,
@@ -1678,6 +1864,7 @@ def build_pchome_direct_mapping_candidate_decision_package(
"execute_search": bool(execute_search),
"candidate_decisions": decision_envelopes,
"machine_review_exception_receipts": machine_review_exception_receipts,
"machine_review_exception_auto_resolution_artifacts": exception_auto_resolution_artifacts,
"manual_review_mode": "exception_only",
},
"decision_acceptance_policy": {
@@ -1694,7 +1881,85 @@ def build_pchome_direct_mapping_candidate_decision_package(
"next_actions": [
"Run controlled read-only search first when candidate_decision_count is zero.",
"Send auto-compare decisions to no-write receipt generation before any persistence.",
"Route machine-review decisions through exception receipts with named failure reasons and next machine actions.",
"Route machine-review decisions through exception receipts and auto-resolution artifacts.",
],
"safety": {
"read_only_preview": True,
"executes_search": bool(execute_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_direct_mapping_candidate_exception_auto_resolution_package(
payload: dict[str, Any],
batch_size: int = 5,
*,
execute_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]:
"""Build a no-write auto-resolution package for machine-review candidate exceptions."""
decision_package = build_pchome_direct_mapping_candidate_decision_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 = decision_package.get("decision_package") or {}
exception_receipts = list(package.get("machine_review_exception_receipts") or [])
artifacts = list(package.get("machine_review_exception_auto_resolution_artifacts") or [])
summary = _summarize_exception_auto_resolution_artifacts(artifacts)
selected_direct_count = int((decision_package.get("summary") or {}).get("selected_direct_mapping_count") or 0)
candidate_decision_count = int((decision_package.get("summary") or {}).get("candidate_decision_count") or 0)
if not selected_direct_count:
result = "NO_DIRECT_MAPPING_TARGETS"
elif artifacts:
result = "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY"
elif candidate_decision_count:
result = "DIRECT_MAPPING_CANDIDATE_EXCEPTIONS_CLEAR"
else:
result = "WAITING_FOR_DIRECT_MAPPING_CANDIDATES"
return {
"policy": DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_POLICY,
"result": result,
"success": bool(decision_package.get("success")),
"generated_at": decision_package.get("generated_at"),
"source_policy": decision_package.get("policy"),
"stats": decision_package.get("stats") or {},
"backlog": decision_package.get("backlog") or {},
"summary": {
"direct_mapping_count": int((decision_package.get("summary") or {}).get("direct_mapping_count") or 0),
"selected_direct_mapping_count": selected_direct_count,
"candidate_decision_count": candidate_decision_count,
"machine_review_exception_receipt_count": len(exception_receipts),
**summary,
},
"auto_resolution_package": {
"stage": "P2_machine_verifiable_exception_auto_resolution",
"execute_search": bool(execute_search),
"exception_receipts": exception_receipts,
"auto_resolution_artifacts": artifacts,
"resolution_mode": "ai_controlled_read_only",
},
"upstream_decision_summary": decision_package.get("summary") or {},
"next_actions": [
"Use variant_bundle_discriminator and named_candidate_evidence_delta before retry search.",
"Use unit_basis_search_expansion for hard-veto unit candidates before another candidate decision package.",
"Only route resolved candidates to no-write receipt and verifier packages before persistence.",
],
"safety": {
"read_only_preview": True,
@@ -2330,6 +2595,10 @@ def build_pchome_growth_ai_automation_readiness(
exception_receipt_count = int(
decision_summary.get("machine_review_exception_receipt_count") or 0
)
exception_auto_resolution_artifact_count = int(
decision_summary.get("exception_auto_resolution_artifact_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)
ready_receipt_count = int(receipt_summary.get("ready_for_auto_persistence_count") or 0)
@@ -2341,11 +2610,13 @@ def build_pchome_growth_ai_automation_readiness(
PRIMARY_HUMAN_GATE_COUNT_KEY: 0,
"ai_exception_count": exception_count,
"exception_receipt_count": exception_receipt_count,
"exception_auto_resolution_artifact_count": exception_auto_resolution_artifact_count,
"retry_search_action_count": retry_search_action_count,
"routes": [
{
"source": "candidate_decision_package",
"condition": "not_ready_for_no_write_receipt",
"auto_resolution": "build_failure_reasons_and_next_machine_action",
"auto_resolution": "build_exception_receipts_and_auto_resolution_artifacts",
},
{
"source": "evidence_receipts",
@@ -2358,6 +2629,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_auto_resolution_artifact_count:
result = "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY"
elif candidate_decision_count:
result = "AI_AUTOMATION_CANDIDATE_DECISIONS_READY"
elif direct_mapping_count and selected_search_targets:
@@ -2392,6 +2665,14 @@ def build_pchome_growth_ai_automation_readiness(
f"等待 {waiting_candidate_count} 筆候選" if waiting_candidate_count else "可輸出 decision envelope",
"將候選分流到 no-write receipt 或 AI 例外決策",
),
_automation_lane(
"candidate_exception_auto_resolution",
"候選例外自動解法",
"ready" if exception_auto_resolution_artifact_count else ("planned" if exception_receipt_count else "waiting"),
exception_auto_resolution_artifact_count,
f"{retry_search_action_count} 組 retry search 動作",
"執行變體/組合判別、命名證據差分與單位基準擴搜尋",
),
_automation_lane(
"evidence_receipts",
"證據收據",
@@ -2424,6 +2705,17 @@ def build_pchome_growth_ai_automation_readiness(
"auto_compare_decision_count": int(decision_summary.get("auto_compare_decision_count") or 0),
"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,
"retry_search_action_count": retry_search_action_count,
"variant_bundle_discriminator_count": int(
decision_summary.get("variant_bundle_discriminator_count") or 0
),
"named_candidate_evidence_delta_count": int(
decision_summary.get("named_candidate_evidence_delta_count") or 0
),
"unit_basis_search_expansion_count": int(
decision_summary.get("unit_basis_search_expansion_count") or 0
),
"receipt_count": receipt_count,
"ready_receipt_count": ready_receipt_count,
"exception_count": exception_count,

120
tests/test_pchome_mapping_backlog_report.py
View File

@@ -72,6 +72,7 @@ from services.pchome_mapping_backlog_service import (
build_pchome_evidence_source_preview,
build_pchome_direct_mapping_auto_search_package,
build_pchome_direct_mapping_candidate_decision_package,
build_pchome_direct_mapping_candidate_exception_auto_resolution_package,
build_pchome_growth_ai_automation_readiness,
build_pchome_mapping_operator_preview,
parse_pchome_product_page_evidence_html,
@@ -409,6 +410,58 @@ def test_direct_mapping_candidate_decision_package_routes_candidates_to_machine_
assert package["safety"]["persists_candidate"] is False
def test_direct_mapping_candidate_exception_auto_resolution_builds_machine_artifacts():
def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score):
return True, "found", [
{
"product_id": "MOMO-VARIANT",
"name": "Direct mapping product 40ml 多款任選",
"price": 899,
"target_pchome_product_id": "PCH-2",
"target_pchome_name": "Direct mapping product 40ml x2",
"target_match_score": 0.51,
"auto_compare_type": "manual_review",
"target_hard_veto": False,
},
{
"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_candidate_exception_auto_resolution_package(
_payload(),
batch_size=1,
execute_search=True,
search_func=fake_search,
)
artifacts = package["auto_resolution_package"]["auto_resolution_artifacts"]
assert package["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
assert package["result"] == "DIRECT_MAPPING_CANDIDATE_EXCEPTION_AUTO_RESOLUTION_READY"
assert package["summary"]["machine_review_exception_receipt_count"] == 2
assert package["summary"]["exception_auto_resolution_artifact_count"] == 2
assert package["summary"]["variant_bundle_discriminator_count"] == 1
assert package["summary"]["named_candidate_evidence_delta_count"] == 1
assert package["summary"]["unit_basis_search_expansion_count"] == 1
assert package["summary"]["retry_search_action_count"] == 1
assert artifacts[0]["artifact_id"].startswith("pchome-direct-mapping-exception-resolution-")
assert artifacts[0]["resolvers"]["variant_bundle_discriminator"]["writes_database"] is False
assert artifacts[0]["resolvers"]["named_candidate_evidence_delta"]["resolution"] == "ready_for_retry_search"
assert "unit_basis_search_expansion" in artifacts[1]["resolvers"]
assert any("40ml" in term.lower() for term in artifacts[1]["resolvers"]["unit_basis_search_expansion"]["expanded_search_terms"])
assert artifacts[1]["guardrails"]["can_auto_execute_read_only"] is True
assert package["summary"]["writes_database_count"] == 0
assert package["safety"]["writes_database"] is False
assert package["safety"]["persists_candidate"] is False
def test_ai_automation_readiness_makes_automation_visible_without_manual_primary_flow():
readiness = build_pchome_growth_ai_automation_readiness(_payload(), batch_size=1)
@@ -476,6 +529,43 @@ def test_ai_automation_readiness_reports_candidate_decisions_after_controlled_se
assert call_count["search"] == 1
def test_ai_automation_readiness_reports_exception_auto_resolution_ready():
def fake_search(targets, limit_per_product, max_products, max_terms_per_product, min_score):
return True, "found", [
{
"product_id": "MOMO-VARIANT",
"name": "Direct mapping product 40ml 多款任選",
"price": 899,
"target_pchome_product_id": "PCH-2",
"target_pchome_name": "Direct mapping product 40ml x2",
"target_match_score": 0.51,
"auto_compare_type": "manual_review",
"target_hard_veto": False,
}
]
readiness = build_pchome_growth_ai_automation_readiness(
_payload(),
batch_size=1,
execute_search=True,
search_func=fake_search,
)
lanes = {lane["key"]: lane for lane in readiness["automation_lanes"]}
assert readiness["result"] == "AI_AUTOMATION_EXCEPTION_AUTO_RESOLUTION_READY"
assert readiness["summary"]["candidate_decision_count"] == 1
assert readiness["summary"]["machine_review_exception_receipt_count"] == 1
assert readiness["summary"]["exception_auto_resolution_artifact_count"] == 1
assert readiness["summary"]["variant_bundle_discriminator_count"] == 1
assert readiness["summary"]["named_candidate_evidence_delta_count"] == 1
assert readiness["ai_exception_auto_resolution"]["exception_auto_resolution_artifact_count"] == 1
assert lanes["candidate_exception_auto_resolution"]["status"] == "ready"
assert lanes["candidate_exception_auto_resolution"]["value"] == 1
assert readiness["summary"]["primary_human_gate_count"] == 0
assert readiness["summary"]["writes_database_count"] == 0
assert readiness["safety"]["writes_database"] is False
def test_unit_package_basis_parser_extracts_quantity_count_and_risk_signals():
single = parse_unit_package_basis("雅詩蘭黛 粉持久完美持妝粉底 30ml")
assert single["package_basis"] == "single_unit_quantity_candidate"
@@ -14604,6 +14694,36 @@ def test_direct_mapping_candidate_decision_package_route_defaults_to_no_search_a
assert payload["safety"]["writes_database"] is False
def test_direct_mapping_candidate_exception_auto_resolution_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 exception auto-resolution 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-candidate-exception-auto-resolution-package?batch_size=1"
):
response = routes.api_pchome_growth_direct_mapping_candidate_exception_auto_resolution_package.__wrapped__()
payload = response.get_json()
assert payload["success"] is True
assert payload["policy"] == "read_only_pchome_growth_direct_mapping_candidate_exception_auto_resolution"
assert payload["source_endpoint"] == (
"/api/ai/pchome-growth/mapping-backlog/direct-mapping-candidate-decision-package"
)
assert payload["result"] == "WAITING_FOR_DIRECT_MAPPING_CANDIDATES"
assert payload["summary"]["exception_auto_resolution_artifact_count"] == 0
assert payload["auto_resolution_package"]["resolution_mode"] == "ai_controlled_read_only"
assert payload["safety"]["executes_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