V10.527 收斂過期 identity 救援隊列

services/competitor_price_feeder.py

@@ -96,6 +96,39 @@ REVALIDATABLE_REVIEW_SQL_REASON_LIST = ", ".join(
 REVALIDATABLE_REVIEW_BLOCK_SQL_REASON_LIST = ", ".join(
     f"'{reason}'" for reason in sorted(REVALIDATABLE_REVIEW_BLOCK_REASONS)
 )
+STALE_IDENTITY_RECOVERY_BLOCK_REASONS = {
+    "accessory_case_conflict",
+    "aroma_lamp_style_selection_gap",
+    "aroma_scent_variant_conflict",
+    "bundle_offer_conflict",
+    "candle_catalog_selection_gap",
+    "catalog_count_omission",
+    "commercial_condition_gap",
+    "count_conflict",
+    "makeup_catalog_selection_gap",
+    "makeup_finish_conflict",
+    "makeup_usage_conflict",
+    "multi_component_conflict",
+    "multi_component_count_conflict",
+    "named_component_quantity_conflict",
+    "nail_tool_function_conflict",
+    "price_ratio_extreme",
+    "price_ratio_wide",
+    "product_line_conflict",
+    "refill_pack_conflict",
+    "romand_lip_line_conflict",
+    "unit_comparable",
+    "variant_descriptor_conflict",
+    "variant_option_conflict",
+    "variant_selection_review",
+}
+STALE_IDENTITY_RECOVERY_BLOCK_SQL_REASON_LIST = ", ".join(
+    f"'{reason}'" for reason in sorted(STALE_IDENTITY_RECOVERY_BLOCK_REASONS)
+)
+STALE_IDENTITY_RECOVERY_BLOCK_NAME_PATTERN = (
+    r"(任選|多款|色號|顏色|款式|香味|香調|即期|短效|航空版|"
+    r"融燭燈|融蠟燈|香氛蠟燭|精油蠟燭|蠟燭|限定|組合任選)"
+)
 
 # ── Feeder 結果 ───────────────────────────────────────
 @dataclass
@@ -1259,7 +1292,7 @@ class CompetitorPriceFeeder:
         不查 PChome、不重新比對、不寫 attempts / prices。
         """
         preview_limit = max(1, min(int(limit), 120))
-        rows = self._fetch_expired_identity_skus(limit=preview_limit)
+        rows = self._fetch_expired_identity_recovery_skus(limit=preview_limit)
         examples: list[dict] = []
         for row in rows[:5]:
             examples.append({
@@ -1280,6 +1313,85 @@ class CompetitorPriceFeeder:
             "boundary": "read_only_no_crawl_no_llm_no_db_write",
         }
 
+    def _fetch_expired_identity_recovery_skus(self, limit: int = 40) -> list:
+        """
+        取得適合 fresh-search recovery 的過期 identity_v2 商品。
+
+        這比一般 expired refresh 更窄：只收過去已是 exact / total_price /
+        price_alert_exact 的正式配對，且排除款式、香味、型態、入數、商業狀態等
+        高風險診斷或名稱訊號。避免把本來應該人工覆核的 stale pair 送進慢速搜尋。
+        """
+        if self.engine is None:
+            raise RuntimeError("需要注入 SQLAlchemy engine")
+
+        from sqlalchemy import text
+        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) AS rn
+                FROM products p
+                JOIN price_records pr ON pr.product_id = p.id
+                WHERE p.status = 'ACTIVE'
+            )
+            SELECT
+                lm.product_id,
+                lm.sku,
+                lm.name,
+                lm.category,
+                lm.momo_price,
+                cp.competitor_product_id,
+                cp.competitor_product_name,
+                cp.match_score,
+                cp.expires_at
+            FROM latest_momo lm
+            JOIN competitor_prices cp
+              ON cp.sku = lm.sku
+             AND cp.source = 'pchome'
+             AND cp.competitor_product_id IS NOT NULL
+             AND cp.competitor_product_id <> ''
+             AND cp.expires_at IS NOT NULL
+             AND cp.expires_at <= CURRENT_TIMESTAMP
+             AND COALESCE(cp.match_score, 0) >= :match_score_floor
+             AND COALESCE(cp.tags, '[]'::jsonb) ? 'identity_v2'
+            WHERE lm.rn = 1
+              AND (
+                  COALESCE(cp.tags, '[]'::jsonb) ? 'price_basis_total_price'
+                  OR cp.match_diagnostic_json->>'price_basis' = 'total_price'
+              )
+              AND (
+                  COALESCE(cp.tags, '[]'::jsonb) ? 'alert_tier_price_alert_exact'
+                  OR cp.match_diagnostic_json->>'alert_tier' = 'price_alert_exact'
+              )
+              AND COALESCE(cp.match_diagnostic_json->>'comparison_mode', 'exact_identity') = 'exact_identity'
+              AND COALESCE(cp.hard_veto, false) = false
+              AND NOT (
+                  COALESCE(cp.match_diagnostic_json->'reasons', '[]'::jsonb)
+                  ?| array[{STALE_IDENTITY_RECOVERY_BLOCK_SQL_REASON_LIST}]
+              )
+              AND COALESCE(lm.name, '') !~* :blocked_name_pattern
+              AND COALESCE(cp.competitor_product_name, '') !~* :blocked_name_pattern
+            ORDER BY
+                cp.expires_at ASC,
+                lm.momo_price DESC NULLS LAST,
+                lm.sku
+            LIMIT :limit
+        """)
+        with self.engine.connect() as conn:
+            rows = conn.execute(
+                sql,
+                {
+                    "limit": max(1, min(int(limit), 120)),
+                    "match_score_floor": MIN_MATCH_SCORE,
+                    "blocked_name_pattern": STALE_IDENTITY_RECOVERY_BLOCK_NAME_PATTERN,
+                },
+            ).fetchall()
+        return [dict(r._mapping) for r in rows]
+
     def _fetch_expired_identity_skus(self, limit: int = 120) -> list:
         """
         取得 identity_v2 已確認、但 PChome 價格快取過期的商品。
@@ -2497,7 +2609,7 @@ class CompetitorPriceFeeder:
         safety、overwrite protection 才能寫入正式 competitor_prices。
         """
         try:
-            skus = self._fetch_expired_identity_skus(limit=max(1, min(int(limit), 120)))
+            skus = self._fetch_expired_identity_recovery_skus(limit=max(1, min(int(limit), 120)))
         except Exception as e:
             logger.error(f"[Feeder] 讀取過期 identity_v2 搜尋救援商品失敗: {e}")
             return FeederResult(0, 0, 0, 0, 1, 0.0)