fix(meituan-etl): 常驻可观测性——压制美团 logger traceback + stdout 行缓冲

本地灰度(20 城 / 每 10min 一轮)暴露两个只在常驻/cron 下出现的问题:
- 美团调用偶发 SSL EOF(本机走代理高并发)/ code=5 被 meituan._call 的 logger.exception
  打完整 traceback,单轮刷数十 KB 日志;ETL 已用 _STATS 统计放弃页数,故把
  shagua.meituan logger 压到 CRITICAL(线上直连少见,统计仍可见放弃数);
- stdout 块缓冲导致 cron/后台 log 攒到 ~4KB 才落盘,常驻几乎看不到进度;
  reconfigure(line_buffering=True) 每行即时 flush。

app/integrations/cities.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+"""美团城市字典(地级市)读取。
+
+数据源:app/integrations/data/meituan_cities.json,由 tools/gen_meituan_cities.py 从
+美团官方「城市字典」Excel 生成(359 个地级市:city_id / 城市名 / 省份)。
+实测一个地级市 cityId 已覆盖其下辖县级市供给(徐州 → 邳州 / 新沂 / 睢宁 …),故无区县层级。
+
+- ETL(scripts/pull_meituan_coupons.py)遍历 all_cities() 全量抓取入库。
+- 读取侧将来「按城市过滤」时,也从这里取 city_id ↔ 城市名映射。
+"""
+from __future__ import annotations
+
+import functools
+import json
+from pathlib import Path
+
+_JSON = Path(__file__).resolve().parent / "data" / "meituan_cities.json"
+
+
+@functools.lru_cache(maxsize=1)
+def all_cities() -> list[dict]:
+    """全部城市 [{city_id, name, province}, ...](359 个地级市)。"""
+    return json.loads(_JSON.read_text(encoding="utf-8"))
+
+
+@functools.lru_cache(maxsize=1)
+def _by_id() -> dict[str, dict]:
+    return {c["city_id"]: c for c in all_cities()}
+
+
+def city_name(city_id: str) -> str | None:
+    """city_id → 城市名(查不到返回 None)。"""
+    c = _by_id().get(city_id)
+    return c["name"] if c else None

scripts/pull_meituan_coupons.py

@@ -1,45 +1,74 @@
-"""美团 CPS 券定时抓取入库(北京试点)。
+"""美团 CPS 券定时抓取入库(全国 359 个地级市)。
 
-把 3 路券抓进 meituan_coupon 表,供「销量/佣金排序」从库里捞、本地排序,不再实时打美团:
+把每个城市的 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 每轮刷新。带文件锁,防止
-上一轮没跑完下一轮又起(本地 5~10min、跨进程 cron 都安全)。
+上一轮没跑完下一轮又起。
 
 用法:
-    # 单轮(打通验证 / 给 cron 用,线上每 1h 一次)
+    # 单轮全量(给 cron 用,线上每 3h 一次)
     python -m scripts.pull_meituan_coupons --once
 
-    # 本地循环(默认每 10min 一轮)
+    # 常驻循环(每 3h 一轮)
-    python -m scripts.pull_meituan_coupons --loop --interval 600
+    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 输出中文/¥
+# Windows 控制台按 UTF-8 输出中文/¥;line_buffering=True 让 print 每行即时 flush——
+# 否则 stdout 重定向到 cron/后台日志文件时是块缓冲,要攒到 ~4KB 或进程退出才落盘,
+# 常驻(--loop)时几乎看不到每轮进度。
 try:
-    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
+    sys.stdout.reconfigure(encoding="utf-8", line_buffering=True)  # type: ignore[attr-defined]
 except Exception:  # noqa: BLE001
     pass
 
-from sqlalchemy import delete, func, select
+from sqlalchemy import delete, func, select  # noqa: E402
-from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.postgresql import insert as pg_insert  # noqa: E402
 
-from app.db.session import SessionLocal
+from app.db.session import SessionLocal, engine  # noqa: E402
-from app.integrations.meituan import MeituanCpsError, _call
+from app.integrations.cities import all_cities  # noqa: E402
-from app.models.meituan_coupon import MeituanCoupon
+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)
 
-CITY_BEIJING = "WKV2HMXUEK634WP64CUCUQGM64"
 QUERY_PATH = "/cps_open/common/api/v1/query_coupon"
 PAGE_SIZE = 20
 MAX_PAGES = 80          # 单路安全上限(搜索 ~52 页、供给 ~70 页)
@@ -49,7 +78,12 @@ RETRY = 7
 # 无 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 视为陈旧(进程异常退出残留),自动接管
+# 多城全量一轮 ~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(避免大面积抓取失败误删库)
 
 SOURCES = [
     {"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
@@ -58,33 +92,44 @@ SOURCES = [
      "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/频繁退避重试;其它错误打印并放弃本页。"""
+    """打美团,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
-            print(f"   [warn] meituan: {msg[:80]}")
+            _bump("err")
             return None
-        except Exception as e:  # noqa: BLE001
+        except Exception:  # noqa: BLE001
-            print(f"   [warn] {type(e).__name__}: {str(e)[:60]}")
             time.sleep(2.0 * (a + 1))
+    _bump("err")
     return None
 
 
-def _pull_search(platform: int, keyword: str) -> list[dict]:
+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_BEIJING, "pageSize": PAGE_SIZE}
+        body = {"platform": platform, "searchText": keyword, "cityId": city_id, "pageSize": PAGE_SIZE}
         if sid:
             body["searchId"] = sid
         else:
@@ -102,14 +147,14 @@ def _pull_search(platform: int, keyword: str) -> list[dict]:
     return rows
 
 
-def _pull_supply(platform: int, biz_lines: list[int]) -> list[dict]:
+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_BEIJING,
+            "cityId": city_id,
             "sortField": 2,  # 供给查询 sortField 必填;我们入库后本地再排,这里给个默认
             "pageSize": PAGE_SIZE,
         }
@@ -157,7 +202,7 @@ def _to_cents(yuan) -> int | None:
         return None
 
 
-def _parse_item(item: dict, source: dict) -> dict | None:
+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 {}
@@ -190,7 +235,7 @@ def _parse_item(item: dict, source: dict) -> dict | None:
         "source": source["code"],
         "platform": source["platform"],
         "biz_line": item.get("bizLine") or cpd.get("bizLine"),
-        "city_id": CITY_BEIJING,
+        "city_id": city_id,
         "product_view_sign": str(sign)[:128],
         "sku_view_id": cpd.get("skuViewId"),
         "name": (name[:256] or None),
@@ -210,6 +255,27 @@ def _parse_item(item: dict, source: dict) -> dict | None:
     }
 
 
+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)
+    return parsed
+
+
 # ───────────────────────── 入库(upsert) ─────────────────────────
 
 def _upsert(db, rows: list[dict], now: datetime) -> tuple[int, int]:
@@ -272,30 +338,53 @@ def _release_lock() -> None:
 
 # ───────────────────────── 主流程 ─────────────────────────
 
-def run_once(prune_hours: int = 24) -> None:
+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 = 0
+        total_up = 0
-        for src in SOURCES:
+        done = 0
-            ts = time.time()
+        fails: list[str] = []
-            if src["kind"] == "search":
+        # 城市级并发抓取(worker 只抓取+解析),主线程逐城串行入库(Session 不跨线程)。
-                items = _pull_search(src["platform"], src["keyword"])
+        with ThreadPoolExecutor(max_workers=concurrency) as pool:
-            else:
+            futs = {
-                items = _pull_supply(src["platform"], src["biz_lines"])
+                pool.submit(_pull_one_city, city, i, concurrency, stagger): city
-            parsed = [p for p in (_parse_item(it, src) for it in items) if p]
+                for i, city in enumerate(cities)
-            up, dup = _upsert(db, parsed, now)
+            }
-            total += up
+            for fut in as_completed(futs):
-            print(f"  {src['label']:18} 抓{len(items):5} 解析{len(parsed):5} "
+                city = futs[fut]
-                  f"入库{up:5} (本源去重{dup:3}) {time.time() - ts:4.0f}s")
+                try:
-        # 清理长期未再出现的陈旧券(美团 sign 轮换 / 券下架后的残留),默认 24h 宽限。
+                    parsed = fut.result()
-        # 护栏:仅在本轮确有入库(total>0)时才清理。美团整体故障时本轮可能 0 入库
+                except Exception as e:  # noqa: BLE001
-        # (脚本不抛异常、只是抓回空),若仍照常 prune,连续故障会按 last_seen 把全表删空。
+                    fails.append(city["name"])
-        if prune_hours and prune_hours > 0 and total > 0:
+                    print(f"  [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
+                    continue
+                up, _dup = _upsert(db, parsed, now)
+                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)
@@ -303,36 +392,62 @@ def run_once(prune_hours: int = 24) -> None:
             db.commit()
             if pruned:
                 print(f"  清理陈旧券(>{prune_hours}h 未再出现): {pruned} 条")
-        elif prune_hours and prune_hours > 0 and total == 0:
+        elif prune_hours and prune_hours > 0:
-            print("  本轮 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()
-        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: 入库 {total} 条, 表总计 {cnt} 行, "
+        with _STATS_LOCK:
-              f"用时 {time.time() - t0:.0f}s")
+            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 券定时抓取入库")
+    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=600, help="循环间隔秒(默认 600=10min)")
+    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)
+                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)
+        run_once(args.prune_hours, args.concurrency, args.stagger, cities)
 
 
 if __name__ == "__main__":

tools/gen_meituan_cities.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""从美团「城市字典」Excel 生成随仓库的 JSON(app/integrations/data/meituan_cities.json)。
+
+本地一次性工具:美团每年更新城市字典时,把新 Excel 放进来重跑即可。
+Excel 不入库、不进仓库(二进制、需 openpyxl);生成的 JSON 随仓库提交,部署到服务器后
+ETL(scripts/pull_meituan_coupons.py)直接读它遍历全部城市。
+
+字典口径:359 个地级市,经实测一个地级市 cityId 已覆盖其下辖县级市(如徐州→邳州/新沂),
+故无需区县层级。
+
+用法:
+    python tools/gen_meituan_cities.py ["城市字典xxx.xlsx" 路径]
+    (不传则用默认 e:\\codes\\城市字典2025 (1).xlsx)
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+import openpyxl
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+DEFAULT_EXCEL = r"e:\codes\城市字典2025 (1).xlsx"
+OUT = os.path.join(ROOT, "app", "integrations", "data", "meituan_cities.json")
+
+
+def main() -> None:
+    excel = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_EXCEL
+    wb = openpyxl.load_workbook(excel, data_only=True)
+    ws = wb.worksheets[0]
+    rows = list(ws.iter_rows(values_only=True))
+    header = rows[0]
+
+    cities = []
+    seen = set()
+    for r in rows[1:]:
+        cid = str(r[0]).strip() if r[0] else ""
+        name = str(r[1]).strip() if len(r) > 1 and r[1] else ""
+        prov = str(r[2]).strip() if len(r) > 2 and r[2] else ""
+        if not cid or cid in seen:
+            continue
+        seen.add(cid)
+        cities.append({"city_id": cid, "name": name, "province": prov})
+
+    os.makedirs(os.path.dirname(OUT), exist_ok=True)
+    with open(OUT, "w", encoding="utf-8") as f:
+        json.dump(cities, f, ensure_ascii=False, indent=1)
+
+    print(f"表头: {header}")
+    print(f"写出 {len(cities)} 城 → {OUT}")
+    print("样例:", cities[:3])
+
+
+if __name__ == "__main__":
+    main()