強化 PChome 覆核與候選決策閉環

services/competitor_match_review_service.py

@@ -278,6 +278,27 @@ def _promote_manual_match(conn, attempt: dict[str, Any], source: str) -> None:
     })
 
 
+def _expire_current_manual_candidate(conn, attempt: dict[str, Any], source: str) -> None:
+    """Expire a current official match when the operator rejects its candidate."""
+    candidate_id = str(attempt.get("best_competitor_product_id") or "").strip()
+    sku = str(attempt.get("sku") or "").strip()
+    if not sku or not candidate_id:
+        return
+
+    conn.execute(text("""
+        UPDATE competitor_prices
+        SET expires_at = CURRENT_TIMESTAMP,
+            crawled_at = CURRENT_TIMESTAMP
+        WHERE sku = :sku
+          AND source = :source
+          AND competitor_product_id = :candidate_id
+    """), {
+        "sku": sku,
+        "source": source,
+        "candidate_id": candidate_id,
+    })
+
+
 def record_competitor_match_review(
     engine,
     sku: str,
@@ -307,6 +328,8 @@ def record_competitor_match_review(
 
         if review_action == "accept_identity":
             _promote_manual_match(conn, attempt, source)
+        else:
+            _expire_current_manual_candidate(conn, attempt, source)
 
         _insert_manual_attempt(conn, attempt, action_meta, source)
         conn.execute(text("""

services/competitor_price_feeder.py

@@ -37,6 +37,7 @@ logger = logging.getLogger(__name__)
 # ── 比對參數 ─────────────────────────────────────────
 MIN_MATCH_SCORE  = 0.76  # 低於此分數不寫入；核心比價寧可待審也不能錯配
 REPLACE_DIFFERENT_PRODUCT_SCORE = 0.84  # 已有不同 PChome 商品時，需超高信心才覆蓋
+EARLY_STOP_MATCH_SCORE = 0.90  # 搜尋候選池只有強同款才提前停止，避免次佳候選卡住後續精準搜尋詞
 SEARCH_LIMIT     = 12    # 每個搜尋詞取 PChome 前 N 筆
 MAX_SEARCH_TERMS  = 3     # 每個 MOMO 商品最多嘗試幾組搜尋詞
 BATCH_SIZE       = 30    # 每批 DB 寫入筆數
@@ -178,9 +179,19 @@ def _find_best_match_detail(
     Returns:
         (PChomeProduct, score, diagnostics) or None
     """
+    ranked = _rank_match_details(momo_name, pchome_products, momo_price=momo_price)
+    return ranked[0] if ranked else None
+
+
+def _rank_match_details(
+    momo_name: str,
+    pchome_products: list,
+    momo_price: float = None,
+) -> list[tuple]:
+    """Score all PChome candidates and return them from strongest to weakest."""
     from services.marketplace_product_matcher import score_marketplace_match
 
-    best, best_score, best_diagnostics = None, 0.0, None
+    ranked = []
     for p in pchome_products:
         diagnostics = score_marketplace_match(
             momo_name,
@@ -188,11 +199,8 @@ def _find_best_match_detail(
             momo_price=momo_price,
             competitor_price=getattr(p, "price", None),
         )
-        score = diagnostics.score
-        if score > best_score:
-            best, best_score, best_diagnostics = p, score, diagnostics
-
-    return (best, best_score, best_diagnostics) if best else None
+        ranked.append((p, diagnostics.score, diagnostics))
+    return sorted(ranked, key=lambda item: item[1], reverse=True)
 
 
 def _find_best_match(momo_name: str, pchome_products: list) -> Optional[tuple]:
@@ -205,7 +213,7 @@ def _find_best_match(momo_name: str, pchome_products: list) -> Optional[tuple]:
 
 
 def _search_pchome_candidates(crawler, momo_name: str, keywords: list = None, momo_price: float = None) -> list:
-    """以多組搜尋詞擴大 PChome 候選池，找到可信候選後提早停止。"""
+    """以多組搜尋詞擴大 PChome 候選池，只在強同款時提前停止。"""
     candidates = []
     seen_ids = set()
     for keyword in keywords or _build_search_keywords(momo_name):
@@ -218,7 +226,7 @@ def _search_pchome_candidates(crawler, momo_name: str, keywords: list = None, mo
             seen_ids.add(product.product_id)
             candidates.append(product)
         best = _find_best_match_detail(momo_name, candidates, momo_price=momo_price)
-        if best and best[1] >= 0.76:
+        if best and best[1] >= EARLY_STOP_MATCH_SCORE:
             break
     return candidates
 
@@ -791,8 +799,8 @@ class CompetitorPriceFeeder:
                     skipped_no += 1
                     continue
 
-                result = _find_best_match_detail(momo_name, products, momo_price=momo_price)
-                if not result:
+                ranked_matches = _rank_match_details(momo_name, products, momo_price=momo_price)
+                if not ranked_matches:
                     self._record_match_attempt(
                         sku,
                         momo_name,
@@ -807,17 +815,31 @@ class CompetitorPriceFeeder:
                     skipped_no += 1
                     continue
 
-                best_product, score, diagnostics = result
-                manual_review = self._fetch_latest_manual_review_for_candidate(
-                    sku,
-                    getattr(best_product, "product_id", None),
-                    source=source,
-                )
-                manual_action = (manual_review or {}).get("review_action")
-                if manual_action == "reject_identity":
+                selected_match = None
+                manually_rejected_ids: list[str] = []
+                for candidate_product, candidate_score, candidate_diagnostics in ranked_matches:
+                    candidate_review = self._fetch_latest_manual_review_for_candidate(
+                        sku,
+                        getattr(candidate_product, "product_id", None),
+                        source=source,
+                    )
+                    if (candidate_review or {}).get("review_action") == "reject_identity":
+                        manually_rejected_ids.append(str(getattr(candidate_product, "product_id", "") or ""))
+                        continue
+                    selected_match = (
+                        candidate_product,
+                        candidate_score,
+                        candidate_diagnostics,
+                        candidate_review,
+                    )
+                    break
+
+                if not selected_match:
+                    best_product, score, diagnostics = ranked_matches[0]
+                    rejected_note = ",".join(product_id for product_id in manually_rejected_ids if product_id)
                     logger.info(
-                        f"[Feeder] {sku} 候選已被人工否決，跳過正式寫入 | "
-                        f"candidate={getattr(best_product, 'product_id', None)}"
+                        f"[Feeder] {sku} 所有可信候選都已被人工否決，跳過正式寫入 | "
+                        f"rejected_candidates={rejected_note}"
                     )
                     self._record_match_attempt(
                         sku,
@@ -829,12 +851,18 @@ class CompetitorPriceFeeder:
                         attempt_status="manual_rejected",
                         best_product=best_product,
                         best_score=score,
-                        error_message=f"manual_review_rejected; {_format_match_diagnostics(diagnostics)}",
+                        error_message=(
+                            f"manual_review_rejected; rejected_candidates={rejected_note}; "
+                            f"{_format_match_diagnostics(diagnostics)}"
+                        ),
                         source=source,
                     )
                     attempts_written += 1
                     skipped_low += 1
                     continue
+
+                best_product, score, diagnostics, manual_review = selected_match
+                manual_action = (manual_review or {}).get("review_action")
                 if manual_action == "unit_price_required":
                     logger.info(
                         f"[Feeder] {sku} 候選已被人工標記為單位價比較，不寫正式總價差 | "