diff --git a/.env.example b/.env.example index 5e29c1e..20ef954 100644 --- a/.env.example +++ b/.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。 diff --git a/alembic/versions/11a1d08c6f55_merge_meituan_coupon_与_ops_price_.py b/alembic/versions/11a1d08c6f55_merge_meituan_coupon_与_ops_price_.py new file mode 100644 index 0000000..f445fda --- /dev/null +++ b/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 diff --git a/alembic/versions/meituan_coupon_table.py b/alembic/versions/meituan_coupon_table.py new file mode 100644 index 0000000..284f357 --- /dev/null +++ b/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') diff --git a/app/api/v1/meituan.py b/app/api/v1/meituan.py index 7e3e2bc..9d7510e 100644 --- a/app/api/v1/meituan.py +++ b/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) diff --git a/app/core/config.py b/app/core/config.py index 68e66f0..c6eff5d 100644 --- a/app/core/config.py +++ b/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: diff --git a/app/integrations/meituan.py b/app/integrations/meituan.py index 4bb7188..6d1b10a 100644 --- a/app/integrations/meituan.py +++ b/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 diff --git a/app/models/__init__.py b/app/models/__init__.py index ca2d6d1..2f95779 100644 --- a/app/models/__init__.py +++ b/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 diff --git a/app/models/meituan_coupon.py b/app/models/meituan_coupon.py new file mode 100644 index 0000000..df6cd10 --- /dev/null +++ b/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"" + ) diff --git a/app/schemas/meituan.py b/app/schemas/meituan.py index b8cf1ab..67cf36c 100644 --- a/app/schemas/meituan.py +++ b/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): diff --git a/scripts/pull_meituan_coupons.py b/scripts/pull_meituan_coupons.py new file mode 100644 index 0000000..7bd6837 --- /dev/null +++ b/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()