串接 PChome 單位價覆核隊列

services/competitor_intel_repository.py

@@ -22,6 +22,36 @@ from sqlalchemy import inspect, text
 
 
 PCHOME_MATCH_SCORE_FLOOR = 0.76
+UNIT_COMPARABLE_STATUSES = {"unit_comparable", "refresh_unit_comparable"}
+ACTIONABLE_ATTEMPT_STATUSES = {
+    "unit_comparable",
+    "refresh_unit_comparable",
+    "identity_veto",
+    "low_score",
+    "expired_match",
+    "refresh_no_result",
+    "no_result",
+}
+ATTEMPT_STATUS_LABELS = {
+    "unit_comparable": "需單位價比較",
+    "refresh_unit_comparable": "需單位價比較",
+    "identity_veto": "身份否決",
+    "low_score": "低信心待審",
+    "expired_match": "價格過期待刷新",
+    "refresh_no_result": "刷新找不到商品",
+    "no_result": "找不到同款",
+    "never_attempted": "尚未搜尋",
+}
+ATTEMPT_ACTION_LABELS = {
+    "unit_comparable": "人工確認檔期、贈品與單位價",
+    "refresh_unit_comparable": "人工確認檔期、贈品與單位價",
+    "identity_veto": "確認是否為不同商品線或規格",
+    "low_score": "人工審核候選商品身份",
+    "expired_match": "重新刷新 PChome 價格",
+    "refresh_no_result": "調整搜尋詞後重抓",
+    "no_result": "補充搜尋詞或品牌關鍵字",
+    "never_attempted": "排入 PChome 補抓",
+}
 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"
@@ -48,6 +78,30 @@ def _month_label(value: Any) -> str:
     return str(value or "")[:7]
 
 
+def _attempt_status_label(status: Any) -> str:
+    return ATTEMPT_STATUS_LABELS.get(str(status or ""), str(status or "待比對"))
+
+
+def _attempt_action_label(status: Any) -> str:
+    return ATTEMPT_ACTION_LABELS.get(str(status or ""), "人工確認比對證據")
+
+
+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:
+        return None
+    try:
+        from services.marketplace_product_matcher import build_unit_price_comparison
+        return build_unit_price_comparison(
+            row.get("name") or row.get("momo_product_name") or "",
+            row.get("best_competitor_product_name") or "",
+            row.get("momo_price"),
+            row.get("best_competitor_price"),
+        )
+    except Exception:
+        return {"comparable": False, "reason": "build_error"}
+
+
 def clear_competitor_intel_cache() -> None:
     """Clear cached PChome/MOMO intelligence after crawler/import updates."""
     with _CACHE_LOCK:
@@ -124,15 +178,39 @@ def fetch_competitor_coverage(engine) -> dict:
 
 def _fetch_competitor_coverage_uncached(engine) -> dict:
     """讀取目前 PChome 比價覆蓋率與待審分類。"""
-    if not inspect(engine).has_table("competitor_prices"):
+    inspector = inspect(engine)
+    if not inspector.has_table("competitor_prices"):
         return {
             "active_with_price": 0,
             "valid_matches": 0,
             "pending": 0,
             "match_rate": 0,
             "attempt_status": {},
+            "unit_comparable_count": 0,
+            "actionable_review_count": 0,
         }
 
+    has_match_attempts = inspector.has_table("competitor_match_attempts")
+    attempt_cte = """
+        latest_attempt AS (
+            SELECT
+                NULL AS sku,
+                NULL AS attempt_status
+            WHERE FALSE
+        )
+    """
+    if has_match_attempts:
+        attempt_cte = """
+        latest_attempt AS (
+            SELECT DISTINCT ON (sku)
+                sku,
+                attempt_status
+            FROM competitor_match_attempts
+            WHERE source = 'pchome'
+            ORDER BY sku, attempted_at DESC NULLS LAST
+        )
+        """
+
     sql = text(f"""
         WITH latest_momo AS (
             SELECT
@@ -156,14 +234,7 @@ def _fetch_competitor_coverage_uncached(engine) -> dict:
               AND COALESCE(cp.tags, '[]'::jsonb) ? 'identity_v2'
             ORDER BY cp.sku, cp.crawled_at DESC NULLS LAST
         ),
-        latest_attempt AS (
-            SELECT DISTINCT ON (sku)
-                sku,
-                attempt_status
-            FROM competitor_match_attempts
-            WHERE source = 'pchome'
-            ORDER BY sku, attempted_at DESC NULLS LAST
-        )
+        {attempt_cte}
         SELECT
             (SELECT COUNT(*) FROM latest_momo WHERE rn = 1) AS active_with_price,
             (SELECT COUNT(*) FROM valid_competitor) AS valid_matches,
@@ -190,12 +261,16 @@ def _fetch_competitor_coverage_uncached(engine) -> dict:
         str(row.get("attempt_status")): int(row.get("status_count") or 0)
         for row in rows
     }
+    unit_count = sum(statuses.get(status, 0) for status in UNIT_COMPARABLE_STATUSES)
+    actionable_count = sum(statuses.get(status, 0) for status in ACTIONABLE_ATTEMPT_STATUSES)
     return {
         "active_with_price": active,
         "valid_matches": valid,
         "pending": pending,
         "match_rate": round(valid / max(active, 1) * 100, 1),
         "attempt_status": statuses,
+        "unit_comparable_count": unit_count,
+        "actionable_review_count": actionable_count,
         "match_score_floor": PCHOME_MATCH_SCORE_FLOOR,
     }
 
@@ -394,6 +469,133 @@ def _fetch_top_competitor_risks_uncached(engine, limit: int = 10) -> list[dict]:
     return result
 
 
+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:v1:limit={limit}:floor={PCHOME_MATCH_SCORE_FLOOR}",
+        lambda: _fetch_competitor_review_queue_uncached(engine, limit=limit),
+    )
+
+
+def _fetch_competitor_review_queue_uncached(engine, limit: int = 12) -> list[dict]:
+    inspector = inspect(engine)
+    if not (
+        inspector.has_table("products")
+        and inspector.has_table("price_records")
+        and inspector.has_table("competitor_prices")
+        and inspector.has_table("competitor_match_attempts")
+    ):
+        return []
+
+    limit = max(1, min(int(limit or 12), 50))
+    sql = text(f"""
+        WITH latest_momo AS (
+            SELECT
+                p.id AS product_id,
+                p.i_code AS sku,
+                p.name,
+                p.category,
+                pr.price AS momo_price,
+                ROW_NUMBER() OVER (PARTITION BY p.id ORDER BY pr.timestamp DESC, pr.id DESC) AS rn
+            FROM products p
+            JOIN price_records pr ON pr.product_id = p.id
+            WHERE p.status = 'ACTIVE'
+        ),
+        valid_competitor AS (
+            SELECT DISTINCT ON (cp.sku)
+                cp.sku
+            FROM competitor_prices cp
+            WHERE cp.source = 'pchome'
+              AND (cp.expires_at IS NULL OR cp.expires_at > CURRENT_TIMESTAMP)
+              AND cp.price IS NOT NULL
+              AND cp.price > 0
+              AND COALESCE(cp.match_score, 0) >= {PCHOME_MATCH_SCORE_FLOOR}
+              AND COALESCE(cp.tags, '[]'::jsonb) ? 'identity_v2'
+            ORDER BY cp.sku, cp.crawled_at DESC NULLS LAST
+        ),
+        latest_attempt AS (
+            SELECT DISTINCT ON (cma.sku)
+                cma.sku,
+                cma.attempt_status,
+                cma.candidate_count,
+                cma.best_competitor_product_id,
+                cma.best_competitor_product_name,
+                cma.best_competitor_price,
+                cma.best_match_score,
+                cma.error_message,
+                cma.attempted_at
+            FROM competitor_match_attempts cma
+            WHERE cma.source = 'pchome'
+            ORDER BY cma.sku, cma.attempted_at DESC NULLS LAST
+        )
+        SELECT
+            lm.sku,
+            lm.name,
+            lm.category,
+            lm.momo_price,
+            la.attempt_status,
+            la.candidate_count,
+            la.best_competitor_product_id,
+            la.best_competitor_product_name,
+            la.best_competitor_price,
+            la.best_match_score,
+            la.error_message,
+            la.attempted_at
+        FROM latest_momo lm
+        JOIN latest_attempt la ON la.sku = lm.sku
+        LEFT JOIN valid_competitor vc ON vc.sku = lm.sku
+        WHERE lm.rn = 1
+          AND vc.sku IS NULL
+          AND la.attempt_status IN (
+              'unit_comparable',
+              'refresh_unit_comparable',
+              'identity_veto',
+              'low_score',
+              'expired_match',
+              'refresh_no_result',
+              'no_result'
+          )
+        ORDER BY
+            CASE
+                WHEN la.attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 0
+                WHEN la.attempt_status = 'identity_veto' THEN 1
+                WHEN la.attempt_status = 'low_score' THEN 2
+                WHEN la.attempt_status = 'expired_match' THEN 3
+                ELSE 4
+            END,
+            lm.momo_price DESC NULLS LAST,
+            la.best_match_score DESC NULLS LAST,
+            la.attempted_at DESC NULLS LAST
+        LIMIT :limit
+    """)
+    with engine.connect() as conn:
+        rows = conn.execute(sql, {"limit": limit}).mappings().all()
+
+    queue = []
+    for row in rows:
+        item = dict(row)
+        unit_comparison = _build_unit_comparison_for_attempt(item)
+        queue.append({
+            "sku": str(item.get("sku") or ""),
+            "name": item.get("name") or "",
+            "category": item.get("category") or "",
+            "momo_price": _num(item.get("momo_price")),
+            "attempt_status": item.get("attempt_status") or "",
+            "status_label": _attempt_status_label(item.get("attempt_status")),
+            "action_label": _attempt_action_label(item.get("attempt_status")),
+            "candidate_count": int(item.get("candidate_count") or 0),
+            "candidate_pc_id": item.get("best_competitor_product_id"),
+            "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 "",
+            "attempted_at": _date_label(item.get("attempted_at")),
+            "unit_comparison": unit_comparison,
+        })
+    return queue
+
+
 def fetch_competitor_comparison_results(
     engine,
     start_date: Optional[Union[date, datetime, str]] = None,
@@ -549,18 +751,13 @@ def fetch_competitor_comparison_results(
         pchome_id = row.get("competitor_product_id")
         found = bool(row.get("pchome_price"))
         match_status = "matched" if found else (row.get("attempt_status") or "no_valid_match")
-        unit_comparison = None
-        if match_status in {"unit_comparable", "refresh_unit_comparable"}:
-            try:
-                from services.marketplace_product_matcher import build_unit_price_comparison
-                unit_comparison = build_unit_price_comparison(
-                    row.get("name") or "",
-                    row.get("best_competitor_product_name") or "",
-                    row.get("momo_price"),
-                    row.get("best_competitor_price"),
-                )
-            except Exception:
-                unit_comparison = {"comparable": False, "reason": "build_error"}
+        unit_comparison = _build_unit_comparison_for_attempt({
+            "attempt_status": match_status,
+            "name": row.get("name") or "",
+            "best_competitor_product_name": row.get("best_competitor_product_name") or "",
+            "momo_price": row.get("momo_price"),
+            "best_competitor_price": row.get("best_competitor_price"),
+        })
         results.append({
             "found": found,
             "momo_icode": str(row.get("sku") or ""),
@@ -577,6 +774,8 @@ def fetch_competitor_comparison_results(
             "match_score": _num(row.get("match_score")),
             "momo_revenue": _num(row.get("momo_revenue")),
             "match_status": match_status,
+            "match_status_label": _attempt_status_label(match_status),
+            "action_label": _attempt_action_label(match_status),
             "candidate_count": int(row.get("candidate_count") or 0),
             "best_match_score": _num(row.get("best_match_score")),
             "match_diagnostic": row.get("error_message") or "",
@@ -591,5 +790,6 @@ def build_competitor_intel_payload(engine, days: int = 30) -> dict:
         "coverage": fetch_competitor_coverage(engine),
         "trend": fetch_competitor_gap_trend(engine, days=days),
         "top_risks": fetch_top_competitor_risks(engine, limit=10),
+        "review_queue": fetch_competitor_review_queue(engine, limit=12),
         "match_score_floor": PCHOME_MATCH_SCORE_FLOOR,
     }

services/openclaw_strategist_service.py

@@ -30,7 +30,7 @@ from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 
 from database.manager import get_session
-from sqlalchemy import bindparam, text
+from sqlalchemy import bindparam, inspect, text
 
 from services.ai_call_logger import log_ai_call  # Operation Ollama-First v5.0 P1
 from services.action_plan_dedupe import (
@@ -563,11 +563,29 @@ def _fetch_competitor_summary() -> Dict[str, Any]:
               AND COALESCE(cp.tags, '[]'::jsonb) ? 'identity_v2'
         """)).fetchone()
         if row and row[0]:
+            attempt_row = None
+            if session.bind is not None and inspect(session.bind).has_table("competitor_match_attempts"):
+                attempt_row = session.execute(text("""
+                    WITH latest_attempt AS (
+                        SELECT DISTINCT ON (sku)
+                            sku,
+                            attempt_status
+                        FROM competitor_match_attempts
+                        WHERE source = 'pchome'
+                        ORDER BY sku, attempted_at DESC NULLS LAST
+                    )
+                    SELECT
+                        SUM(CASE WHEN attempt_status IN ('unit_comparable', 'refresh_unit_comparable') THEN 1 ELSE 0 END) AS unit_comparable_count,
+                        SUM(CASE WHEN attempt_status IN ('unit_comparable', 'refresh_unit_comparable', 'identity_veto', 'low_score', 'expired_match', 'no_result', 'refresh_no_result') THEN 1 ELSE 0 END) AS review_queue_count
+                    FROM latest_attempt
+                """)).fetchone()
             return {
                 "total_skus": int(row[0]),
                 "avg_gap_pct": round(float(row[1] or 0), 1),
                 "undercut_count": int(row[2] or 0),
                 "premium_count": int(row[3] or 0),
+                "unit_comparable_count": int((attempt_row[0] if attempt_row else 0) or 0),
+                "review_queue_count": int((attempt_row[1] if attempt_row else 0) or 0),
             }
         return {}
     except Exception as e:
@@ -1342,6 +1360,7 @@ def generate_weekly_strategy_report(
   平均價差：{competitor_summary.get('avg_gap_pct', 0):+.1f}%
   被競品削價數：{competitor_summary.get('undercut_count', 0)} 個
   我方具優勢數：{competitor_summary.get('premium_count', 0)} 個
+  需單位價覆核：{competitor_summary.get('unit_comparable_count', 0)} 個
 
 TOP 威脅品項（近48h Hermes 偵測）：
 {_format_threats(threats)}
@@ -1615,6 +1634,7 @@ def _legacy_full_gemini_daily_report() -> dict:
   監控SKU：{competitor_summary.get('total_skus', 0)} 個
   被削價風險：{competitor_summary.get('undercut_count', 0)} 個（價差超過10%）
   平均價差：{competitor_summary.get('avg_gap_pct', 0):+.1f}%
+  單位價/身份覆核隊列：{competitor_summary.get('review_queue_count', 0)} 個
 
 請按以下結構輸出（使用 HTML <b> 標題）：
 
@@ -1785,6 +1805,7 @@ def generate_monthly_report() -> dict:
   監控SKU：{competitor_summary.get('total_skus', 0)} 個
   月均價差：{competitor_summary.get('avg_gap_pct', 0):+.1f}%
   被削價風險SKU：{competitor_summary.get('undercut_count', 0)} 個
+  需單位價覆核SKU：{competitor_summary.get('unit_comparable_count', 0)} 個
 
 【價格變動概況】
   本月調價次數：{price_trend_data.get('price_changes', 0)} 次