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

feat(ai): add automation surface summary
Some checks failed
CD Pipeline / deploy (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-02 18:47:00 +08:00
parent b69d173f18
commit 5b8e4b3568
10 changed files with 479 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

252
services/pchome_mapping_backlog_service.py
View File

@@ -93,6 +93,9 @@ DIRECT_MAPPING_RETRY_CANDIDATE_EXCEPTION_CONTROLLED_APPLY_ROLLBACK_EVIDENCE_POLI
"read_only_pchome_growth_direct_mapping_retry_candidate_exception_controlled_apply_rollback_evidence"
)
AI_AUTOMATION_READINESS_POLICY = "read_only_pchome_growth_ai_automation_readiness"
AI_AUTOMATION_SURFACE_SUMMARY_POLICY = (
"read_only_pchome_growth_ai_automation_surface_summary"
)
EVIDENCE_ENRICHMENT_PREVIEW_POLICY = "read_only_pchome_growth_evidence_enrichment_preview"
EVIDENCE_SOURCE_PREVIEW_POLICY = "read_only_pchome_growth_evidence_source_preview"
PRODUCT_PAGE_EVIDENCE_PARSER_POLICY = "read_only_pchome_product_page_evidence_parser"
@@ -7067,6 +7070,255 @@ def build_pchome_growth_ai_automation_readiness(
}
def _surface_signal(
key: str,
label: str,
value: str,
detail: str,
status: str,
next_machine_action: str,
) -> dict[str, Any]:
return {
"key": key,
"label": label,
"value": value,
"detail": detail,
"status": status,
"next_machine_action": next_machine_action,
}
def _count_surface_lanes(lanes: list[dict[str, Any]], statuses: set[str]) -> int:
return sum(1 for lane in lanes if str(lane.get("status") or "") in statuses)
def _safe_surface_lanes(lanes: list[dict[str, Any]]) -> list[dict[str, Any]]:
safe_lanes: list[dict[str, Any]] = []
for lane in lanes:
safe_lanes.append(
{
"key": lane.get("key") or "",
"label": lane.get("label") or "",
"status": lane.get("status") or "unknown",
"value": lane.get("value") or 0,
"detail": lane.get("detail") or "",
"next_machine_action": lane.get("next_action") or "",
"writes_database": bool(lane.get("writes_database")),
"ai_exception_mode": lane.get("ai_exception_mode")
or AI_EXCEPTION_MODE_MACHINE_VERIFIABLE,
}
)
return safe_lanes
def _pick_next_surface_machine_action(
summary: dict[str, Any],
lanes: list[dict[str, Any]],
) -> str:
if int(summary.get("controlled_apply_drift_count") or 0):
return "執行漂移恢復、重新回讀與 rollback path 比對"
if int(summary.get("exception_resolution_closeout_receipt_count") or 0):
return "執行 retry search 與 verifier receipts 收斂"
if int(summary.get("exception_auto_resolution_artifact_count") or 0):
return "收斂候選例外的命名證據、組合判別與單位基準"
if int(summary.get("candidate_decision_count") or 0):
return "把候選決策送入 no-write receipt 與證據 verifier"
if int(summary.get("selected_search_target_count") or 0):
return "執行 controlled read-only 同款搜尋候選"
for wanted_status in ("blocked", "blocked_until_verifier", "waiting", "planned"):
for lane in lanes:
if str(lane.get("status") or "") == wanted_status:
next_action = str(lane.get("next_action") or "").strip()
if next_action:
return next_action
return "持續回讀、漂移監控與可回滾驗證"
def build_pchome_growth_ai_automation_surface_summary(
readiness: dict[str, Any],
) -> dict[str, Any]:
"""Convert readiness internals into a first-viewport product summary."""
summary = readiness.get("summary") or {}
lanes = [lane for lane in readiness.get("automation_lanes") or [] if isinstance(lane, dict)]
safe_lanes = _safe_surface_lanes(lanes)
safe_lane_count = len(safe_lanes)
ready_lane_count = _count_surface_lanes(lanes, {"active", "ready", "completed"})
completed_lane_count = _count_surface_lanes(lanes, {"completed"})
blocked_lane_count = _count_surface_lanes(lanes, {"blocked", "blocked_until_verifier"})
primary_human_gate_count = int(summary.get(PRIMARY_HUMAN_GATE_COUNT_KEY) or 0)
writes_database_count = int(summary.get("writes_database_count") or 0)
direct_mapping_count = int(summary.get("direct_mapping_count") or 0)
selected_search_target_count = int(summary.get("selected_search_target_count") or 0)
candidate_decision_count = int(summary.get("candidate_decision_count") or 0)
ready_receipt_count = int(summary.get("ready_receipt_count") or 0)
receipt_count = int(summary.get("receipt_count") or 0)
replay_selector_count = int(
summary.get("controlled_apply_replay_selector_count") or 0
)
replay_readback_pass_count = int(
summary.get("controlled_apply_replay_readback_pass_count") or 0
)
receipt_materialized_count = int(
summary.get("controlled_apply_receipt_materialized_count") or 0
)
closeout_verified_count = int(
summary.get("controlled_apply_closeout_verified_count") or 0
)
drift_count = int(summary.get("controlled_apply_drift_count") or 0)
drift_verified_count = int(summary.get("controlled_apply_drift_verified_count") or 0)
exception_count = int(summary.get("ai_exception_count") or 0)
next_machine_action = _pick_next_surface_machine_action(summary, lanes)
if closeout_verified_count:
landed_value = "已完成回讀"
landed_detail = (
f"{replay_readback_pass_count}/{replay_selector_count} 筆受控落地已收斂，"
"持續漂移監控。"
)
landed_status = "good"
elif receipt_materialized_count:
landed_value = f"{receipt_materialized_count} 筆收據"
landed_detail = "落地收據已產生，等待同輪回讀完成。"
landed_status = "warn"
elif selected_search_target_count or candidate_decision_count or receipt_count:
landed_value = "AI 主流程接管"
landed_detail = (
f"{selected_search_target_count} 筆搜尋目標、"
f"{candidate_decision_count} 筆候選決策、{ready_receipt_count} 筆收據就緒。"
)
landed_status = "good"
else:
landed_value = "等待資料"
landed_detail = f"目前還有 {direct_mapping_count} 筆商品對應缺口等待自動化輸入。"
landed_status = "waiting"
if replay_readback_pass_count:
verified_value = f"{replay_readback_pass_count} 筆已回讀"
verified_detail = "受控落地、收據與正式資料回讀可追蹤。"
verified_status = "good" if not drift_count else "warn"
elif ready_receipt_count:
verified_value = f"{ready_receipt_count} 筆收據就緒"
verified_detail = "證據收據已可進 verifier / dry-run persistence。"
verified_status = "good"
elif ready_lane_count:
verified_value = f"{ready_lane_count}/{safe_lane_count} 條通過"
verified_detail = "安全 lane 已集中成同一個第一視窗摘要。"
verified_status = "warn" if blocked_lane_count else "good"
else:
verified_value = "等待回讀"
verified_detail = "等待下一批可驗證候選或證據收據。"
verified_status = "waiting"
if drift_count:
change_value = f"{drift_count} 筆漂移"
change_detail = "已進入自動恢復與 rollback path 比對。"
change_status = "bad"
elif exception_count:
change_value = f"{exception_count} 筆 AI 例外"
change_detail = "例外已轉成 machine-verifiable auto-resolution。"
change_status = "warn"
elif blocked_lane_count:
change_value = f"{blocked_lane_count} 條待驗證"
change_detail = "等待 verifier、rollback 與回讀證明補齊。"
change_status = "warn"
else:
change_value = "無漂移"
change_detail = (
f"漂移驗證 {drift_verified_count} 筆；異動只呈現營運狀態，證據留在證據層。"
)
change_status = "good"
signals = [
_surface_signal(
"automated-landing",
"已自動落地",
landed_value,
landed_detail,
landed_status,
next_machine_action,
),
_surface_signal(
"verified",
"已驗證",
verified_value,
verified_detail,
verified_status,
next_machine_action,
),
_surface_signal(
"change-state",
"異動狀態",
change_value,
change_detail,
change_status,
next_machine_action,
),
_surface_signal(
"next-machine-action",
"下一步",
next_machine_action,
"下一步只顯示可自動執行、可回讀、可回滾的機器動作。",
"good" if not drift_count else "warn",
next_machine_action,
),
]
return {
"policy": AI_AUTOMATION_SURFACE_SUMMARY_POLICY,
"success": bool(readiness.get("success")),
"result": readiness.get("result") or "AI_AUTOMATION_WAITING_FOR_GROWTH_INPUT",
"generated_at": readiness.get("generated_at"),
"summary": {
"safe_lane_count": safe_lane_count,
"ready_lane_count": ready_lane_count,
"completed_lane_count": completed_lane_count,
"blocked_lane_count": blocked_lane_count,
"direct_mapping_count": direct_mapping_count,
"selected_search_target_count": selected_search_target_count,
"candidate_decision_count": candidate_decision_count,
"ready_receipt_count": ready_receipt_count,
"controlled_apply_replay_selector_count": replay_selector_count,
"controlled_apply_replay_readback_pass_count": replay_readback_pass_count,
"controlled_apply_receipt_materialized_count": receipt_materialized_count,
"controlled_apply_closeout_verified_count": closeout_verified_count,
"controlled_apply_drift_count": drift_count,
"controlled_apply_drift_verified_count": drift_verified_count,
"ai_exception_count": exception_count,
PRIMARY_HUMAN_GATE_COUNT_KEY: primary_human_gate_count,
LEGACY_PRIMARY_FLOW_COUNT_KEY: 0,
"writes_database_count": writes_database_count,
"external_network_execute_count": int(
summary.get("external_network_execute_count") or 0
),
"next_machine_action": next_machine_action,
},
"golden_signals": signals,
"safe_automation_lanes": safe_lanes,
"surface_contract": {
"first_viewport_required": True,
"signal_keys": [
"automated-landing",
"verified",
"change-state",
"next-machine-action",
],
"raw_evidence_hidden_from_first_viewport": True,
"primary_flow": "ai_controlled",
"exception_resolution": "ai_machine_verifiable",
},
"safety": {
"read_only_preview": True,
"writes_database": False,
"writes_database_count": writes_database_count,
"persists_receipt": False,
"dispatches_telegram": False,
"llm_calls_in_preview": False,
"gemini_allowed": False,
PRIMARY_HUMAN_GATE_COUNT_KEY: primary_human_gate_count,
},
}
def _receipt_payload_hash(receipt: dict[str, Any]) -> str:
payload = {
"receipt_id": receipt.get("receipt_id") or "",