feat: add momo review candidate queue

services/external_market_offer_service.py

@@ -1435,3 +1435,185 @@ def build_external_source_readiness(engine=None) -> dict[str, Any]:
         "connector_contract": build_connector_contracts(),
         "plain_summary": "MOMO 先用；蝦皮與酷澎先保留接口，暫不進告警。",
     }
+
+
+def list_momo_review_candidates(engine, *, limit: int = 20) -> dict[str, Any]:
+    """列出待人工確認的 MOMO 候選，供前台直接處理。"""
+    limit = max(1, min(int(limit or 20), 50))
+    generated_at = datetime.now().isoformat(timespec="seconds")
+    required_tables = {"external_offers"}
+
+    with engine.connect() as conn:
+        missing_tables = sorted(table for table in required_tables if not _has_table(conn, table))
+        if missing_tables:
+            return {
+                "success": False,
+                "generated_at": generated_at,
+                "rows": [],
+                "count": 0,
+                "missing_tables": missing_tables,
+                "message": "待確認候選暫時無法讀取，缺少必要資料表。",
+            }
+
+        rows = conn.execute(text("""
+            SELECT
+                id,
+                source_product_id,
+                title,
+                product_url,
+                image_url,
+                price,
+                pchome_product_id,
+                momo_sku,
+                match_status,
+                quality_score,
+                data_quality_status,
+                quality_notes_json,
+                raw_payload_json,
+                observed_at,
+                updated_at
+            FROM external_offers
+            WHERE source_code = 'momo_reference'
+              AND ingestion_method = 'targeted_momo_review'
+              AND (
+                  match_status = 'needs_review'
+                  OR data_quality_status = 'needs_review'
+              )
+            ORDER BY observed_at DESC, id DESC
+            LIMIT :limit
+        """), {"limit": limit * 4}).mappings().all()
+
+    seen: set[tuple[str, str]] = set()
+    items: list[dict[str, Any]] = []
+    for row in rows:
+        raw_payload = _load_json_dict(row.get("raw_payload_json"))
+        quality_notes = _load_json_list(row.get("quality_notes_json"))
+        key = (
+            str(row.get("pchome_product_id") or "").strip(),
+            str(row.get("source_product_id") or "").strip(),
+        )
+        if key in seen:
+            continue
+        seen.add(key)
+
+        reasons = [
+            str(reason)
+            for reason in (raw_payload.get("match_reasons") or [])
+            if str(reason or "").strip()
+        ]
+        if not reasons:
+            reasons = [str(note) for note in quality_notes if str(note or "").strip()]
+        pchome_price = _to_float(raw_payload.get("pchome_public_price"))
+        momo_price = _to_float(row.get("price"))
+        gap_pct = _to_float(raw_payload.get("target_gap_pct"))
+
+        items.append({
+            "id": int(row.get("id")),
+            "pchome_product_id": row.get("pchome_product_id"),
+            "pchome_product_name": raw_payload.get("pchome_public_name") or "",
+            "pchome_price": pchome_price,
+            "momo_sku": row.get("momo_sku") or row.get("source_product_id"),
+            "momo_title": row.get("title"),
+            "momo_price": momo_price,
+            "product_url": row.get("product_url"),
+            "image_url": row.get("image_url"),
+            "quality_score": round(_to_float(row.get("quality_score")) or 0.0, 2),
+            "alert_tier": raw_payload.get("alert_tier") or "identity_review",
+            "price_basis": raw_payload.get("price_basis") or "manual_review",
+            "gap_pct": gap_pct,
+            "match_reasons": reasons[:5],
+            "observed_at": str(row.get("observed_at") or ""),
+            "updated_at": str(row.get("updated_at") or ""),
+            "plain_status": "待確認同款或色號",
+            "suggested_next_action": "確認同款後才進入價格判斷；不是同款就排除。",
+        })
+        if len(items) >= limit:
+            break
+
+    return {
+        "success": True,
+        "generated_at": generated_at,
+        "rows": items,
+        "count": len(items),
+        "message": "已整理 MOMO 待確認候選。",
+    }
+
+
+def update_momo_review_candidate(engine, offer_id: int, action: str, *, note: str = "") -> dict[str, Any]:
+    """確認或排除一筆 MOMO 待確認候選。"""
+    try:
+        offer_id = int(offer_id)
+    except (TypeError, ValueError):
+        return {"success": False, "message": "缺少有效的候選編號。"}
+    action = str(action or "").strip().lower()
+    if action not in {"confirm", "reject"}:
+        return {"success": False, "message": "請選擇確認同款或排除候選。"}
+
+    generated_at = datetime.now().isoformat(timespec="seconds")
+    new_match_status = "verified" if action == "confirm" else "rejected"
+    new_quality_status = "verified" if action == "confirm" else "rejected"
+    label = "人工確認同款" if action == "confirm" else "人工排除候選"
+    review_note = str(note or "").strip()[:240]
+
+    with engine.begin() as conn:
+        if not _has_table(conn, "external_offers"):
+            return {
+                "success": False,
+                "generated_at": generated_at,
+                "message": "待確認候選暫時無法更新，缺少必要資料表。",
+            }
+
+        row = conn.execute(text("""
+            SELECT id, match_status, data_quality_status, quality_notes_json, raw_payload_json
+            FROM external_offers
+            WHERE id = :offer_id
+              AND source_code = 'momo_reference'
+              AND ingestion_method = 'targeted_momo_review'
+            LIMIT 1
+        """), {"offer_id": offer_id}).mappings().first()
+        if not row:
+            return {
+                "success": False,
+                "generated_at": generated_at,
+                "message": "找不到這筆待確認候選。",
+            }
+
+        raw_payload = _load_json_dict(row.get("raw_payload_json"))
+        raw_payload["review_state"] = new_match_status
+        raw_payload["reviewed_at"] = generated_at
+        raw_payload["review_action"] = action
+        if review_note:
+            raw_payload["review_note"] = review_note
+        tags = raw_payload.get("tags") if isinstance(raw_payload.get("tags"), list) else []
+        tag_to_add = "manual_verified" if action == "confirm" else "manual_rejected"
+        raw_payload["tags"] = [*tags, tag_to_add] if tag_to_add not in tags else tags
+
+        notes = [str(item) for item in _load_json_list(row.get("quality_notes_json")) if str(item or "").strip()]
+        notes.append(label if not review_note else f"{label}：{review_note}")
+
+        conn.execute(text("""
+            UPDATE external_offers
+            SET match_status = :match_status,
+                data_quality_status = :data_quality_status,
+                quality_notes_json = :quality_notes_json,
+                raw_payload_json = :raw_payload_json,
+                updated_at = CURRENT_TIMESTAMP
+            WHERE id = :offer_id
+        """), {
+            "offer_id": offer_id,
+            "match_status": new_match_status,
+            "data_quality_status": new_quality_status,
+            "quality_notes_json": json.dumps(notes[-6:], ensure_ascii=False),
+            "raw_payload_json": json.dumps(raw_payload, ensure_ascii=False),
+        })
+        mark_pchome_growth_cache_stale()
+
+    return {
+        "success": True,
+        "generated_at": generated_at,
+        "id": offer_id,
+        "action": action,
+        "match_status": new_match_status,
+        "data_quality_status": new_quality_status,
+        "message": "已確認同款，會進入作戰清單。" if action == "confirm" else "已排除候選，不會再進入待確認。",
+    }