feat(frontend): 保存 PChome 競品價格歷史

services/chart_generator_service.py

@@ -103,7 +103,7 @@ def _fetch_price_history(sku: str, days: int = 30) -> Dict[str, Any]:
 
             comp_rows = session.execute(text("""
                 SELECT crawled_at::date AS dt, AVG(price) AS price
-                FROM competitor_prices
+                FROM competitor_price_history
                 WHERE sku = :sku
                   AND source = 'pchome'
                   AND crawled_at >= NOW() - INTERVAL ':days days'
@@ -433,7 +433,7 @@ def price_history_heatmap(days: int = 30) -> Optional[bytes]:
                 SELECT p.category,
                        cp.crawled_at::date AS dt,
                        AVG((cp.price - pr.price) / NULLIF(pr.price, 0) * 100) AS gap_pct
-                FROM competitor_prices cp
+                FROM competitor_price_history cp
                 JOIN products p ON p.i_code = cp.sku
                 JOIN (
                     SELECT DISTINCT ON (product_id) product_id, price

services/competitor_price_feeder.py

@@ -5,7 +5,8 @@
 
 角色：獨立背景 Worker（生產者端）
 架構位置：
-  [本 Worker — 每 4 小時跑一次] → competitor_prices DB 表
+  [本 Worker — 每 4 小時跑一次] → competitor_prices DB 表（最新快取）
+                              → competitor_price_history DB 表（歷史快照）
                                         ↓
   [AI Pipeline] → fetch_candidates() LEFT JOIN competitor_prices（消費者端）
 
@@ -15,7 +16,7 @@
 - 語意化標籤 (tags) 讓 Hermes 獲得更豐富的情境
 
 爬取邏輯：
-  MOMO 商品名稱 → PChome 關鍵字搜尋 → 模糊比對最佳匹配 → 寫入 competitor_prices
+  MOMO 商品名稱 → PChome 關鍵字搜尋 → 模糊比對最佳匹配 → 寫入 competitor_prices + competitor_price_history
 
 依賴：
   services/pchome_crawler.py   — 搜尋 + 批量 API
@@ -47,6 +48,7 @@ class FeederResult:
     skipped_low_score: int
     errors: int
     duration_sec: float
+    history_written: int = 0
 
 
 def _extract_tags(pchome_product) -> list:
@@ -183,6 +185,68 @@ class CompetitorPriceFeeder:
 
     def __init__(self, engine=None):
         self.engine = engine
+        self._history_table_ready = False
+
+    def _ensure_competitor_price_history_table(self, conn):
+        """確保競品價格歷史表存在；排程可自癒補表，不依賴手動 migration。"""
+        if self._history_table_ready:
+            return
+
+        from sqlalchemy import text
+        if conn.dialect.name == "postgresql":
+            conn.execute(text("""
+                CREATE TABLE IF NOT EXISTS competitor_price_history (
+                    id BIGSERIAL PRIMARY KEY,
+                    sku VARCHAR(50) NOT NULL,
+                    source VARCHAR(30) NOT NULL DEFAULT 'pchome',
+                    momo_product_id INTEGER,
+                    momo_price NUMERIC(10,2),
+                    price NUMERIC(10,2) NOT NULL,
+                    original_price NUMERIC(10,2),
+                    discount_pct INTEGER,
+                    competitor_product_id VARCHAR(100),
+                    competitor_product_name TEXT,
+                    match_score NUMERIC(4,3),
+                    tags JSONB DEFAULT '[]'::jsonb,
+                    crawled_at TIMESTAMP NOT NULL DEFAULT NOW()
+                )
+            """))
+            conn.execute(text("""
+                CREATE INDEX IF NOT EXISTS idx_comp_price_history_sku_source_time
+                ON competitor_price_history (sku, source, crawled_at DESC)
+            """))
+            conn.execute(text("""
+                CREATE INDEX IF NOT EXISTS idx_comp_price_history_competitor_id
+                ON competitor_price_history (competitor_product_id)
+            """))
+        else:
+            conn.execute(text("""
+                CREATE TABLE IF NOT EXISTS competitor_price_history (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    sku VARCHAR(50) NOT NULL,
+                    source VARCHAR(30) NOT NULL DEFAULT 'pchome',
+                    momo_product_id INTEGER,
+                    momo_price NUMERIC(10,2),
+                    price NUMERIC(10,2) NOT NULL,
+                    original_price NUMERIC(10,2),
+                    discount_pct INTEGER,
+                    competitor_product_id VARCHAR(100),
+                    competitor_product_name TEXT,
+                    match_score NUMERIC(4,3),
+                    tags TEXT DEFAULT '[]',
+                    crawled_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+                )
+            """))
+            conn.execute(text("""
+                CREATE INDEX IF NOT EXISTS idx_comp_price_history_sku_source_time
+                ON competitor_price_history (sku, source, crawled_at DESC)
+            """))
+            conn.execute(text("""
+                CREATE INDEX IF NOT EXISTS idx_comp_price_history_competitor_id
+                ON competitor_price_history (competitor_product_id)
+            """))
+
+        self._history_table_ready = True
 
     def _fetch_active_skus(self) -> list:
         """
@@ -196,10 +260,25 @@ class CompetitorPriceFeeder:
 
         from sqlalchemy import text
         sql = text("""
-            SELECT DISTINCT p.i_code AS sku, p.name, p.category
+            SELECT
+                p.id AS product_id,
+                p.i_code AS sku,
+                p.name,
+                p.category,
+                (
+                    SELECT pr.price
+                    FROM price_records pr
+                    WHERE pr.product_id = p.id
+                    ORDER BY pr.timestamp DESC
+                    LIMIT 1
+                ) AS momo_price
             FROM products p
-            JOIN price_records pr ON pr.product_id = p.id
             WHERE p.status = 'ACTIVE'
+              AND EXISTS (
+                  SELECT 1
+                  FROM price_records pr
+                  WHERE pr.product_id = p.id
+              )
             ORDER BY p.i_code
         """)
         with self.engine.connect() as conn:
@@ -212,13 +291,17 @@ class CompetitorPriceFeeder:
         product,           # PChomeProduct
         match_score: float,
         tags: list,
+        momo_product_id: int = None,
+        momo_price: float = None,
         source: str = "pchome",
     ):
-        """單筆寫入/更新 competitor_prices"""
+        """單筆寫入/更新最新快取，並追加一筆歷史快照。"""
         from sqlalchemy import text
         _taipei = timezone(timedelta(hours=8))
         expires_at = (datetime.now(_taipei) + timedelta(hours=TTL_HOURS)).strftime("%Y-%m-%d %H:%M:%S")
+        tags_json = json.dumps(tags, ensure_ascii=False)
         with self.engine.begin() as conn:
+            self._ensure_competitor_price_history_table(conn)
             conn.execute(text("""
                 INSERT INTO competitor_prices
                     (sku, source, price, original_price, discount_pct,
@@ -247,9 +330,33 @@ class CompetitorPriceFeeder:
                 "comp_id":       product.product_id,
                 "comp_name":     product.name[:200],
                 "match_score":   match_score,
-                "tags":          json.dumps(tags, ensure_ascii=False),
+                "tags":          tags_json,
                 "expires_at":    expires_at,
             })
+            conn.execute(text("""
+                INSERT INTO competitor_price_history
+                    (sku, source, momo_product_id, momo_price,
+                     price, original_price, discount_pct,
+                     competitor_product_id, competitor_product_name,
+                     match_score, tags, crawled_at)
+                VALUES
+                    (:sku, :source, :momo_product_id, :momo_price,
+                     :price, :original_price, :discount_pct,
+                     :comp_id, :comp_name,
+                     :match_score, :tags, CURRENT_TIMESTAMP)
+            """), {
+                "sku":             sku,
+                "source":          source,
+                "momo_product_id": momo_product_id,
+                "momo_price":      momo_price,
+                "price":           product.price,
+                "original_price":  product.original_price,
+                "discount_pct":    product.discount,
+                "comp_id":         product.product_id,
+                "comp_name":       product.name[:200],
+                "match_score":     match_score,
+                "tags":            tags_json,
+            })
 
     def run(self, source: str = "pchome") -> FeederResult:
         """
@@ -283,10 +390,13 @@ class CompetitorPriceFeeder:
         skipped_no = 0
         skipped_low = 0
         errors = 0
+        history_written = 0
 
         for item in skus:
             sku      = item["sku"]
             momo_name = item["name"]
+            momo_product_id = item.get("product_id")
+            momo_price = item.get("momo_price")
 
             # 用商品名稱前 20 字搜尋（避免 query 過長）
             keyword = momo_name[:20].strip()
@@ -313,8 +423,17 @@ class CompetitorPriceFeeder:
                     continue
 
                 tags = _extract_tags(best_product)
-                self._upsert_competitor_price(sku, best_product, score, tags, source)
+                self._upsert_competitor_price(
+                    sku,
+                    best_product,
+                    score,
+                    tags,
+                    momo_product_id=momo_product_id,
+                    momo_price=momo_price,
+                    source=source,
+                )
                 matched += 1
+                history_written += 1
                 logger.debug(
                     f"[Feeder] {sku} → PChome ${best_product.price} "
                     f"score={score:.3f} tags={tags}"
@@ -327,7 +446,8 @@ class CompetitorPriceFeeder:
         duration = round(time.time() - start, 2)
         logger.info(
             f"[Feeder] 完成 matched={matched} skipped_no={skipped_no} "
-            f"skipped_low={skipped_low} errors={errors} 耗時={duration}s"
+            f"skipped_low={skipped_low} errors={errors} "
+            f"history_written={history_written} 耗時={duration}s"
         )
         return FeederResult(
             total_skus=len(skus),
@@ -336,6 +456,7 @@ class CompetitorPriceFeeder:
             skipped_low_score=skipped_low,
             errors=errors,
             duration_sec=duration,
+            history_written=history_written,
         )