補齊比價覆核隊列分頁

services/competitor_intel_repository.py

@@ -102,6 +102,28 @@ def _build_unit_comparison_for_attempt(row: dict[str, Any]) -> Optional[dict[str
         return {"comparable": False, "reason": "build_error"}
 
 
+def _format_competitor_review_item(row: dict[str, Any]) -> dict[str, Any]:
+    item = dict(row)
+    unit_comparison = _build_unit_comparison_for_attempt(item)
+    return {
+        "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,
+    }
+
+
 def clear_competitor_intel_cache() -> None:
     """Clear cached PChome/MOMO intelligence after crawler/import updates."""
     with _CACHE_LOCK:
@@ -478,6 +500,177 @@ def fetch_competitor_review_queue(engine, limit: int = 12) -> list[dict]:
     )
 
 
+def fetch_competitor_review_queue_page(
+    engine,
+    page: int = 1,
+    per_page: int = 50,
+    search_query: str = "",
+    category: str = "",
+) -> dict:
+    """Paginated PChome review queue for operator-facing Dashboard pages."""
+    page = max(1, int(page or 1))
+    per_page = max(1, min(int(per_page or 50), 100))
+    search_query = (search_query or "").strip()
+    category = (category or "").strip()
+    cache_key = (
+        "review_queue_page:v1:"
+        f"page={page}:per={per_page}:q={search_query.lower()}:cat={category}:"
+        f"floor={PCHOME_MATCH_SCORE_FLOOR}"
+    )
+    return _cached_payload(
+        cache_key,
+        lambda: _fetch_competitor_review_queue_page_uncached(
+            engine,
+            page=page,
+            per_page=per_page,
+            search_query=search_query,
+            category=category,
+        ),
+        ttl_seconds=min(COMPETITOR_INTEL_CACHE_TTL_SECONDS, 300),
+    )
+
+
+def _review_queue_cte_and_filter(search_query: str = "", category: str = "") -> tuple[str, dict[str, Any]]:
+    params: dict[str, Any] = {}
+    filters = [
+        "lm.rn = 1",
+        "vc.sku IS NULL",
+        """la.attempt_status IN (
+              'unit_comparable',
+              'refresh_unit_comparable',
+              'identity_veto',
+              'low_score',
+              'expired_match',
+              'refresh_no_result',
+              'no_result'
+          )""",
+    ]
+    if search_query:
+        params["search_like"] = f"%{search_query.lower()}%"
+        filters.append("(LOWER(lm.name) LIKE :search_like OR LOWER(lm.sku) LIKE :search_like)")
+    if category:
+        params["category"] = category
+        filters.append("lm.category = :category")
+
+    where_sql = "\n          AND ".join(filters)
+    cte = 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
+        ),
+        review_rows AS (
+            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,
+                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 AS priority_rank
+            FROM latest_momo lm
+            JOIN latest_attempt la ON la.sku = lm.sku
+            LEFT JOIN valid_competitor vc ON vc.sku = lm.sku
+            WHERE {where_sql}
+        )
+    """
+    return cte, params
+
+
+def _fetch_competitor_review_queue_page_uncached(
+    engine,
+    page: int = 1,
+    per_page: int = 50,
+    search_query: str = "",
+    category: str = "",
+) -> 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 {"items": [], "total": 0, "page": max(1, int(page or 1)), "per_page": per_page}
+
+    page = max(1, int(page or 1))
+    per_page = max(1, min(int(per_page or 50), 100))
+    cte, params = _review_queue_cte_and_filter(search_query=search_query, category=category)
+    page_params = {
+        **params,
+        "limit": per_page,
+        "offset": (page - 1) * per_page,
+    }
+    count_sql = text(cte + " SELECT COUNT(*) AS total FROM review_rows")
+    page_sql = text(cte + """
+        SELECT *
+        FROM review_rows
+        ORDER BY
+            priority_rank ASC,
+            momo_price DESC NULLS LAST,
+            best_match_score DESC NULLS LAST,
+            attempted_at DESC NULLS LAST
+        LIMIT :limit OFFSET :offset
+    """)
+
+    with engine.connect() as conn:
+        total = int(conn.execute(count_sql, params).scalar() or 0)
+        rows = conn.execute(page_sql, page_params).mappings().all()
+
+    return {
+        "items": [_format_competitor_review_item(dict(row)) for row in rows],
+        "total": total,
+        "page": page,
+        "per_page": per_page,
+    }
+
+
 def _fetch_competitor_review_queue_uncached(engine, limit: int = 12) -> list[dict]:
     inspector = inspect(engine)
     if not (
@@ -572,28 +765,7 @@ def _fetch_competitor_review_queue_uncached(engine, limit: int = 12) -> list[dic
     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
+    return [_format_competitor_review_item(dict(row)) for row in rows]
 
 
 def fetch_competitor_comparison_results(