Attach decision envelopes to review queue

services/competitor_intel_repository.py

@@ -295,6 +295,145 @@ def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str
         return {"comparable": False, "reason": "build_error"}
 
 
+def _review_action_code(attempt_status: str) -> str:
+    if attempt_status == "rescore_accepted_current":
+        return "review_accept_identity"
+    if attempt_status in UNIT_COMPARABLE_STATUSES or attempt_status == "manual_unit_price_required":
+        return "unit_price_required"
+    if attempt_status in {"no_result", "refresh_no_result", "manual_needs_research"}:
+        return "needs_research"
+    if attempt_status in {"identity_veto", "manual_rejected"}:
+        return "verify_or_reject_identity"
+    if attempt_status in {"expired_match", "protected_existing_match"}:
+        return "refresh_or_compare_identity"
+    return "human_review"
+
+
+def _review_data_quality(attempt_status: str, item: dict[str, Any]) -> str:
+    if attempt_status in {"no_result", "refresh_no_result", "never_attempted"}:
+        return "missing"
+    if not item.get("candidate_pc_id") or not item.get("candidate_pc_name"):
+        return "missing"
+    if item.get("candidate_pc_price", 0) <= 0 or item.get("momo_price", 0) <= 0:
+        return "partial"
+    if attempt_status == "rescore_accepted_current":
+        return "complete"
+    return "partial"
+
+
+def _review_severity(attempt_status: str, item: dict[str, Any]) -> str:
+    momo_price = _num(item.get("momo_price"))
+    candidate_price = _num(item.get("candidate_pc_price"))
+    price_gap_pct = 0.0
+    if momo_price > 0 and candidate_price > 0:
+        price_gap_pct = (momo_price - candidate_price) / max(candidate_price, 1) * 100
+
+    if attempt_status == "rescore_accepted_current" and price_gap_pct >= 10:
+        return "P1"
+    if attempt_status == "rescore_accepted_current":
+        return "P2"
+    if attempt_status in UNIT_COMPARABLE_STATUSES:
+        return "P2"
+    if attempt_status in {"recoverable_low_score", "expired_match"}:
+        return "P3"
+    return "P4"
+
+
+def _build_review_decision_envelope(item: dict[str, Any]) -> dict[str, Any]:
+    """Build the shared evidence contract for an operator review queue item."""
+    attempt_status = str(item.get("attempt_status") or "")
+    momo_price = _num(item.get("momo_price"))
+    candidate_price = _num(item.get("candidate_pc_price"))
+    gap_amount = None
+    gap_pct = None
+    if momo_price > 0 and candidate_price > 0:
+        gap_amount = round(momo_price - candidate_price, 2)
+        gap_pct = round((momo_price - candidate_price) / max(candidate_price, 1) * 100, 1)
+
+    evidence: list[dict[str, Any]] = [
+        {
+            "type": "review_status",
+            "metric": "attempt_status",
+            "value": attempt_status,
+            "basis": item.get("status_label") or _attempt_status_label(attempt_status),
+        },
+        {
+            "type": "match",
+            "metric": "match_score",
+            "value": round(_num(item.get("best_match_score")), 3),
+            "basis": "/".join(
+                part
+                for part in (
+                    item.get("match_type") or "unknown",
+                    item.get("price_basis") or "unknown",
+                    item.get("alert_tier") or "unknown",
+                )
+                if part
+            ),
+            "confidence": round(_num(item.get("best_match_score")), 3) or None,
+        },
+    ]
+    if gap_pct is not None:
+        evidence.append({
+            "type": "price",
+            "metric": "candidate_gap_pct",
+            "value": f"{gap_pct:+.1f}%",
+            "basis": "MOMO latest price + PChome review candidate",
+        })
+    reason_text = item.get("diagnostic_reason_text")
+    if reason_text:
+        evidence.append({
+            "type": "diagnostic",
+            "metric": "reasons",
+            "value": reason_text,
+            "basis": "match_diagnostic_json.reasons",
+        })
+
+    return {
+        "decision_id": (
+            "review_queue:"
+            f"{item.get('sku') or 'unknown'}:"
+            f"{attempt_status or 'unknown'}:"
+            f"{item.get('candidate_pc_id') or 'no_candidate'}"
+        ),
+        "source_agent": "review_queue",
+        "decision_type": "pchome_match_review",
+        "severity": _review_severity(attempt_status, item),
+        "subject": {
+            "sku": str(item.get("sku") or ""),
+            "name": item.get("name") or "",
+            "event_type": "pchome_match_review",
+            "competitor_product_id": item.get("candidate_pc_id") or "",
+            "competitor_product_name": item.get("candidate_pc_name") or "",
+        },
+        "evidence": evidence,
+        "recommended_action": {
+            "action": _review_action_code(attempt_status),
+            "owner": "營運",
+            "requires_hitl": True,
+        },
+        "expected_impact": {
+            "gap_amount": gap_amount,
+            "candidate_gap_pct": gap_pct,
+            "risk_reduction": "medium" if attempt_status in {"rescore_accepted_current", "recoverable_low_score"} else "watch",
+        },
+        "confidence": round(_num(item.get("best_match_score")), 3),
+        "guardrails": {
+            "can_auto_execute": False,
+            "blocked_reason": "PChome 候選需人工覆核；不得自動寫入正式 competitor_prices",
+            "data_quality": _review_data_quality(attempt_status, item),
+            "attempt_status": attempt_status,
+            "match_type": item.get("match_type") or "",
+            "price_basis": item.get("price_basis") or "",
+            "alert_tier": item.get("alert_tier") or "",
+        },
+        "trace": {
+            "source": "competitor_match_attempts",
+            "attempted_at": item.get("attempted_at") or "",
+        },
+    }
+
+
 def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
     item = dict(row)
     unit_comparison = _build_unit_comparison_for_attempt(item)
@@ -310,7 +449,7 @@ def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
     alert_tier = diagnostic_payload.get("alert_tier") or _tag_suffix(tags, "alert_tier") or ""
     evidence_flags = diagnostic_payload.get("evidence_flags") or []
     diagnostic_reasons = _extract_match_diagnostic_reasons(match_diagnostic, diagnostic_payload)
-    return {
+    formatted = {
         "sku": str(item.get("sku") or ""),
         "name": item.get("name") or "",
         "category": item.get("category") or "",
@@ -336,6 +475,8 @@ def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
         "attempted_at": _date_label(item.get("attempted_at")),
         "unit_comparison": unit_comparison,
     }
+    formatted["decision_envelope"] = _build_review_decision_envelope(formatted)
+    return formatted
 
 
 def clear_competitor_intel_cache() -> None:
@@ -791,7 +932,7 @@ def fetch_competitor_review_queue(engine, limit: int = 12) -> list[dict]:
     """可行動的 PChome 比對覆核隊列，供 Dashboard / AI / PPT 共用。"""
     limit = max(1, min(int(limit or 12), 50))
     return _cached_payload(
-        f"review_queue:v2:limit={limit}:floor={PCHOME_MATCH_SCORE_FLOOR}",
+        f"review_queue:v3:limit={limit}:floor={PCHOME_MATCH_SCORE_FLOOR}",
         lambda: _fetch_competitor_review_queue_uncached(engine, limit=limit),
     )
 
@@ -814,7 +955,7 @@ def fetch_competitor_review_queue_page(
     if status_filter not in REVIEW_STATUS_FILTER_GROUPS:
         status_filter = ""
     cache_key = (
-        "review_queue_page:v2:"
+        "review_queue_page:v3:"
         f"page={page}:per={per_page}:q={search_query.lower()}:cat={category}:"
         f"status={status_filter}:"
         f"count={int(bool(count_total))}:"