"""美团 CPS 券定时抓取入库(北京试点)。

把 3 路券抓进 meituan_coupon 表,供「销量/佣金排序」从库里捞、本地排序,不再实时打美团:
  1. search_waimai : 到家/外卖, 搜「外卖」          翻到尽头
  2. search_meishi : 到家/外卖, 搜「美食」          翻到尽头
  3. store_supply  : 到店, 多业务线供给(到餐+到综+酒店+门票) 翻到尽头

按 (source, product_view_sign) upsert 存最新态;last_seen 每轮刷新。带文件锁,防止
上一轮没跑完下一轮又起(本地 5~10min、跨进程 cron 都安全)。

用法:
    # 单轮(打通验证 / 给 cron 用,线上每 1h 一次)
    python -m scripts.pull_meituan_coupons --once

    # 本地循环(默认每 10min 一轮)
    python -m scripts.pull_meituan_coupons --loop --interval 600
"""
from __future__ import annotations

import argparse
import hashlib
import os
import re
import sys
import tempfile
import time
from datetime import datetime, timedelta, timezone

# Windows 控制台按 UTF-8 输出中文/¥
try:
    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
except Exception:  # noqa: BLE001
    pass

from sqlalchemy import delete, func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert

from app.db.session import SessionLocal
from app.integrations.meituan import MeituanCpsError, _call
from app.models.meituan_coupon import MeituanCoupon

CITY_BEIJING = "WKV2HMXUEK634WP64CUCUQGM64"
QUERY_PATH = "/cps_open/common/api/v1/query_coupon"
PAGE_SIZE = 20
MAX_PAGES = 80          # 单路安全上限(搜索 ~52 页、供给 ~70 页)
PAGE_SLEEP = 0.35       # 页间配速,缓解 402
RETRY = 7
# 运行锁默认放系统临时目录(任何账号可写),不依赖代码目录 data/ 的写权限——
# 无 sudo 部署时 data/ 常属 root,cps 写不了会导致每轮 PermissionError、cron 抓不进数据。
# 需要指定位置时用环境变量 MEITUAN_ETL_LOCK 覆盖。
LOCK_FILE = os.environ.get("MEITUAN_ETL_LOCK") or os.path.join(tempfile.gettempdir(), "meituan_etl.lock")
LOCK_STALE_SEC = 30 * 60   # 锁超过 30min 视为陈旧(进程异常退出残留),自动接管

SOURCES = [
    {"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
    {"code": "search_meishi", "label": "外卖·搜美食", "kind": "search", "platform": 1, "keyword": "美食"},
    {"code": "store_supply", "label": "到店·多业务线供给", "kind": "supply",
     "platform": 2, "biz_lines": [1, 2, 3, 4]},
]


# ───────────────────────── 美团调用 ─────────────────────────

def _call_retry(body: dict) -> dict | None:
    """打美团,402/频繁退避重试;其它错误打印并放弃本页。"""
    for a in range(RETRY):
        try:
            return _call(QUERY_PATH, body)
        except MeituanCpsError as e:
            msg = str(e)
            if "402" in msg or "频繁" in msg:
                time.sleep(2.5 * (a + 1))
                continue
            print(f"   [warn] meituan: {msg[:80]}")
            return None
        except Exception as e:  # noqa: BLE001
            print(f"   [warn] {type(e).__name__}: {str(e)[:60]}")
            time.sleep(2.0 * (a + 1))
    return None


def _pull_search(platform: int, keyword: str) -> list[dict]:
    rows: list[dict] = []
    sid = None
    pg = 1
    while pg <= MAX_PAGES:
        body = {"platform": platform, "searchText": keyword, "cityId": CITY_BEIJING, "pageSize": PAGE_SIZE}
        if sid:
            body["searchId"] = sid
        else:
            body["pageNo"] = pg
        r = _call_retry(body)
        if not r:
            break
        sid = r.get("searchId")
        data = r.get("data") or []
        rows.extend(data)
        if not r.get("hasNext") or not data:
            break
        pg += 1
        time.sleep(PAGE_SLEEP)
    return rows


def _pull_supply(platform: int, biz_lines: list[int]) -> list[dict]:
    rows: list[dict] = []
    sid = None
    biz_param = [{"bizLine": b} for b in biz_lines]
    for _ in range(MAX_PAGES):
        body = {
            "multipleSupplyList": [{"platform": platform, "bizLineParamList": biz_param}],
            "cityId": CITY_BEIJING,
            "sortField": 2,  # 供给查询 sortField 必填;我们入库后本地再排,这里给个默认
            "pageSize": PAGE_SIZE,
        }
        if sid:
            body["searchId"] = sid
        r = _call_retry(body)
        if not r:
            break
        sid = r.get("searchId")
        data = r.get("data") or []
        rows.extend(data)
        if not r.get("hasNext") or not data:
            break
        time.sleep(PAGE_SLEEP)
    return rows


# ───────────────────────── 解析 ─────────────────────────

_SV_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(w|万|k)?", re.I)


def _sale_volume_num(s: str | None) -> int | None:
    """'热销1w+' → 10000, '热销500+' → 500(取下界,排序用)。"""
    if not s:
        return None
    m = _SV_RE.search(str(s))
    if not m:
        return None
    v = float(m.group(1))
    u = (m.group(2) or "").lower()
    if u in ("w", "万"):
        v *= 10000
    elif u == "k":
        v *= 1000
    return int(v)


def _to_cents(yuan) -> int | None:
    if yuan in (None, "", "null"):
        return None
    try:
        return round(float(yuan) * 100)
    except (TypeError, ValueError):
        return None


def _parse_item(item: dict, source: dict) -> dict | None:
    cpd = item.get("couponPackDetail") or {}
    br = item.get("brandInfo") or {}
    ci = item.get("commissionInfo") or {}
    poi = item.get("availablePoiInfo") or {}
    dp = item.get("deliverablePoiInfo") or {}

    sign = cpd.get("productViewSign") or cpd.get("skuViewId")
    if not sign:
        return None

    name = cpd.get("name") or ""
    brand = br.get("brandName") or ""
    price_cents = _to_cents(cpd.get("sellPrice"))

    comm_pct = None
    if ci.get("commissionPercent") is not None:
        try:
            comm_pct = float(ci["commissionPercent"]) / 100.0  # 140 → 1.4(%)
        except (TypeError, ValueError):
            comm_pct = None

    dist = dp.get("deliveryDistance")
    try:
        dist = float(dist) if dist not in (None, "", "null") else None
    except (TypeError, ValueError):
        dist = None

    dedup_raw = f"{brand}|{name}|{price_cents}"
    return {
        "source": source["code"],
        "platform": source["platform"],
        "biz_line": item.get("bizLine") or cpd.get("bizLine"),
        "city_id": CITY_BEIJING,
        "product_view_sign": str(sign)[:128],
        "sku_view_id": cpd.get("skuViewId"),
        "name": (name[:256] or None),
        "brand_name": (brand[:128] or None),
        "sell_price_cents": price_cents,
        "original_price_cents": _to_cents(cpd.get("originalPrice")),
        "head_url": ((cpd.get("headUrl") or "").split("@")[0][:512] or None),
        "sale_volume": cpd.get("saleVolume"),
        "sale_volume_num": _sale_volume_num(cpd.get("saleVolume")),
        "commission_percent": comm_pct,
        "commission_amount_cents": _to_cents(ci.get("commission")),
        "poi_name": ((dp.get("poiName") or "")[:128] or None),
        "available_poi_num": poi.get("availablePoiNum"),
        "delivery_distance_m": dist,
        "dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
        "raw": item,
    }


# ───────────────────────── 入库(upsert) ─────────────────────────

def _upsert(db, rows: list[dict], now: datetime) -> tuple[int, int]:
    """按 (source, product_view_sign) upsert。返回 (入库去重后条数, 本源内重复条数)。"""
    if not rows:
        return 0, 0
    dedup: dict[tuple, dict] = {}
    for r in rows:
        dedup[(r["source"], r["product_view_sign"])] = r  # 同轮内同键保留最后一条
    payload = list(dedup.values())
    for r in payload:
        r["last_seen"] = now
        r["updated_at"] = now

    chunk = 500
    for i in range(0, len(payload), chunk):
        part = payload[i:i + chunk]
        stmt = pg_insert(MeituanCoupon).values(part)
        update_cols = {
            c: getattr(stmt.excluded, c)
            for c in part[0]
            if c not in ("source", "product_view_sign", "first_seen")
        }
        stmt = stmt.on_conflict_do_update(
            constraint="uq_meituan_coupon_source_sign", set_=update_cols
        )
        db.execute(stmt)
        db.commit()
    return len(payload), len(rows) - len(payload)


# ───────────────────────── 运行锁 ─────────────────────────

def _acquire_lock() -> bool:
    os.makedirs(os.path.dirname(LOCK_FILE) or ".", exist_ok=True)
    try:
        fd = os.open(LOCK_FILE, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.write(fd, f"{os.getpid()} {time.time()}".encode())
        os.close(fd)
        return True
    except FileExistsError:
        try:
            with open(LOCK_FILE) as f:
                parts = f.read().split()
            ts = float(parts[1]) if len(parts) > 1 else 0.0
            if time.time() - ts > LOCK_STALE_SEC:
                os.remove(LOCK_FILE)
                return _acquire_lock()
        except Exception:  # noqa: BLE001
            pass
        return False


def _release_lock() -> None:
    try:
        os.remove(LOCK_FILE)
    except OSError:
        pass


# ───────────────────────── 主流程 ─────────────────────────

def run_once(prune_hours: int = 24) -> None:
    if not _acquire_lock():
        print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
        return
    t0 = time.time()
    now = datetime.now(timezone.utc)
    db = SessionLocal()
    try:
        total = 0
        for src in SOURCES:
            ts = time.time()
            if src["kind"] == "search":
                items = _pull_search(src["platform"], src["keyword"])
            else:
                items = _pull_supply(src["platform"], src["biz_lines"])
            parsed = [p for p in (_parse_item(it, src) for it in items) if p]
            up, dup = _upsert(db, parsed, now)
            total += up
            print(f"  {src['label']:18} 抓{len(items):5} 解析{len(parsed):5} "
                  f"入库{up:5} (本源去重{dup:3}) {time.time() - ts:4.0f}s")
        # 清理长期未再出现的陈旧券(美团 sign 轮换 / 券下架后的残留),默认 24h 宽限。
        # 护栏:仅在本轮确有入库(total>0)时才清理。美团整体故障时本轮可能 0 入库
        # (脚本不抛异常、只是抓回空),若仍照常 prune,连续故障会按 last_seen 把全表删空。
        if prune_hours and prune_hours > 0 and total > 0:
            cutoff = now - timedelta(hours=prune_hours)
            pruned = db.execute(
                delete(MeituanCoupon).where(MeituanCoupon.last_seen < cutoff)
            ).rowcount
            db.commit()
            if pruned:
                print(f"  清理陈旧券(>{prune_hours}h 未再出现): {pruned} 条")
        elif prune_hours and prune_hours > 0 and total == 0:
            print("  本轮 0 入库(疑似上游故障),跳过清理以防误删全表")
        cnt = db.execute(select(func.count()).select_from(MeituanCoupon)).scalar()
        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: 入库 {total} 条, 表总计 {cnt} 行, "
              f"用时 {time.time() - t0:.0f}s")
    finally:
        db.close()
        _release_lock()


def main() -> None:
    ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库")
    ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
    ap.add_argument("--loop", action="store_true", help="循环跑")
    ap.add_argument("--interval", type=int, default=600, help="循环间隔秒(默认 600=10min)")
    ap.add_argument("--prune-hours", type=int, default=24,
                    help="清理超过 N 小时未再出现的陈旧券(默认 24;0=不清理)")
    args = ap.parse_args()

    if args.loop:
        print(f"循环模式: 每 {args.interval}s 一轮 (Ctrl-C 退出)")
        while True:
            try:
                run_once(args.prune_hours)
            except Exception as e:  # noqa: BLE001
                print(f"[{datetime.now():%H:%M:%S}] 本轮异常: {type(e).__name__}: {e}")
                _release_lock()
            time.sleep(args.interval)
    else:
        run_once(args.prune_hours)


if __name__ == "__main__":
    main()