新增 PChome 覆核匯出與診斷原因

services/competitor_intel_repository.py

@@ -59,6 +59,20 @@ ATTEMPT_ACTION_LABELS = {
     "no_result": "補充搜尋詞或品牌關鍵字",
     "never_attempted": "排入 PChome 補抓",
 }
+MATCH_DIAGNOSTIC_REASON_LABELS = {
+    "brand_conflict": "品牌不符",
+    "product_line_conflict": "商品線不符",
+    "type_conflict": "品類不符",
+    "volume_conflict": "容量差異",
+    "weight_conflict": "重量差異",
+    "count_conflict": "件數差異",
+    "component_count_conflict": "入數差異",
+    "multi_component_conflict": "組合差異",
+    "refill_pack_conflict": "補充包差異",
+    "unit_comparable": "需單位價",
+    "price_ratio_extreme": "價差極端",
+    "price_ratio_wide": "價差過大",
+}
 COMPETITOR_INTEL_CACHE_TTL_SECONDS = int(os.getenv("COMPETITOR_INTEL_CACHE_TTL_SECONDS", "1800"))
 _BASE_DIR = Path(__file__).resolve().parents[1]
 _CACHE_FILE = _BASE_DIR / "data" / "competitor_intel_cache.pkl"
@@ -93,6 +107,35 @@ def _attempt_action_label(status: Any) -> str:
     return ATTEMPT_ACTION_LABELS.get(str(status or ""), "人工確認比對證據")
 
 
+def _extract_match_diagnostic_reasons(diagnostic_text: Any) -> list[dict[str, str]]:
+    """Translate matcher diagnostics into short operator-facing reason chips."""
+    text_value = str(diagnostic_text or "")
+    if not text_value:
+        return []
+
+    reason_blob = ""
+    for part in text_value.split(";"):
+        key, _, value = part.strip().partition("=")
+        if key.strip() == "reasons":
+            reason_blob = value.strip()
+            break
+    if not reason_blob:
+        return []
+
+    reasons: list[dict[str, str]] = []
+    seen: set[str] = set()
+    for raw_reason in reason_blob.replace("|", ",").split(","):
+        code = raw_reason.strip()
+        if not code or code in seen:
+            continue
+        seen.add(code)
+        reasons.append({
+            "code": code,
+            "label": MATCH_DIAGNOSTIC_REASON_LABELS.get(code, code.replace("_", " ")),
+        })
+    return reasons
+
+
 def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str, Any]]:
     status = str(row.get("attempt_status") or "")
     if status not in UNIT_COMPARABLE_STATUSES:
@@ -112,6 +155,8 @@ def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str
 def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
     item = dict(row)
     unit_comparison = _build_unit_comparison_for_attempt(item)
+    match_diagnostic = item.get("error_message") or ""
+    diagnostic_reasons = _extract_match_diagnostic_reasons(match_diagnostic)
     return {
         "sku": str(item.get("sku") or ""),
         "name": item.get("name") or "",
@@ -125,7 +170,9 @@ def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
         "candidate_pc_name": item.get("best_competitor_product_name") or "",
         "candidate_pc_price": _num(item.get("best_competitor_price")),
         "best_match_score": _num(item.get("best_match_score")),
-        "match_diagnostic": item.get("error_message") or "",
+        "match_diagnostic": match_diagnostic,
+        "diagnostic_reasons": diagnostic_reasons,
+        "diagnostic_reason_text": "、".join(reason["label"] for reason in diagnostic_reasons),
         "attempted_at": _date_label(item.get("attempted_at")),
         "unit_comparison": unit_comparison,
     }