feat(ai): 補抓 PChome 待比對商品

services/competitor_price_feeder.py

@@ -358,6 +358,47 @@ class CompetitorPriceFeeder:
             rows = conn.execute(sql).fetchall()
         return [dict(r._mapping) for r in rows]
 
+    def _fetch_unmatched_priority_skus(self, limit: int = 80) -> list:
+        """
+        取得目前沒有有效 PChome 配對的高價 ACTIVE 商品，供補強流程優先處理。
+        """
+        if self.engine is None:
+            raise RuntimeError("需要注入 SQLAlchemy engine")
+
+        from sqlalchemy import text
+        sql = text("""
+            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
+            FROM latest_momo lm
+            LEFT JOIN competitor_prices cp
+              ON cp.sku = lm.sku
+             AND cp.source = 'pchome'
+             AND (cp.expires_at IS NULL OR cp.expires_at > CURRENT_TIMESTAMP)
+            WHERE lm.rn = 1
+              AND cp.sku IS NULL
+            ORDER BY 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), 300))}).fetchall()
+        return [dict(r._mapping) for r in rows]
+
     def _upsert_competitor_price(
         self,
         sku: str,
@@ -431,16 +472,7 @@ class CompetitorPriceFeeder:
                 "tags":            tags_json,
             })
 
-    def run(self, source: str = "pchome") -> FeederResult:
-        """
-        執行一輪競品價格抓取與寫入
-
-        Args:
-            source: 競品來源代碼（目前支援 'pchome'）
-
-        Returns:
-            FeederResult
-        """
+    def _run_sku_items(self, skus: list, source: str = "pchome", label: str = "PChome 競品價格") -> FeederResult:
         start = time.time()
 
         if source != "pchome":
@@ -450,14 +482,7 @@ class CompetitorPriceFeeder:
         from services.pchome_crawler import PChomeCrawler
         crawler = PChomeCrawler(timeout=30, delay=RATE_DELAY)
 
-        # Step 1: 取得監控清單
-        try:
-            skus = self._fetch_active_skus()
-        except Exception as e:
-            logger.error(f"[Feeder] 讀取商品清單失敗: {e}")
-            return FeederResult(0, 0, 0, 0, 1, time.time() - start)
-
-        logger.info(f"[Feeder] 開始抓取 {len(skus)} 支商品的 PChome 競品價格")
+        logger.info(f"[Feeder] 開始抓取 {len(skus)} 支商品的 {label}")
 
         matched = 0
         skipped_no = 0
@@ -529,6 +554,34 @@ class CompetitorPriceFeeder:
             history_written=history_written,
         )
 
+    def run(self, source: str = "pchome") -> FeederResult:
+        """
+        執行一輪競品價格抓取與寫入
+
+        Args:
+            source: 競品來源代碼（目前支援 'pchome'）
+
+        Returns:
+            FeederResult
+        """
+        try:
+            skus = self._fetch_active_skus()
+        except Exception as e:
+            logger.error(f"[Feeder] 讀取商品清單失敗: {e}")
+            return FeederResult(0, 0, 0, 0, 1, 0.0)
+
+        return self._run_sku_items(skus, source=source, label="PChome 競品價格")
+
+    def run_unmatched_priority(self, limit: int = 80, source: str = "pchome") -> FeederResult:
+        """優先補抓尚未有有效 PChome 配對的高價商品。"""
+        try:
+            skus = self._fetch_unmatched_priority_skus(limit=limit)
+        except Exception as e:
+            logger.error(f"[Feeder] 讀取待比對優先商品失敗: {e}")
+            return FeederResult(0, 0, 0, 0, 1, 0.0)
+
+        return self._run_sku_items(skus, source=source, label="待比對優先補抓")
+
 
 # ─────────────────────────────────────────────
 # CLI 測試（不依賴 DB，直接測試爬蟲 + 比對邏輯）