"""美团 CPS 券定时抓取入库(全国 359 个地级市)。

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

城市来自 app/integrations/data/meituan_cities.json(美团官方城市字典,359 地级市)。
实测一个地级市 cityId 已覆盖其下辖县级市(徐州 → 邳州/新沂…),故按地级市抓即可。

并发:城市级并发(每城内部仍顺序跑 3 路),主线程串行入库(Session 不跨线程)。
实测单城全量 ~110s/~2300 条;15 并发抓完全国一轮 ~50–60min,402 占比 ~3% 且退避全消化。
mentor 要求每 3h 全量一轮,窗口充裕,故默认 12 并发 + 启动错峰,把瞬时峰值与 402 压更低。

按 (source, product_view_sign) upsert 存最新态;last_seen 每轮刷新。带文件锁,防止
上一轮没跑完下一轮又起。

用法:
    # 单轮全量(给 cron 用,线上每 3h 一次)
    python -m scripts.pull_meituan_coupons --once

    # 常驻循环(每 3h 一轮)
    python -m scripts.pull_meituan_coupons --loop --interval 10800

    # 本地测试:只抓前 N 个城市 / 指定城市
    python -m scripts.pull_meituan_coupons --once --limit 3
    python -m scripts.pull_meituan_coupons --once --city-ids OCZOBCJDEXKE7KBN3BD7AYQG2Q

部署(服务器):推荐 cron 跑 --once(每 3h),避免长驻进程孤儿:
    0 */3 * * *  cd /path/to/app-server && .venv/bin/python -m scripts.pull_meituan_coupons --once >> data/etl.log 2>&1
"""
from __future__ import annotations

import argparse
import hashlib
import logging
import os
import re
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone

# Windows 控制台按 UTF-8 输出中文/¥;line_buffering=True 让 print 每行即时 flush——
# 否则 stdout 重定向到 cron/后台日志文件时是块缓冲,要攒到 ~4KB 或进程退出才落盘,
# 常驻(--loop)时几乎看不到每轮进度。
try:
    sys.stdout.reconfigure(encoding="utf-8", line_buffering=True)  # type: ignore[attr-defined]
except Exception:  # noqa: BLE001
    pass

import httpx  # noqa: E402
from sqlalchemy import delete, func, select  # noqa: E402
from sqlalchemy.dialects.postgresql import insert as pg_insert  # noqa: E402

from app.db.session import SessionLocal, engine  # noqa: E402
from app.integrations.cities import all_cities  # noqa: E402
from app.integrations.meituan import MeituanCpsError, _call  # noqa: E402
from app.models.meituan_coupon import MeituanCoupon  # noqa: E402

# dev 默认 create_engine(echo=True) 会逐条打 SQL,本脚本逐城 upsert 会把日志刷爆;
# ETL 不需要 SQL 日志,显式关掉(入库不受影响;线上 prod 本就 echo=False)。
engine.echo = False

# 美团调用偶发错误(本机走代理高并发时的 SSL EOF / 上游 code=5 等)会被 meituan._call 的
# logger.exception 打完整 traceback,并发抓取下单轮可刷数十 KB 日志。ETL 自己用 _STATS 统计
# 「放弃页数」,无需逐条 traceback,故把美团 logger 压到 CRITICAL(线上直连少见此类错误)。
logging.getLogger("shagua.meituan").setLevel(logging.CRITICAL)

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")
# 多城全量一轮 ~50–60min,锁陈旧阈值放宽到 90min,避免把「正在跑的轮次」误判为残留而接管。
LOCK_STALE_SEC = 90 * 60

DEFAULT_CONCURRENCY = 12   # 并发城市数(实测 15 并发 402 占 3% 可退避消化;12 更稳,3h 窗口充裕)
STARTUP_STAGGER = 0.3      # 首批城市启动错峰间隔秒(削平瞬时峰值,实测能压低 402)
PRUNE_FAIL_RATIO_MAX = 0.05  # 失败城占比超此值则本轮跳过 prune(避免大面积抓取失败误删库)

# 头图大小/类型富集:每城抓完后,按 head_url 去重并发 HEAD 取 Content-Length / Content-Type,
# 写进 image_size / image_type。图片走美团图片 CDN(p*.meituan.net / img.meituan.net),直连可达、
# 与 CPS 网关 402 限流无关;任何失败 → None,不阻断入库。城市级已 12 并发,这里每城再开 8,
# 叠加峰值 ~96 个 HEAD,CDN 扛得住(实测单进程 50 并发稳)。
IMAGE_META_WORKERS = 8     # 每城 HEAD 并发数
IMAGE_META_TIMEOUT = 8.0   # 单图 HEAD 超时秒

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]},
]

# 跨线程统计(并发抓取下汇总请求量 / 402 / 放弃页数,供观测限流)
_STATS_LOCK = threading.Lock()
_STATS = {"req": 0, "r402": 0, "err": 0}


def _bump(key: str) -> None:
    with _STATS_LOCK:
        _STATS[key] += 1


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

def _call_retry(body: dict) -> dict | None:
    """打美团,402/频繁退避重试;其它错误放弃本页。并发下不逐条 print(避免刷屏),计入 _STATS。"""
    for a in range(RETRY):
        _bump("req")
        try:
            return _call(QUERY_PATH, body)
        except MeituanCpsError as e:
            msg = str(e)
            if "402" in msg or "频繁" in msg:
                _bump("r402")
                time.sleep(2.5 * (a + 1))
                continue
            _bump("err")
            return None
        except Exception:  # noqa: BLE001
            time.sleep(2.0 * (a + 1))
    _bump("err")
    return None


def _pull_search(city_id: str, platform: int, keyword: str) -> list[dict]:
    rows: list[dict] = []
    sid = None
    pg = 1
    while pg <= MAX_PAGES:
        body = {"platform": platform, "searchText": keyword, "cityId": city_id, "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(city_id: str, 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_id,
            "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 _strip_nul(v):
    """递归去掉字符串里的 NUL(0x00):PostgreSQL 的 text / jsonb 字段都不接受该字节,
    美团偶有脏数据(某券文本含 NUL)会让整批 upsert 抛 DataError。入库前统一清洗。"""
    if isinstance(v, str):
        return v.replace(chr(0), "")
    if isinstance(v, dict):
        return {k: _strip_nul(x) for k, x in v.items()}
    if isinstance(v, list):
        return [_strip_nul(x) for x in v]
    return v


def _parse_item(item: dict, source: dict, city_id: str) -> 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}"
    # 入库前递归清洗 NUL(美团脏数据偶含 0x00,PostgreSQL text/jsonb 拒绝整批 → DataError)
    return _strip_nul({
        "source": source["code"],
        "platform": source["platform"],
        "biz_line": item.get("bizLine") or cpd.get("bizLine"),
        "city_id": city_id,
        "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),
        # 占位;由 _fill_image_meta 抓完本城后并发 HEAD 回填(键必须在,upsert 才会带上这两列)
        "image_size": None,
        "image_type": 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,
    })


def _head_image_meta(client: httpx.Client, url: str) -> tuple[int | None, str | None]:
    """HEAD 单张头图,取 (字节大小, MIME 类型);任何失败返回 (None, None),不阻断入库。"""
    try:
        r = client.head(url, timeout=IMAGE_META_TIMEOUT, follow_redirects=True)
        if r.status_code >= 400:
            return None, None
        cl = r.headers.get("content-length")
        size = int(cl) if cl and cl.isdigit() else None
        ctype = (r.headers.get("content-type") or "").split(";")[0].strip()[:32] or None
        return size, ctype
    except Exception:  # noqa: BLE001
        return None, None


def _fill_image_meta(rows: list[dict]) -> None:
    """就地为本城 rows 填 image_size / image_type:按 head_url 去重后并发 HEAD,再回填每行。
    trust_env=False 直连图片 CDN(绕本机代理,CDN 无需走 CPS 代理)。"""
    urls = {r["head_url"] for r in rows if r.get("head_url")}
    if not urls:
        return
    meta: dict[str, tuple[int | None, str | None]] = {}
    with httpx.Client(trust_env=False, timeout=IMAGE_META_TIMEOUT) as client:
        with ThreadPoolExecutor(max_workers=IMAGE_META_WORKERS) as pool:
            futs = {pool.submit(_head_image_meta, client, u): u for u in urls}
            for f in as_completed(futs):
                meta[futs[f]] = f.result()
    for r in rows:
        m = meta.get(r.get("head_url"))
        if m:
            r["image_size"], r["image_type"] = m


def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
    """抓单个城市的 3 路券并解析。worker 线程内执行(只抓取+解析,不碰 DB)。

    启动错峰:首批并发的 `concurrency` 个城市按 index 错开首请求,削平瞬时峰值(压低 402)。
    """
    if stagger:
        time.sleep((index % concurrency) * stagger)
    cid = city["city_id"]
    parsed: list[dict] = []
    for src in SOURCES:
        if src["kind"] == "search":
            items = _pull_search(cid, src["platform"], src["keyword"])
        else:
            items = _pull_supply(cid, src["platform"], src["biz_lines"])
        for it in items:
            p = _parse_item(it, src, cid)
            if p:
                parsed.append(p)
    _fill_image_meta(parsed)  # 抓完本城 → 并发 HEAD 回填头图 image_size/image_type
    return parsed


# ───────────────────────── 入库(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,
    concurrency: int = DEFAULT_CONCURRENCY,
    stagger: float = STARTUP_STAGGER,
    cities: list[dict] | None = None,
) -> None:
    if not _acquire_lock():
        print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
        return
    t0 = time.time()
    now = datetime.now(timezone.utc)
    with _STATS_LOCK:
        _STATS.update(req=0, r402=0, err=0)
    cities = cities if cities is not None else all_cities()
    n_city = len(cities)
    db = SessionLocal()
    try:
        total_up = 0
        done = 0
        fails: list[str] = []
        # 城市级并发抓取(worker 只抓取+解析),主线程逐城串行入库(Session 不跨线程)。
        with ThreadPoolExecutor(max_workers=concurrency) as pool:
            futs = {
                pool.submit(_pull_one_city, city, i, concurrency, stagger): city
                for i, city in enumerate(cities)
            }
            for fut in as_completed(futs):
                city = futs[fut]
                try:
                    parsed = fut.result()
                except Exception as e:  # noqa: BLE001
                    fails.append(city["name"])
                    print(f"  [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
                    continue
                try:
                    up, _dup = _upsert(db, parsed, now)
                except Exception as e:  # noqa: BLE001
                    db.rollback()  # 事务已 abort,必须 rollback 才能继续给下一城复用 Session
                    fails.append(city["name"])
                    print(f"  [入库失败] {city['name']}: {type(e).__name__}: {str(e)[:80]}")
                    continue
                total_up += up
                done += 1
                if done % 20 == 0 or done == n_city:
                    print(f"  [{done}/{n_city}] {city['name']:10} 抓{len(parsed):5} 入库{up:5} "
                          f"(累计入库 {total_up}, 用时 {time.time() - t0:.0f}s)")

        # 清理陈旧券(美团 sign 轮换 / 券下架残留),默认 24h 宽限。两道护栏防误删:
        #  ① total_up>0:本轮 0 入库(疑似上游整体故障,脚本不抛异常只抓回空)时跳过,否则
        #     会按 last_seen 把全表删空;
        #  ② 失败城占比 ≤5%:大面积城市抓取失败(限流/网络)时跳过,避免误删还在架上的券。
        fail_ratio = len(fails) / max(1, n_city)
        if prune_hours and prune_hours > 0 and total_up > 0 and fail_ratio <= PRUNE_FAIL_RATIO_MAX:
            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:
            reason = "本轮 0 入库" if total_up == 0 else f"失败城占比 {fail_ratio:.0%}>{PRUNE_FAIL_RATIO_MAX:.0%}"
            print(f"  跳过 prune({reason},防误删)")

        cnt = db.execute(select(func.count()).select_from(MeituanCoupon)).scalar()
        with _STATS_LOCK:
            req, r402, err = _STATS["req"], _STATS["r402"], _STATS["err"]
        fail_tail = (": " + ",".join(fails[:10])) if fails else ""
        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: {done}/{n_city} 城, 入库 {total_up} 条, "
              f"表总计 {cnt} 行, 请求 {req}(402 {r402} / 放弃 {err}), "
              f"失败城 {len(fails)}{fail_tail}, 用时 {time.time() - t0:.0f}s")
    finally:
        db.close()
        _release_lock()


def _select_cities(limit: int, city_ids: str) -> list[dict]:
    cities = all_cities()
    if city_ids.strip():
        want = {c.strip() for c in city_ids.split(",") if c.strip()}
        return [c for c in cities if c["city_id"] in want]
    if limit and limit > 0:
        return cities[:limit]
    return cities


def main() -> None:
    ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库(全国 359 地级市)")
    ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
    ap.add_argument("--loop", action="store_true", help="循环跑")
    ap.add_argument("--interval", type=int, default=10800,
                    help="循环间隔秒(默认 10800=3h,对齐每 3h 全量一轮)")
    ap.add_argument("--prune-hours", type=int, default=24,
                    help="清理超过 N 小时未再出现的陈旧券(默认 24;0=不清理)")
    ap.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY,
                    help=f"并发城市数(默认 {DEFAULT_CONCURRENCY};实测 15 并发 402 占 3%% 可退避消化)")
    ap.add_argument("--stagger", type=float, default=STARTUP_STAGGER,
                    help=f"首批城市启动错峰间隔秒(默认 {STARTUP_STAGGER};削平瞬时峰值压低 402)")
    ap.add_argument("--limit", type=int, default=0, help="只抓前 N 个城市(测试用,0=全部)")
    ap.add_argument("--city-ids", default="", help="只抓这些 cityId(逗号分隔,测试用,优先于 --limit)")
    args = ap.parse_args()

    cities = _select_cities(args.limit, args.city_ids)
    print(f"城市数: {len(cities)} / 并发: {args.concurrency} / 错峰: {args.stagger}s / 间隔: {args.interval}s")

    if args.loop:
        print(f"循环模式: 每 {args.interval}s 一轮 (Ctrl-C 退出)")
        while True:
            try:
                run_once(args.prune_hours, args.concurrency, args.stagger, cities)
            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, args.concurrency, args.stagger, cities)


if __name__ == "__main__":
    main()
