新增美团 CPS 券定时抓取入库(北京试点),作为"销量/佣金排序"的本地数据源 (#18)

背景:美团搜索/供给对销量排序支持差(实测乱序)、且有 402 限流和单次召回上限,不适合
每次实时打接口排序。改为定时把券抓进本地 meituan_coupon 表,查询时从库里捞、自己按
销量/佣金排。

本次内容:
- app/models/meituan_coupon.py:新表模型。字段尽量全(售价/原价/销量档/佣金率/佣金额/
  门店/距离/品牌/图 等),raw 列存整条原始返回避免漏字段;(source, product_view_sign) 唯一。
- alembic/versions/meituan_coupon_table.py:建表迁移(挂在 withdraw_review_ad_watch 之后)。
- scripts/pull_meituan_coupons.py:ETL,抓 3 路并按 (source, product_view_sign) upsert:
  1) 外卖·搜"外卖"(platform=1)翻到尽头
  2) 外卖·搜"美食"(platform=1)翻到尽头
  3) 到店·多业务线供给(到餐+到综+酒店+门票)翻到尽头
  支持 --once(单轮,给 cron;线上每 1h)/ --loop --interval(本地循环,默认 10min)
  + 运行锁防重叠 + --prune-hours 清理陈旧券(默认 24h)。
- app/models/__init__.py:注册新模型。

实测:北京一轮约 2.5 分钟、入库约 2700 条;销量/佣金可直接从库里排序。

给接手同事:
- 仅北京(city_id 固定);v2 再加 per商圈 union 突破全城单次召回上限。
- 销量仅粗档位且约 51% 的券才有,查询需 WHERE sale_volume_num IS NOT NULL
  (PG 的 DESC 默认把 NULL 排最前);佣金、价格 100% 有值。
- productViewSign/skuViewId 跨渠道会变,不能当全局 id;跨源去重用 dedup_key=md5(品牌|名|价)。
- 待做:查询接口(从库捞 + dedup_key 去重 + 按销量/佣金排,返回前端)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

---------

Co-authored-by: chenshuobo <1119780489@qq.com>
Reviewed-on: #18

.env.example

@@ -40,6 +40,9 @@ MT_CPS_APP_KEY=
 MT_CPS_APP_SECRET=
 # 默认渠道追踪标识(sid),用于区分不同 app 的 CPS 数据
 MT_CPS_DEFAULT_SID=sgbjia
+# 美团调用走的代理。⚠️ 本机/内网开发直连美团会 SSL EOF,必须填本地代理(如 http://127.0.0.1:7897);
+# 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。
+MT_CPS_PROXY=
 
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。

alembic/versions/11a1d08c6f55_merge_meituan_coupon_与_ops_price_.py

@@ -0,0 +1,26 @@
+"""merge meituan_coupon 与 ops/price_observation 迁移头
+
+Revision ID: 11a1d08c6f55
+Revises: meituan_coupon_table, opsrename01
+Create Date: 2026-06-08 00:06:01.633796
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '11a1d08c6f55'
+down_revision: Union[str, Sequence[str], None] = ('meituan_coupon_table', 'opsrename01')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

alembic/versions/meituan_coupon_table.py

@@ -0,0 +1,62 @@
+"""meituan_coupon table(美团 CPS 券本地缓存,供销量/佣金排序从库里捞、本地排序)
+
+Revision ID: meituan_coupon_table
+Revises: withdraw_review_ad_watch
+Create Date: 2026-06-06 12:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'meituan_coupon_table'
+down_revision: Union[str, Sequence[str], None] = 'withdraw_review_ad_watch'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'meituan_coupon',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('source', sa.String(length=16), nullable=False),
+        sa.Column('platform', sa.Integer(), nullable=False),
+        sa.Column('biz_line', sa.Integer(), nullable=True),
+        sa.Column('city_id', sa.String(length=32), nullable=False),
+        sa.Column('product_view_sign', sa.String(length=128), nullable=False),
+        sa.Column('sku_view_id', sa.String(length=128), nullable=True),
+        sa.Column('name', sa.String(length=256), nullable=True),
+        sa.Column('brand_name', sa.String(length=128), nullable=True),
+        sa.Column('sell_price_cents', sa.Integer(), nullable=True),
+        sa.Column('original_price_cents', sa.Integer(), nullable=True),
+        sa.Column('head_url', sa.String(length=512), nullable=True),
+        sa.Column('sale_volume', sa.String(length=32), nullable=True),
+        sa.Column('sale_volume_num', sa.Integer(), nullable=True),
+        sa.Column('commission_percent', sa.Float(), nullable=True),
+        sa.Column('commission_amount_cents', sa.Integer(), nullable=True),
+        sa.Column('poi_name', sa.String(length=128), nullable=True),
+        sa.Column('available_poi_num', sa.Integer(), nullable=True),
+        sa.Column('delivery_distance_m', sa.Float(), nullable=True),
+        sa.Column('dedup_key', sa.String(length=64), nullable=False),
+        sa.Column('raw', sa.JSON().with_variant(postgresql.JSONB(), 'postgresql'), nullable=False),
+        sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('source', 'product_view_sign', name='uq_meituan_coupon_source_sign'),
+    )
+    op.create_index('ix_meituan_coupon_source', 'meituan_coupon', ['source'])
+    op.create_index('ix_meituan_coupon_city_id', 'meituan_coupon', ['city_id'])
+    op.create_index('ix_meituan_coupon_brand_name', 'meituan_coupon', ['brand_name'])
+    op.create_index('ix_meituan_coupon_sale_volume_num', 'meituan_coupon', ['sale_volume_num'])
+    op.create_index('ix_meituan_coupon_commission_percent', 'meituan_coupon', ['commission_percent'])
+    op.create_index('ix_meituan_coupon_dedup_key', 'meituan_coupon', ['dedup_key'])
+    op.create_index('ix_meituan_coupon_last_seen', 'meituan_coupon', ['last_seen'])
+
+
+def downgrade() -> None:
+    op.drop_table('meituan_coupon')

app/api/v1/meituan.py

@@ -7,10 +7,14 @@ from __future__ import annotations
 import logging
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from app.core.config import settings
+from app.db.session import get_db
 from app.integrations.meituan import MeituanCpsError, get_referral_link, query_coupon
+from app.models.meituan_coupon import MeituanCoupon
 from app.schemas.meituan import (
     CouponCard,
     CouponListRequest,
@@ -19,6 +23,7 @@ from app.schemas.meituan import (
     FeedResponse,
     ReferralLinkRequest,
     ReferralLinkResponse,
+    TopSalesRequest,
 )
 
 logger = logging.getLogger("shagua.meituan")
@@ -79,13 +84,21 @@ def _interleave(waimai: list[dict], daodian: list[dict]) -> list[CouponCard]:
     return items
 
 
-@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉, 无限流)")
+def _commission_pct(card: CouponCard) -> float:
+    """'1.4%' → 1.4;解析失败按 0(会被智能推荐过滤掉)。"""
+    try:
+        return float(card.commission_rate.rstrip("%"))
+    except (ValueError, AttributeError):
+        return 0.0
+
+
+@router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
 def feed(req: FeedRequest) -> FeedResponse:
     if not settings.mt_cps_configured:
         return FeedResponse(items=[], has_next=False, page=req.page)
-    page_idx = req.page - 1
     lon, lat = req.longitude, req.latitude
-    logger.info("[feed] page=%s lon=%.6f lat=%.6f", req.page, lon, lat)
+    tab = (req.tab or "").strip()
+    logger.info("[feed] tab=%s page=%s lon=%.6f lat=%.6f", tab or "(default)", req.page, lon, lat)
 
     def _fetch_topic(platform: int, biz_line: int | None, topic: int) -> list[dict]:
         try:
@@ -97,6 +110,28 @@ def feed(req: FeedRequest) -> FeedResponse:
         except MeituanCpsError:
             return []
 
+    # 距离最近:拉齐全部榜单轮次,合并去重,后端在【完整池子】上全局按距离由近及远排,一次性返回。
+    # (距离排序必须在完整池上做、不能逐页排——这正是之前放前端不合理的根因。)
+    if tab == "distance":
+        with ThreadPoolExecutor(max_workers=len(_TOPIC_ROUNDS) * 2) as pool:
+            futs = []
+            for wm_topic, dd_topic in _TOPIC_ROUNDS:
+                futs.append(pool.submit(_fetch_topic, 1, None, wm_topic))
+                futs.append(pool.submit(_fetch_topic, 2, 1, dd_topic))
+            raws = [f.result() for f in futs]
+        seen: set[str] = set()
+        cards: list[CouponCard] = []
+        for raw_list in raws:
+            for it in raw_list:
+                card = CouponCard.from_raw(it)
+                if card.product_view_sign and card.product_view_sign not in seen:
+                    seen.add(card.product_view_sign)
+                    cards.append(card)
+        cards.sort(key=lambda c: c.distance_meters if c.distance_meters is not None else float("inf"))
+        return FeedResponse(items=cards, has_next=False, page=1)
+
+    # 智能推荐(rec,默认):沿用逐轮分页的混合 feed,后端过滤掉佣金率 < 3%。
+    page_idx = req.page - 1
     if page_idx >= len(_TOPIC_ROUNDS):
         return FeedResponse(items=[], has_next=False, page=req.page)
 
@@ -107,6 +142,8 @@ def feed(req: FeedRequest) -> FeedResponse:
         waimai, daodian = f_wm.result(), f_dd.result()
 
     items = _interleave(waimai, daodian)
+    if tab == "rec":
+        items = [c for c in items if _commission_pct(c) >= 3.0]
     has_next = page_idx + 1 < len(_TOPIC_ROUNDS)
     return FeedResponse(items=items, has_next=has_next, page=req.page)
 
@@ -130,3 +167,37 @@ def referral_link(req: ReferralLinkRequest) -> ReferralLinkResponse:
     link_map = raw.get("referralLinkMap") or {}
     link = raw.get("data") or link_map.get("1") or link_map.get("3") or next(iter(link_map.values()), "")
     return ReferralLinkResponse(link=link, link_map=link_map)
+
+
+@router.post("/top-sales", response_model=CouponListResponse,
+             summary="销量最高(从离线库 meituan_coupon 按销量降序 + 跨源去重,不实时打美团)")
+def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponListResponse:
+    # 只取有销量档的(美团很多券没销量,排不了);按销量降序、同销量再按佣金降序
+    stmt = select(MeituanCoupon).where(MeituanCoupon.sale_volume_num.isnot(None))
+    if req.platform is not None:
+        stmt = stmt.where(MeituanCoupon.platform == req.platform)
+    stmt = stmt.order_by(
+        MeituanCoupon.sale_volume_num.desc(),
+        MeituanCoupon.commission_percent.desc(),
+    )
+    rows = db.execute(stmt).scalars().all()
+
+    # 跨源去重(dedup_key = 品牌|名|价):按销量降序遍历,每个 dedup_key 只留第一条(=销量最高那条)。
+    # 用 raw(整条原始返回)重建 CouponCard,字段与实时接口完全一致,前端无需改渲染。
+    seen: set[str] = set()
+    cards: list[CouponCard] = []
+    for row in rows:
+        if row.dedup_key in seen:
+            continue
+        seen.add(row.dedup_key)
+        try:
+            card = CouponCard.from_raw(row.raw or {})
+        except Exception:  # noqa: BLE001
+            continue
+        if card.product_view_sign:
+            cards.append(card)
+
+    start = (req.page - 1) * req.page_size
+    page_items = cards[start:start + req.page_size]
+    has_next = start + req.page_size < len(cards)
+    return CouponListResponse(items=page_items, has_next=has_next, search_id=None)

app/core/config.py

@@ -81,6 +81,9 @@ class Settings(BaseSettings):
     MT_CPS_HOST: str = "https://media.meituan.com"
     MT_CPS_TIMEOUT_SEC: int = 15
     MT_CPS_DEFAULT_SID: str = "sgbjia"
+    # 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
+    # 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
+    MT_CPS_PROXY: str = ""
 
     @property
     def mt_cps_configured(self) -> bool:

app/integrations/meituan.py

@@ -62,8 +62,12 @@ def _call(path: str, body_obj: dict[str, Any]) -> dict[str, Any]:
     }
 
     url = f"{settings.MT_CPS_HOST}{path}"
+    # 美团调用走 MT_CPS_PROXY(本机开发直连会 SSL EOF,必须走代理;线上留空=直连)。
+    # trust_env=False:不读进程环境的 HTTP_PROXY,只认配置,避免被错误/失效代理误导。
+    proxy = settings.MT_CPS_PROXY or None
     try:
-        resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
+        with httpx.Client(proxy=proxy, trust_env=False, timeout=settings.MT_CPS_TIMEOUT_SEC) as client:
+            resp = client.post(url, content=body, headers=headers)
     except httpx.HTTPError as e:
         logger.exception("[MT] http error calling %s", url)
         raise MeituanCpsError(f"meituan http error: {e}") from e

app/models/__init__.py

@@ -8,6 +8,7 @@ from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
+from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401
 from app.models.ops_stat_config import OpsStatConfig  # noqa: F401
 from app.models.price_observation import PriceObservation  # noqa: F401

app/models/meituan_coupon.py

@@ -0,0 +1,93 @@
+"""美团 CPS 券本地缓存表(meituan_coupon)。
+
+把美团联盟 CPS 的券定时抓进本地库,供「销量 / 佣金排序」等查询**直接从库里捞、本地排序**,
+不再每次实时打美团接口——美团搜索/供给对销量排序支持差(实测乱序)、且有 402 限流和召回上限。
+
+数据来源(source,北京试点):
+  - search_waimai : 到家/外卖, 搜索关键词「外卖」, 翻到尽头
+  - search_meishi : 到家/外卖, 搜索关键词「美食」, 翻到尽头
+  - store_supply  : 到店, 多业务线供给(到餐+到综+酒店+门票), 翻到尽头
+
+每 5min(本地测试)/ 1h(线上)全量抓一次,按 (source, product_view_sign) upsert 存最新态,
+last_seen 每轮刷新(可据此清理长期未再出现的陈旧券)。
+
+去重说明:`product_view_sign` / `sku_view_id` 都是「按召回渠道生成」的,跨渠道(搜索 vs 供给)
+会变,**不能当商品全局唯一 id**。因此:
+  - 存储:按 (source, product_view_sign) upsert(同源短周期内 sign 稳定)。
+  - 查询:用 `dedup_key = md5(brand|name|price)` 跨源去重,再按 `sale_volume_num` / `commission_percent` 排序。
+
+价格统一存「分」(cents),与 price_report / comparison_record 一致。`raw` 保留整条原始返回
+(字段越详细越好,避免后续漏字段还要重抓)。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, Float, Integer, String, UniqueConstraint, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class MeituanCoupon(Base):
+    __tablename__ = "meituan_coupon"
+    __table_args__ = (
+        UniqueConstraint("source", "product_view_sign", name="uq_meituan_coupon_source_sign"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # ===== 来源 / 分类 =====
+    # source: search_waimai | search_meishi | store_supply
+    source: Mapped[str] = mapped_column(String(16), index=True, nullable=False)
+    platform: Mapped[int] = mapped_column(Integer, nullable=False)              # 1 到家/外卖, 2 到店
+    biz_line: Mapped[int | None] = mapped_column(Integer, nullable=True)        # 到店: 1到餐 2到综 3酒店 4门票
+    city_id: Mapped[str] = mapped_column(String(32), index=True, nullable=False)
+
+    # ===== 召回 id(按渠道生成,跨渠道会变;product_view_sign 用于换推广链) =====
+    product_view_sign: Mapped[str] = mapped_column(String(128), nullable=False)
+    sku_view_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+
+    # ===== 商品本体 =====
+    name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    brand_name: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    sell_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    original_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    head_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+
+    # ===== 销量(美团只给粗档位:热销1w+;num=排序用的下界数值,如 1w+ → 10000) =====
+    sale_volume: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    sale_volume_num: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+
+    # ===== 佣金(percent 如 1.4 表示 1.4%;amount 为分) =====
+    commission_percent: Mapped[float | None] = mapped_column(Float, index=True, nullable=True)
+    commission_amount_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    # ===== 门店 / 距离 =====
+    poi_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    available_poi_num: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    delivery_distance_m: Mapped[float | None] = mapped_column(Float, nullable=True)
+
+    # ===== 跨源去重键 + 原始返回 =====
+    dedup_key: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    raw: Mapped[dict] = mapped_column(
+        JSON().with_variant(JSONB(), "postgresql"), nullable=False, default=dict
+    )
+
+    # ===== 抓取时间窗 =====
+    first_seen: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    last_seen: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<MeituanCoupon id={self.id} source={self.source} "
+            f"name={self.name!r} sale={self.sale_volume} comm={self.commission_percent}>"
+        )

app/schemas/meituan.py

@@ -134,6 +134,12 @@ class FeedRequest(BaseModel):
     latitude: float = Field(..., description="纬度")
     page: int = Field(1, ge=1)
     page_size: int = Field(20, ge=1, le=20)
+    # 筛选/排序口径,后端据此处理后返回(前端不再自己筛/排):
+    #  rec      = 智能推荐:榜单混合 feed 去掉佣金率 < 3%(分页)
+    #  distance = 距离最近:拉齐全部轮次后全局按距离由近及远(一次性返回, has_next=False)
+    #  留空/其它 = 原混合 feed 不筛(老客户端兼容,新 app 会显式传 tab)
+    # (销量最高 sales 走 /coupons 同城热销,不在本接口)
+    tab: str = Field("", description="rec 智能推荐 / distance 距离最近 / 空=不筛(兼容)")
 
 
 class FeedResponse(BaseModel):
@@ -142,6 +148,13 @@ class FeedResponse(BaseModel):
     page: int = 1
 
 
+class TopSalesRequest(BaseModel):
+    """销量最高 tab:从离线库 meituan_coupon 按销量降序取(不实时打美团)。"""
+    page: int = Field(1, ge=1)
+    page_size: int = Field(20, ge=1, le=50)
+    platform: int | None = Field(None, description="可选: 1只外卖 / 2只到店; 不填=全部(全城销量)")
+
+
 # ───────────────── 换链 请求 / 响应 ─────────────────
 
 class ReferralLinkRequest(BaseModel):

scripts/pull_meituan_coupons.py

@@ -0,0 +1,331 @@
+"""美团 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 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
+LOCK_FILE = "data/.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 宽限
+        if prune_hours and prune_hours > 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} 条")
+        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()