#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ AI 建議挑品 Agent 以真實 DB 資料建立可操作的 PChome 銷售挑品清單： - MOMO 最新價格 - PChome 最新競品價格與商品 ID - PChome 歷史快照 - 近 7 天銷售資料（若 daily_sales_snapshot 可用） 此 Agent 不補假資料；資料不足的欄位只降低分數或略過。 """ import json import logging import os from dataclasses import dataclass from datetime import datetime from typing import Any, Dict, List logger = logging.getLogger(__name__) @dataclass class ProductPickResult: candidates: int written: int picks: List[Dict[str, Any]] generated_at: str def _to_float(value, default=0.0) -> float: if value is None: return default try: return float(value) except (TypeError, ValueError): return default def _load_json_tags(value) -> List[str]: if not value: return [] if isinstance(value, list): return value try: parsed = json.loads(value) return parsed if isinstance(parsed, list) else [] except Exception: return [] def _has_daily_sales_snapshot(conn) -> bool: from sqlalchemy import text try: if conn.dialect.name == "postgresql": row = conn.execute(text("SELECT to_regclass('daily_sales_snapshot') AS table_name")).mappings().first() return bool(row and row.get("table_name")) row = conn.execute(text(""" SELECT name FROM sqlite_master WHERE type='table' AND name='daily_sales_snapshot' """)).first() return bool(row) except Exception: return False def _daily_sales_columns(conn) -> Dict[str, str]: """依正式匯入表實際欄位挑選可用欄名。""" from sqlalchemy import text if conn.dialect.name == "postgresql": rows = conn.execute(text(""" SELECT column_name FROM information_schema.columns WHERE table_name = 'daily_sales_snapshot' """)).fetchall() columns = {row[0] for row in rows} elif conn.dialect.name == "sqlite": rows = conn.execute(text("PRAGMA table_info(daily_sales_snapshot)")).fetchall() columns = {row[1] for row in rows} else: result = conn.execute(text("SELECT * FROM daily_sales_snapshot LIMIT 0")) columns = set(result.keys()) def first_available(candidates): return next((col for col in candidates if col in columns), None) return { "sku": first_available(["商品ID", "Product ID", "ID", "i_code", "Item Code"]), "date": first_available(["snapshot_date", "日期", "訂單日期", "交易日期", "Date"]), "revenue": first_available(["總業績", "銷售金額", "業績", "金額", "Amount", "Sales", "Total"]), "qty": first_available(["數量", "銷售數量", "銷量", "Qty", "Quantity"]), "profit": first_available(["毛利", "Profit", "利潤"]), "cost": first_available(["總成本", "成本", "Cost", "進價"]), } def _quote_identifier(identifier: str) -> str: return '"' + identifier.replace('"', '""') + '"' def _identity_match_condition(conn, alias: str = "cp") -> str: if conn.dialect.name == "postgresql": return f"AND COALESCE({alias}.tags, '[]'::jsonb) ? 'identity_v2'" return f"AND COALESCE({alias}.tags, '') LIKE '%identity_v2%'" def _sales_join_by_momo_sku_enabled() -> bool: """舊 MOMO SKU 直連銷售表的路徑預設關閉，避免把 PChome 業績 ID 誤當 MOMO SKU。""" return os.getenv("PCHOME_SALES_JOIN_BY_MOMO_SKU_ENABLED", "false").strip().lower() in { "1", "true", "yes", "on", } def _fetch_candidates_without_sales(conn, limit: int) -> List[Dict[str, Any]]: """DB-portable fallback query used when sales-aware PostgreSQL SQL is unavailable.""" from sqlalchemy import text identity_condition = _identity_match_condition(conn, "cp") fallback = text(f""" WITH latest_momo AS ( SELECT p.id AS product_id, p.i_code AS sku, p.name, p.url, 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.url, lm.category, lm.momo_price, cp.price AS pchome_price, cp.original_price, cp.discount_pct, cp.competitor_product_id, cp.competitor_product_name, cp.match_score, cp.tags, cp.crawled_at, 0 AS history_points, NULL AS min_pchome_price, NULL AS max_pchome_price, 0 AS sales_7d, 0 AS sales_prev_7d, 0 AS qty_7d, 0 AS profit_7d, 0 AS cost_7d FROM latest_momo lm 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) AND cp.match_score >= 0.76 {identity_condition} WHERE lm.rn = 1 ORDER BY cp.match_score DESC, cp.crawled_at DESC LIMIT :limit """) return [dict(row) for row in conn.execute(fallback, {"limit": max(limit * 6, 100)}).mappings().all()] def _fetch_candidates(conn, limit: int) -> List[Dict[str, Any]]: from sqlalchemy import text if conn.dialect.name != "postgresql": return _fetch_candidates_without_sales(conn, limit) sales_join = "" sales_select = "0 AS sales_7d, 0 AS sales_prev_7d, 0 AS qty_7d, 0 AS profit_7d, 0 AS cost_7d" sales_cols = {} if _sales_join_by_momo_sku_enabled() and _has_daily_sales_snapshot(conn): try: sales_cols = _daily_sales_columns(conn) except Exception as exc: logger.warning("[ProductPickAgent] daily_sales_snapshot column probe failed: %s", exc) sales_cols = {} if not all([sales_cols.get("sku"), sales_cols.get("date"), sales_cols.get("revenue"), sales_cols.get("qty")]): sales_cols = {} if sales_cols: sku_col = _quote_identifier(sales_cols["sku"]) date_col = _quote_identifier(sales_cols["date"]) revenue_col = _quote_identifier(sales_cols["revenue"]) qty_col = _quote_identifier(sales_cols["qty"]) profit_col = _quote_identifier(sales_cols["profit"]) if sales_cols.get("profit") else None cost_col = _quote_identifier(sales_cols["cost"]) if sales_cols.get("cost") else None profit_expr = f"COALESCE({profit_col}::numeric, 0)" if profit_col else "0" cost_expr = f"COALESCE({cost_col}::numeric, 0)" if cost_col else "0" sales_join = """ LEFT JOIN ( SELECT {sku_col} AS sku, SUM(CASE WHEN {date_col}::date >= CURRENT_DATE - 7 THEN COALESCE({revenue_col}::numeric, 0) ELSE 0 END) AS sales_7d, SUM(CASE WHEN {date_col}::date >= CURRENT_DATE - 14 AND {date_col}::date < CURRENT_DATE - 7 THEN COALESCE({revenue_col}::numeric, 0) ELSE 0 END) AS sales_prev_7d, SUM(CASE WHEN {date_col}::date >= CURRENT_DATE - 7 THEN COALESCE({qty_col}::numeric, 0) ELSE 0 END) AS qty_7d, SUM(CASE WHEN {date_col}::date >= CURRENT_DATE - 7 THEN {profit_expr} ELSE 0 END) AS profit_7d, SUM(CASE WHEN {date_col}::date >= CURRENT_DATE - 7 THEN {cost_expr} ELSE 0 END) AS cost_7d FROM daily_sales_snapshot GROUP BY {sku_col} ) sales ON sales.sku = lm.sku """.format( sku_col=sku_col, date_col=date_col, revenue_col=revenue_col, qty_col=qty_col, profit_expr=profit_expr, cost_expr=cost_expr, ) sales_select = """ COALESCE(sales.sales_7d, 0) AS sales_7d, COALESCE(sales.sales_prev_7d, 0) AS sales_prev_7d, COALESCE(sales.qty_7d, 0) AS qty_7d, COALESCE(sales.profit_7d, 0) AS profit_7d, COALESCE(sales.cost_7d, 0) AS cost_7d """ identity_condition = _identity_match_condition(conn, "cp") sql = text(f""" WITH latest_momo AS ( SELECT p.id AS product_id, p.i_code AS sku, p.name, p.url, 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' ), history_stats AS ( SELECT sku, source, COUNT(*) AS history_points, MIN(price) AS min_pchome_price, MAX(price) AS max_pchome_price FROM competitor_price_history WHERE source = 'pchome' AND crawled_at >= CURRENT_TIMESTAMP - INTERVAL '30 days' AND COALESCE(match_score, 0) >= 0.76 AND COALESCE(tags, '[]'::jsonb) ? 'identity_v2' GROUP BY sku, source ) SELECT lm.product_id, lm.sku, lm.name, lm.url, lm.category, lm.momo_price, cp.price AS pchome_price, cp.original_price, cp.discount_pct, cp.competitor_product_id, cp.competitor_product_name, cp.match_score, cp.tags, cp.crawled_at, COALESCE(hs.history_points, 0) AS history_points, hs.min_pchome_price, hs.max_pchome_price, {sales_select} FROM latest_momo lm 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) AND cp.match_score >= 0.76 {identity_condition} LEFT JOIN history_stats hs ON hs.sku = lm.sku AND hs.source = cp.source {sales_join} WHERE lm.rn = 1 ORDER BY cp.match_score DESC, cp.crawled_at DESC LIMIT :limit """) try: return [dict(row) for row in conn.execute(sql, {"limit": max(limit * 6, 100)}).mappings().all()] except Exception as exc: logger.warning("[ProductPickAgent] sales-aware query failed, fallback without sales: %s", exc) try: conn.rollback() except Exception: logger.debug("[ProductPickAgent] rollback after sales-aware query failure failed", exc_info=True) return _fetch_candidates_without_sales(conn, limit) def _score_candidate(row: Dict[str, Any]) -> Dict[str, Any]: momo_price = _to_float(row.get("momo_price")) pchome_price = _to_float(row.get("pchome_price")) match_score = _to_float(row.get("match_score")) sales_7d = _to_float(row.get("sales_7d")) sales_prev_7d = _to_float(row.get("sales_prev_7d")) qty_7d = _to_float(row.get("qty_7d")) profit_7d = _to_float(row.get("profit_7d")) cost_7d = _to_float(row.get("cost_7d")) history_points = int(_to_float(row.get("history_points"))) min_pchome_price = _to_float(row.get("min_pchome_price")) tags = _load_json_tags(row.get("tags")) gap_pct = ((momo_price - pchome_price) / pchome_price * 100) if pchome_price else 0 sales_delta = ((sales_7d - sales_prev_7d) / sales_prev_7d * 100) if sales_prev_7d else None if not profit_7d and cost_7d and sales_7d: profit_7d = sales_7d - cost_7d margin_rate = (profit_7d / sales_7d * 100) if sales_7d and profit_7d else None price_score = max(0, min(40, gap_pct * 1.9 + 8)) match_component = max(0, min(30, match_score * 30)) sales_component = 0 if sales_7d > 0: sales_component += min(9, sales_7d / 30000 * 9) if qty_7d > 0: sales_component += min(4, qty_7d / 20 * 4) if sales_delta is not None and sales_delta > 0: sales_component += min(7, sales_delta / 40 * 7) margin_component = 0 if margin_rate is not None: margin_component = max(0, min(10, margin_rate / 35 * 10)) history_component = min(12, history_points * 2.4) promo_component = 0 if any(tag in tags for tag in ["on_sale", "discount_10pct", "discount_20pct", "discount_30pct"]): promo_component += 5 if "high_rating" in tags: promo_component += 3 if "low_stock" in tags: promo_component -= 4 price_position_component = 0 if min_pchome_price and pchome_price: if pchome_price <= min_pchome_price * 1.03: price_position_component = 6 elif pchome_price <= min_pchome_price * 1.08: price_position_component = 3 opportunity_score = min( 100, price_score + sales_component + margin_component + promo_component + price_position_component, ) evidence_quality = min( 100, match_component + history_component + (12 if sales_7d > 0 else 0) + (8 if margin_rate is not None else 0) + (8 if row.get("competitor_product_id") and row.get("competitor_product_name") else 0) + (6 if row.get("crawled_at") else 0), ) score = round(min(100, opportunity_score + evidence_quality * 0.35), 1) confidence = round(max(0.45, min(0.98, (score * 0.65 + evidence_quality * 0.35) / 100)), 3) missing_evidence = [] if history_points < 3: missing_evidence.append("PChome 價格歷史不足 3 筆") if sales_7d <= 0: missing_evidence.append("近 7 天銷售額缺口") if qty_7d <= 0: missing_evidence.append("近 7 天銷量缺口") if margin_rate is None: missing_evidence.append("毛利/成本缺口") if not row.get("competitor_product_id"): missing_evidence.append("PChome 商品 ID 缺口") if confidence >= 0.78 and evidence_quality >= 70: confidence_band = "high" elif confidence >= 0.65 and evidence_quality >= 55: confidence_band = "medium" else: confidence_band = "needs_evidence" if gap_pct >= 10: angle = "PChome 價格優勢明顯" elif gap_pct >= 3: angle = "PChome 小幅價格優勢" elif sales_7d > 0: angle = "近期有銷售動能，可搭配內容或檔期測試" else: angle = "比對信心足夠，可列入觀察型挑品" reason_parts = [ f"{angle}，PChome ${pchome_price:,.0f} vs MOMO ${momo_price:,.0f}", f"價差 {gap_pct:+.1f}%", f"比對信心 {match_score:.2f}", ] if sales_7d > 0: reason_parts.append(f"近 7 天銷售額 ${sales_7d:,.0f}") if margin_rate is not None: reason_parts.append(f"近 7 天毛利率 {margin_rate:.1f}%") if history_points: reason_parts.append(f"已有 {history_points} 筆 PChome 歷史快照") if price_position_component: reason_parts.append("目前 PChome 價格接近 30 天低點") if "high_rating" in tags: reason_parts.append("PChome 商品評價訊號佳") if "low_stock" in tags: reason_parts.append("PChome 庫存偏低，需留意供貨") if missing_evidence: reason_parts.append("待補證據：" + "、".join(missing_evidence[:3])) return { **row, "gap_pct": round(gap_pct, 1), "sales_7d_delta": round(sales_delta, 1) if sales_delta is not None else 0, "pick_score": score, "confidence": confidence, "evidence_quality": round(evidence_quality, 1), "opportunity_score": round(opportunity_score, 1), "margin_rate": round(margin_rate, 1) if margin_rate is not None else None, "confidence_band": confidence_band, "missing_evidence": missing_evidence, "reason": "；".join(reason_parts), } def _write_pick(conn, pick: Dict[str, Any]) -> None: from sqlalchemy import text footprint = { "agent": { "name": "PChomeProductPickAgent", "version": "v1", "generated_at": datetime.now().isoformat(timespec="seconds"), "inputs": ["products", "price_records", "competitor_prices", "competitor_price_history", "daily_sales_snapshot"], "score": pick["pick_score"], "opportunity_score": pick.get("opportunity_score"), "evidence_quality": pick.get("evidence_quality"), "margin_rate": pick.get("margin_rate"), "confidence_band": pick.get("confidence_band"), "missing_evidence": pick.get("missing_evidence", []), }, "competitor": { "source": "pchome", "product_id": pick.get("competitor_product_id"), "product_name": pick.get("competitor_product_name"), "match_score": _to_float(pick.get("match_score")), }, } conn.execute(text(""" INSERT INTO ai_price_recommendations (sku, name, reason, strategy, confidence, momo_price, pchome_price, gap_pct, sales_7d_delta, model_footprint, status, created_at, updated_at) VALUES (:sku, :name, :reason, 'product_pick', :confidence, :momo_price, :pchome_price, :gap_pct, :sales_7d_delta, :footprint, 'pending', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT (sku) DO UPDATE SET reason = EXCLUDED.reason, strategy = 'product_pick', confidence = EXCLUDED.confidence, momo_price = EXCLUDED.momo_price, pchome_price = EXCLUDED.pchome_price, gap_pct = EXCLUDED.gap_pct, sales_7d_delta = EXCLUDED.sales_7d_delta, model_footprint = EXCLUDED.model_footprint, status = 'pending', updated_at = CURRENT_TIMESTAMP """), { "sku": pick["sku"], "name": pick["name"], "reason": pick["reason"], "confidence": pick["confidence"], "momo_price": pick["momo_price"], "pchome_price": pick["pchome_price"], "gap_pct": pick["gap_pct"], "sales_7d_delta": pick["sales_7d_delta"], "footprint": json.dumps(footprint, ensure_ascii=False), }) def _supersede_old_picks(conn, current_skus: List[str]) -> None: from sqlalchemy import bindparam, text if not current_skus: conn.execute(text(""" UPDATE ai_price_recommendations SET status = 'superseded', updated_at = CURRENT_TIMESTAMP WHERE strategy = 'product_pick' AND status = 'pending' """)) return stmt = text(""" UPDATE ai_price_recommendations SET status = 'superseded', updated_at = CURRENT_TIMESTAMP WHERE strategy = 'product_pick' AND status = 'pending' AND sku NOT IN :current_skus """).bindparams(bindparam("current_skus", expanding=True)) conn.execute(stmt, {"current_skus": [str(sku) for sku in current_skus]}) def generate_product_pick_list(engine, limit: int = 50) -> ProductPickResult: """產生並保存 AI 建議挑品清單。""" generated_at = datetime.now().isoformat(timespec="seconds") with engine.connect() as conn: rows = _fetch_candidates(conn, limit) with engine.begin() as conn: scored = [_score_candidate(row) for row in rows if _to_float(row.get("pchome_price")) > 0] picks = [ pick for pick in scored if pick["pick_score"] >= 45 and (_to_float(pick.get("match_score")) >= 0.76) ] picks.sort(key=lambda item: item["pick_score"], reverse=True) picks = picks[:limit] for pick in picks: _write_pick(conn, pick) _supersede_old_picks(conn, [pick["sku"] for pick in picks]) return ProductPickResult( candidates=len(rows), written=len(picks), picks=picks, generated_at=generated_at, )