chenshuobo
3fa20dbbf2
智能推荐 / 销量最高两 tab 的离线库此前只有北京(ETL 写死 cityId),改为遍历 美团官方城市字典 359 个地级市全量抓取。实测一个地级市 cityId 已覆盖其下辖县级市 (徐州→邳州/新沂/睢宁等),按地级市抓即可,无需区县层级。 - 城市字典:tools/gen_meituan_cities.py 从美团 Excel 生成随仓库 JSON (app/integrations/data/meituan_cities.json,359 城),app/integrations/cities.py 读取; - ETL:city_id 参数化 + 城市级并发(默认 12)+ worker 启动错峰削平瞬时峰值 + 主线程逐城串行入库(Session 不跨线程); - 配速实测:单城全量 ~110s/~2300 条;15 并发抓完一轮 ~50-60min,402 占 3% 退避全消化; 每 3h 一轮全量(--interval 10800),窗口充裕; - prune 双护栏:本轮 0 入库 或 失败城占比 >5% 时跳过,防上游故障/大面积限流误删全表; - 仅写入侧;读取侧(rec/top-sales 按城过滤)待后续(依赖用户定位→cityId 映射,字典无经纬度)。
447 lines
18 KiB
Python
447 lines
18 KiB
Python
"""美团 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 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 输出中文/¥
|
||
try:
|
||
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
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
|
||
|
||
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(避免大面积抓取失败误删库)
|
||
|
||
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 _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}"
|
||
return {
|
||
"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),
|
||
"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 _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]:
|
||
"""按 (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
|
||
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)
|
||
).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()
|