"""美团 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()