336 lines
12 KiB
Python
336 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
AI 建議挑品 Agent
|
||
|
||
以真實 DB 資料建立可操作的 PChome 銷售挑品清單：
|
||
- MOMO 最新價格
|
||
- PChome 最新競品價格與商品 ID
|
||
- PChome 歷史快照
|
||
- 近 7 天銷售資料（若 daily_sales_snapshot 可用）
|
||
|
||
此 Agent 不補假資料；資料不足的欄位只降低分數或略過。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
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 _fetch_candidates(conn, limit: int) -> List[Dict[str, Any]]:
|
||
from sqlalchemy import text
|
||
|
||
sales_join = ""
|
||
sales_select = "0 AS sales_7d, 0 AS sales_prev_7d, 0 AS qty_7d"
|
||
if _has_daily_sales_snapshot(conn):
|
||
sales_join = """
|
||
LEFT JOIN (
|
||
SELECT
|
||
"商品ID" AS sku,
|
||
SUM(CASE WHEN snapshot_date >= CURRENT_DATE - 7
|
||
THEN COALESCE("銷售金額"::numeric, 0) ELSE 0 END) AS sales_7d,
|
||
SUM(CASE WHEN snapshot_date >= CURRENT_DATE - 14
|
||
AND snapshot_date < CURRENT_DATE - 7
|
||
THEN COALESCE("銷售金額"::numeric, 0) ELSE 0 END) AS sales_prev_7d,
|
||
SUM(CASE WHEN snapshot_date >= CURRENT_DATE - 7
|
||
THEN COALESCE("數量"::numeric, 0) ELSE 0 END) AS qty_7d
|
||
FROM daily_sales_snapshot
|
||
GROUP BY "商品ID"
|
||
) sales ON sales.sku = lm.sku
|
||
"""
|
||
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
|
||
"""
|
||
|
||
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'
|
||
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.42
|
||
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)
|
||
fallback = text("""
|
||
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
|
||
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.42
|
||
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 _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"))
|
||
history_points = int(_to_float(row.get("history_points")))
|
||
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
|
||
|
||
price_score = max(0, min(38, gap_pct * 1.8 + 8))
|
||
match_component = max(0, min(24, match_score * 24))
|
||
sales_component = 0
|
||
if sales_7d > 0:
|
||
sales_component += min(10, sales_7d / 30000 * 10)
|
||
if qty_7d > 0:
|
||
sales_component += min(5, qty_7d / 20 * 5)
|
||
if sales_delta is not None and sales_delta > 0:
|
||
sales_component += min(8, sales_delta / 40 * 8)
|
||
history_component = min(10, history_points * 2)
|
||
promo_component = 5 if any(tag in tags for tag in ["on_sale", "discount_10pct", "discount_20pct", "discount_30pct"]) else 0
|
||
score = round(min(100, price_score + match_component + sales_component + history_component + promo_component), 1)
|
||
|
||
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 history_points:
|
||
reason_parts.append(f"已有 {history_points} 筆 PChome 歷史快照")
|
||
|
||
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": round(max(0.45, min(0.98, score / 100)), 3),
|
||
"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"],
|
||
},
|
||
"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 generate_product_pick_list(engine, limit: int = 30) -> ProductPickResult:
|
||
"""產生並保存 AI 建議挑品清單。"""
|
||
generated_at = datetime.now().isoformat(timespec="seconds")
|
||
with engine.begin() as conn:
|
||
rows = _fetch_candidates(conn, limit)
|
||
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.42)
|
||
]
|
||
picks.sort(key=lambda item: item["pick_score"], reverse=True)
|
||
picks = picks[:limit]
|
||
for pick in picks:
|
||
_write_pick(conn, pick)
|
||
|
||
return ProductPickResult(
|
||
candidates=len(rows),
|
||
written=len(picks),
|
||
picks=picks,
|
||
generated_at=generated_at,
|
||
)
|