V10.582 強化比價通知與身份證據

services/competitor_price_feeder.py

@@ -832,6 +832,11 @@ def _product_id_key(product_id: str) -> str:
     return re.sub(r"[^A-Z0-9]", "", str(product_id or "").upper())
 
 
+def _candidate_match_name(product) -> str:
+    """Return the identity-rich PChome text used for scoring, falling back to display name."""
+    return (getattr(product, "match_name", None) or getattr(product, "name", None) or "").strip()
+
+
 def _find_best_match_detail(
     momo_name: str,
     pchome_products: list,
@@ -863,7 +868,7 @@ def _rank_match_details(
     for p in pchome_products:
         diagnostics = score_marketplace_match(
             momo_name,
-            p.name,
+            _candidate_match_name(p),
             momo_price=momo_price,
             competitor_price=getattr(p, "price", None),
         )

services/nemoton_dispatcher_service.py

@@ -543,6 +543,8 @@ def _build_price_decision_envelope(
             "sku": sku,
             "name": name,
             "event_type": "price_competition",
+            "momo_price": momo_value if momo_value > 0 else None,
+            "competitor_price": comp_value if comp_value > 0 else None,
             "competitor_product_id": competitor_product_id,
             "competitor_product_name": str(competitor_product_name or "")[:120],
         },
@@ -554,6 +556,10 @@ def _build_price_decision_envelope(
             "requires_hitl": True,
         },
         "expected_impact": {
+            "momo_price": momo_value if momo_value > 0 else None,
+            "competitor_price": comp_value if comp_value > 0 else None,
+            "candidate_gap_pct": round(gap_value, 1),
+            "sales_7d_delta_pct": round(sales_value, 1),
             "revenue_loss_7d": round(loss_value, 2),
             "gap_amount": gap_amount,
             "recommended_price": recommended_price,

services/pchome_crawler.py

@@ -39,6 +39,8 @@ class PChomeProduct:
     review_count: int         # 評論數
     is_on_sale: bool          # 是否特價中
     crawled_at: datetime      # 爬取時間
+    subtitle: str = ''        # PChome Nick / 副標，常含容量、入數與濃度
+    match_name: str = ''      # 給 matcher 使用的身份文字；UI/DB 顯示仍用 name
 
     def to_dict(self) -> dict:
         """轉換為字典"""
@@ -47,6 +49,17 @@ class PChomeProduct:
         return data
 
 
+def _build_match_name(name: str, subtitle: str) -> str:
+    """Build an identity-rich title without duplicating the PChome display name."""
+    display_name = str(name or '').strip()
+    nick = str(subtitle or '').strip()
+    if not nick or nick == display_name:
+        return display_name
+    if display_name and nick.startswith(display_name):
+        return nick
+    return f"{display_name} {nick}".strip()
+
+
 class PChomeCrawler:
     """PChome 24h 爬蟲"""
 
@@ -334,9 +347,12 @@ class PChomeCrawler:
 
             image_url = f"{self.IMAGE_BASE_URL}{pic_path}" if pic_path else ''
 
+            name = data.get('Name', '') or ''
+            subtitle = data.get('Nick', '') or ''
+
             return PChomeProduct(
                 product_id=product_id,
-                name=data.get('Name', ''),
+                name=name,
                 price=price,
                 original_price=original_price,
                 discount=discount,
@@ -347,7 +363,9 @@ class PChomeCrawler:
                 rating=data.get('RatingValue'),
                 review_count=data.get('ReviewCount', 0),
                 is_on_sale=data.get('isOnSale', False),
-                crawled_at=crawled_at
+                crawled_at=crawled_at,
+                subtitle=subtitle,
+                match_name=_build_match_name(name, subtitle),
             )
 
         except Exception as e:

services/telegram_templates.py

@@ -776,7 +776,9 @@ def _action_label(action_code: str) -> str:
     labels = {
         "price_follow_review": "確認是否跟價或改用促銷防守",
         "review_accept_identity": "人工確認同款後採納 identity",
+        "review_catalog_comparable": "依型錄證據覆核可比性",
         "unit_price_required": "改用單位價覆核，不寫總價型價差",
+        "identity_or_price_review": "先確認身份，再判斷價格處置",
         "verify_or_reject_identity": "確認候選是否同款；非同款即駁回",
         "compare_existing_identity": "比較既有正式 identity 與新候選",
         "refresh_or_compare_identity": "刷新過期 identity 後再覆核",
@@ -786,6 +788,55 @@ def _action_label(action_code: str) -> str:
     return labels.get(action_code or "", action_code or "人工覆核")
 
 
+_PRICE_MATCH_TYPE_LABELS = {
+    "exact": "高信心同款",
+    "same_product_different_pack": "同商品不同包裝",
+    "same_line_variant": "同系列不同款",
+    "comparable": "可比但需覆核",
+    "no_match": "非同款",
+}
+_PRICE_BASIS_LABELS = {
+    "total_price": "總價可比",
+    "unit_price": "單位價可比",
+    "manual_review": "人工覆核後可比",
+    "none": "不可比",
+}
+_PRICE_ALERT_TIER_LABELS = {
+    "price_alert_exact": "可直接價格告警",
+    "unit_price_review": "單位價覆核",
+    "identity_review": "身份覆核",
+    "suppress": "壓制告警",
+}
+
+
+def _price_match_path(envelope: Dict[str, Any]) -> tuple[str, str, str]:
+    guardrails = envelope.get("guardrails") if isinstance(envelope.get("guardrails"), dict) else {}
+    match_type = str(guardrails.get("match_type") or "")
+    price_basis = str(guardrails.get("price_basis") or "")
+    alert_tier = str(guardrails.get("alert_tier") or "")
+    if match_type and price_basis and alert_tier:
+        return match_type, price_basis, alert_tier
+
+    match_evidence = _find_evidence(envelope, "match_score") or {}
+    basis = str(match_evidence.get("basis") or "")
+    parts = [part.strip() for part in basis.split("/") if part.strip()]
+    if len(parts) >= 3:
+        return match_type or parts[0], price_basis or parts[1], alert_tier or parts[2]
+    return match_type, price_basis, alert_tier
+
+
+def _price_notification_guidance(match_type: str, price_basis: str, alert_tier: str) -> tuple[str, str]:
+    if match_type == "exact" and price_basis == "total_price" and alert_tier == "price_alert_exact":
+        return "直接價格威脅", "可用總價比較；先確認庫存與促銷期，再人工決定跟價或促銷防守。"
+    if price_basis == "unit_price" or alert_tier == "unit_price_review":
+        return "單位價覆核", "先換算單位價與入數，禁止用總價直接判定價格威脅。"
+    if alert_tier == "identity_review" or price_basis == "manual_review":
+        return "身份覆核", "先確認同款、規格、組合與前台狀態，人工採納後才可寫入正式價差。"
+    if alert_tier == "suppress" or match_type == "no_match" or price_basis == "none":
+        return "壓制告警", "目前不可作為價格威脅；保留診斷紀錄，避免誤報。"
+    return "可比性待判讀", "依比對證據人工覆核，未確認前不自動調價、不覆蓋正式 identity。"
+
+
 def _format_price_decision_envelope(envelope: Dict[str, Any]) -> List[str]:
     """將價格/競品決策信封排成可讀的專業 brief。"""
     severity = escape(str(envelope.get("severity") or "info"))
@@ -809,6 +860,22 @@ def _format_price_decision_envelope(envelope: Dict[str, Any]) -> List[str]:
     if blocked_reason:
         lines.append(f"• 邊界：{blocked_reason}")
 
+    match_type, price_basis, alert_tier = _price_match_path(envelope)
+    if match_type or price_basis or alert_tier:
+        guidance_title, guidance_text = _price_notification_guidance(match_type, price_basis, alert_tier)
+        path_labels = [
+            _PRICE_MATCH_TYPE_LABELS.get(match_type, match_type) if match_type else "",
+            _PRICE_BASIS_LABELS.get(price_basis, price_basis) if price_basis else "",
+            _PRICE_ALERT_TIER_LABELS.get(alert_tier, alert_tier) if alert_tier else "",
+        ]
+        lines += [
+            "",
+            "🚦 <b>通知分級</b>",
+            f"• 判讀：<b>{escape(guidance_title)}</b>",
+            f"• 路徑：{escape(' / '.join(part for part in path_labels if part))}",
+            f"• 邊界：{escape(guidance_text)}",
+        ]
+
     sku = escape(str(subject.get("sku") or ""))
     name = escape(_short_text(subject.get("name") or "", 96))
     competitor_id = escape(str(subject.get("competitor_product_id") or subject.get("pchome_id") or ""))
@@ -891,6 +958,24 @@ def _format_price_decision_envelope(envelope: Dict[str, Any]) -> List[str]:
     if evidence_lines:
         lines += ["", "🧩 <b>比對證據</b>", *evidence_lines]
 
+    difference_highlights = envelope.get("difference_highlights")
+    if isinstance(difference_highlights, list) and difference_highlights:
+        diff_lines = []
+        for row in difference_highlights[:3]:
+            if not isinstance(row, dict):
+                continue
+            dimension = escape(str(row.get("dimension") or row.get("label") or "差異"))
+            left = escape(_short_text(row.get("left") or row.get("momo") or "", 42))
+            right = escape(_short_text(row.get("right") or row.get("pchome") or "", 42))
+            if left or right:
+                diff_lines.append(f"• {dimension}：MOMO {left or '—'} / PChome {right or '—'}")
+            else:
+                note = escape(_short_text(row.get("note") or row.get("summary") or "", 84))
+                if note:
+                    diff_lines.append(f"• {dimension}：{note}")
+        if diff_lines:
+            lines += ["", "⚖️ <b>差異提醒</b>", *diff_lines]
+
     action_code = str(recommended_action.get("action") or "human_review")
     owner = escape(str(recommended_action.get("owner") or "未指定"))
     requires_hitl = bool(recommended_action.get("requires_hitl", True))