功能：新手引导视频 + 美团券首页分页索引 (#167)

Co-authored-by: guke <guke@wonderable.ai>
Co-authored-by: 左辰勇 <exinglang@gmail.com>
Reviewed-on: #167
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>

app/admin/main.py

@@ -30,6 +30,7 @@ from app.admin.routers.analytics_health import router as analytics_health_router
 from app.admin.routers.event_logs import router as event_logs_router
 from app.admin.routers.feedback import router as feedback_router
 from app.admin.routers.feedback_qr import router as feedback_qr_router
+from app.admin.routers.guide_video import router as guide_video_router
 from app.admin.routers.huawei_review import router as huawei_review_router
 from app.admin.routers.onboarding import router as onboarding_router
 from app.admin.routers.ops_marquee_seed import router as ops_marquee_seed_router
@@ -40,6 +41,7 @@ from app.admin.routers.wallet import router as wallet_router
 from app.admin.routers.withdraw import router as withdraw_router
 from app.core.config import settings
 from app.core.logging import setup_logging
+from app.integrations import meituan as mt_meituan
 
 setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.admin")
@@ -53,6 +55,8 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     yield
+    # CPS 后台页会打美团(routers/cps.py),那条共享 client 若被建过要在这里关掉连接池
+    mt_meituan.close_client()
     logger.info("admin app shutting down")
 
 
@@ -101,6 +105,7 @@ admin_app.include_router(feedback_router)
 admin_app.include_router(event_logs_router)
 admin_app.include_router(analytics_health_router)
 admin_app.include_router(feedback_qr_router)
+admin_app.include_router(guide_video_router)
 admin_app.include_router(admins_router)
 admin_app.include_router(roles_router)
 admin_app.include_router(audit_router)

app/admin/routers/guide_video.py

@@ -0,0 +1,101 @@
+"""admin 新手引导视频配置:读 / 改开关次数金币 / 上传视频 / 删视频(带审计)。
+
+整份配置存通用 app_config 表(见 app/repositories/guide_video.py),App 领券等候浮层
+每次展示前调 POST /api/v1/guide-video/start 同步。权限:operator 可改(运营维护),
+super 恒可;读为只读(任意已登录 admin)。
+
+⚠️ 视频上限 100MB(settings.GUIDE_VIDEO_MAX_BYTES),已在 admin nginx 为本接口单独放宽
+client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.com.conf。
+"""
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
+
+from app.admin.audit import write_audit
+from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
+from app.admin.schemas.guide_video import GuideVideoConfigOut, GuideVideoConfigUpdate
+from app.core import media
+from app.models.admin import AdminUser
+from app.repositories import guide_video
+
+router = APIRouter(
+    prefix="/admin/api/guide-video",
+    tags=["admin-guide-video"],
+    dependencies=[Depends(get_current_admin)],
+)
+
+
+def _out(db: AdminDb) -> GuideVideoConfigOut:
+    """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
+    return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
+
+
+@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
+def get_config(db: AdminDb) -> GuideVideoConfigOut:
+    return _out(db)
+
+
+@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
+def update_config(
+    body: GuideVideoConfigUpdate,
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("operator"))],
+    db: AdminDb,
+) -> GuideVideoConfigOut:
+    before, after = guide_video.update_config(
+        db,
+        enabled=body.enabled,
+        max_plays=body.max_plays,
+        reward_coin=body.reward_coin,
+        admin_id=admin.id,
+        commit=False,
+    )
+    write_audit(
+        db, admin, action="guide_video.update", target_type="guide_video", target_id=None,
+        detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
+    return _out(db)
+
+
+@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
+async def upload_video(
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("operator"))],
+    db: AdminDb,
+    file: UploadFile = File(...),
+) -> GuideVideoConfigOut:
+    data = await file.read()
+    try:
+        url = media.save_guide_video(data)
+    except media.MediaError as e:
+        raise HTTPException(status_code=400, detail=str(e)) from e
+    before, after = guide_video.set_video(db, url, admin_id=admin.id, commit=False)
+    write_audit(
+        db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
+        detail={"before": before.get("video_url"), "after": url, "bytes": len(data)},
+        ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
+    # 提交成功后再删旧片,避免新片没落库就把旧片丢了
+    media.delete_guide_video(before.get("video_url"))
+    return _out(db)
+
+
+@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
+def delete_video(
+    request: Request,
+    admin: Annotated[AdminUser, Depends(require_role("operator"))],
+    db: AdminDb,
+) -> GuideVideoConfigOut:
+    """移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
+    before, after = guide_video.set_video(db, None, admin_id=admin.id, commit=False)
+    write_audit(
+        db, admin, action="guide_video.delete_video", target_type="guide_video", target_id=None,
+        detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
+    )
+    db.commit()
+    media.delete_guide_video(before.get("video_url"))
+    return _out(db)

app/admin/schemas/guide_video.py

@@ -0,0 +1,25 @@
+"""admin 新手引导视频配置 schemas(开关 / 视频地址 / 前几次 / 每次金币)。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
+
+
+class GuideVideoConfigOut(BaseModel):
+    enabled: bool
+    video_url: str | None = None   # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
+    max_plays: int
+    reward_coin: int
+    updated_at: str | None = None
+    # 只读统计,后台展示用:已有多少次播放、其中已发币多少次。
+    total_plays: int = 0
+    granted_plays: int = 0
+
+
+class GuideVideoConfigUpdate(BaseModel):
+    """部分更新:只改传入(非 None)字段。视频文件走 /video 上传接口。"""
+
+    enabled: bool | None = None
+    max_plays: int | None = Field(default=None, ge=0, le=MAX_PLAYS_LIMIT)
+    reward_coin: int | None = Field(default=None, ge=0, le=REWARD_COIN_LIMIT)

app/api/v1/guide_video.py

@@ -0,0 +1,64 @@
+"""新手引导视频(领券等候浮层前 N 次替代广告)。
+
+路由前缀 `/api/v1/guide-video`(均需 Bearer):
+  POST /start    这次浮层放引导视频还是放广告?命中则**当场计次**并下发 play_token
+  POST /reward   播完 / 中途关闭都调,按 play_token 幂等发固定金币
+
+发币额度以**服务端配置**为准(运营后台可改),客户端只报"播完/关闭",报不了金额,
+所以被破解也刷不到超额金币;次数上限由 guide_video_play 行数(按账号)硬卡。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends
+
+from app.api.deps import CurrentUser, DbSession
+from app.core.ratelimit import rate_limit
+from app.repositories import guide_video as crud_guide
+from app.schemas.guide_video import (
+    GuideVideoRewardIn,
+    GuideVideoRewardOut,
+    GuideVideoStartIn,
+    GuideVideoStartOut,
+)
+
+logger = logging.getLogger("shagua.guide_video")
+
+router = APIRouter(prefix="/api/v1/guide-video", tags=["guide-video"])
+
+
+@router.post(
+    "/start",
+    response_model=GuideVideoStartOut,
+    summary="领券浮层是否放新手引导视频(命中即计次)",
+    dependencies=[Depends(rate_limit(60, 60, "guide-video-start"))],
+)
+def start(payload: GuideVideoStartIn, user: CurrentUser, db: DbSession) -> GuideVideoStartOut:
+    """开播即计数:返回 should_play=True 时服务端已写下这一次,客户端必须真的播。
+
+    没配视频 / 开关关 / 次数用完 → should_play=False,客户端照旧走广告链路(行为不变)。
+    """
+    result = crud_guide.start_play(db, user.id, scene=payload.scene or "coupon")
+    logger.info(
+        "guide video start user_id=%d scene=%s should_play=%s seq=%d remaining=%d",
+        user.id, payload.scene, result["should_play"], result["seq"], result["remaining"],
+    )
+    return GuideVideoStartOut(**result)
+
+
+@router.post(
+    "/reward",
+    response_model=GuideVideoRewardOut,
+    summary="引导视频发金币(播完/中途关闭都发,play_token 幂等)",
+    dependencies=[Depends(rate_limit(60, 60, "guide-video-reward"))],
+)
+def reward(payload: GuideVideoRewardIn, user: CurrentUser, db: DbSession) -> GuideVideoRewardOut:
+    result = crud_guide.grant_play(
+        db, user.id, play_token=payload.play_token, completed=payload.completed
+    )
+    logger.info(
+        "guide video reward user_id=%d token=%s completed=%s granted=%s coin=%d",
+        user.id, payload.play_token[:12], payload.completed, result["granted"], result["coin"],
+    )
+    return GuideVideoRewardOut(**result)

app/api/v1/meituan.py

@@ -5,11 +5,12 @@
 from __future__ import annotations
 
 import logging
-from concurrent.futures import ThreadPoolExecutor, as_completed
+from concurrent.futures import ThreadPoolExecutor
+from typing import TYPE_CHECKING, Any
 
 from fastapi import APIRouter, Depends, HTTPException
 from sqlalchemy import nullslast, select
-from sqlalchemy.orm import Session, aliased
+from sqlalchemy.orm import Session
 
 from app.core.config import settings
 from app.db.session import get_db
@@ -25,8 +26,14 @@ from app.schemas.meituan import (
     ReferralLinkResponse,
     TopSalesRequest,
 )
+from app.utils import mt_search_cursor
 from app.utils.meituan_city import get_meituan_city
 
+if TYPE_CHECKING:  # 仅供类型标注(本模块已开 from __future__ import annotations)
+    from collections.abc import Callable
+
+    from sqlalchemy import ColumnElement
+
 logger = logging.getLogger("shagua.meituan")
 
 
@@ -109,6 +116,86 @@ def _commission_pct(card: CouponCard) -> float:
         return 0.0
 
 
+# ────────────── 离线库分页(智能推荐 / 销量最高 共用) ──────────────
+# 去重+排序阶段**只投影这几列**:够 DISTINCT ON 分组、够排序、够回表定位,且全是定长小字段。
+# ⚠️ 关键性能点:`raw` 是整条美团原始返回(JSONB,每行数 KB)。原实现用 select(MeituanCoupon)
+# 做子查询,等于把整城几千行连 raw 一起塞进两次排序(DISTINCT ON 一次 + 分页一次),
+# 体量轻松超过 work_mem → Postgres 落盘做外部归并排序,而且**每翻一页都要重来一遍**。
+# 拆成「先在小列上排出本页 id,再按 id 回表取 raw」后,排序数据量降到原来的百分之几,
+# JSONB 只解析当前页 ~20 行。
+_DEDUP_COLS = (
+    MeituanCoupon.id,
+    MeituanCoupon.dedup_key,
+    MeituanCoupon.sale_volume_num,
+    MeituanCoupon.commission_percent,
+)
+
+
+def _paged_dedup_ids(
+    db: Session,
+    *,
+    conds: list[ColumnElement[bool]],
+    dedup_order: list[ColumnElement],
+    page_order: Callable[[Any], list[ColumnElement]],
+    page: int,
+    page_size: int,
+) -> tuple[list[int], bool]:
+    """DISTINCT ON(dedup_key) 跨源去重 → 整体排序 → 分页,返回 (本页 id 列表, 是否还有下一页)。
+
+    - `dedup_order`:同一个 dedup_key 的多条里留哪条(如销量最高/佣金最高)。
+    - `page_order`:接收去重子查询的列集合(`sub.c`),返回去重后的整体排序。
+    多取 1 条用于判断 has_next。
+    """
+    deduped = (
+        select(*_DEDUP_COLS)
+        .where(*conds)
+        .distinct(MeituanCoupon.dedup_key)
+        .order_by(MeituanCoupon.dedup_key, *dedup_order)
+        .subquery()
+    )
+    ids = db.execute(
+        select(deduped.c.id)
+        .order_by(*page_order(deduped.c))
+        .offset((page - 1) * page_size)
+        .limit(page_size + 1)
+    ).scalars().all()
+    return list(ids[:page_size]), len(ids) > page_size
+
+
+def _load_raws(db: Session, ids: list[int]) -> list[dict]:
+    """按给定 id 顺序取 raw(只回表本页 ~20 行)。缺行(被 ETL 清掉)静默跳过。"""
+    if not ids:
+        return []
+    raw_by_id = {
+        row_id: raw
+        for row_id, raw in db.execute(
+            select(MeituanCoupon.id, MeituanCoupon.raw).where(MeituanCoupon.id.in_(ids))
+        ).all()
+    }
+    return [raw_by_id[i] for i in ids if i in raw_by_id]
+
+
+def _cards_from_raws(raws: list[dict], *, hide_distance: bool) -> list[CouponCard]:
+    """raw → CouponCard;解析失败的单条跳过,不整页失败。
+
+    hide_distance:离线库里的距离是相对「城市默认点」算的,对用户无意义且误导 —— 智能推荐 /
+    销量最高两个 tab 一律置空,前端「距离 店名」那行只剩店名、自动顶到最左。
+    """
+    cards: list[CouponCard] = []
+    for raw in raws:
+        try:
+            card = CouponCard.from_raw(raw or {})
+        except Exception:  # noqa: BLE001
+            continue
+        if not card.product_view_sign:
+            continue
+        if hide_distance:
+            card.distance_text = None
+            card.distance_meters = None
+        cards.append(card)
+    return cards
+
+
 @router.post("/feed", response_model=FeedResponse, summary="混合feed(外卖+到店交叉);tab=rec智能推荐/distance距离最近")
 def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
     lon, lat = req.longitude, req.latitude
@@ -133,17 +220,21 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
             return [], True
 
     # 距离最近:搜索召回(外卖搜"外卖" + 到店搜"美食",都 sortField=6 离我最近)一页页拉。
-    # 搜索翻页必须用 searchId(pageNo 翻不动),所以每个 feed 页顺序翻到第 N 页;两路并行、page 1 最快。
-    # 无状态、不改 APP(传页码即可);按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
+    # 搜索翻页必须用 searchId(pageNo 翻不动),而接口是无状态的(客户端只传页码)—— 原实现因此
+    # 每次都从第 1 页顺序重放到第 N 页,取第 N 页要向美团发 N 次请求,越往下滑越慢。
+    # 现在把沿途 searchId 记进 [mt_search_cursor],稳态下每翻一页恒定 1 次请求;两路仍并行。
+    # 按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
     if tab == "distance":
         lon_i, lat_i = int(lon * 1_000_000), int(lat * 1_000_000)
 
-        def _search_page_n(platform: int, biz_line: int | None, keyword: str, n: int) -> tuple[list[dict], bool, bool]:
-            """顺序翻到第 n 页(搜索须 searchId 续页),返回(第 n 页 items, 是否还有下一页, 是否调用失败)。"""
-            sid: str | None = None
+        def _replay(
+            platform: int, biz_line: int | None, keyword: str,
+            key: mt_search_cursor.RouteKey, start: int, sid: str | None, n: int,
+        ) -> tuple[list[dict], bool, bool]:
+            """从第 start 页(用 sid 取)顺序翻到第 n 页。start==1 时 sid 应为 None(走 pageNo=1)。"""
             data: list[dict] = []
             has_next = False
-            for pg in range(1, n + 1):
+            for pg in range(start, n + 1):
                 body: dict = {
                     "platform": platform, "searchText": keyword, "sortField": 6,
                     "longitude": lon_i, "latitude": lat_i, "pageSize": 20,
@@ -161,10 +252,27 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
                 data = r.get("data") or []
                 sid = r.get("searchId")
                 has_next = bool(r.get("hasNext")) and bool(data)
+                # 记下「下一页要用哪个 searchId」;没有下一页就别记,免得存进死游标。
+                if sid and has_next:
+                    mt_search_cursor.remember(key, pg + 1, sid)
                 if not data or (not has_next and pg < n):
                     return [], False, False   # 没那么多页了(非错误)
             return data, has_next, False
 
+        def _search_page_n(platform: int, biz_line: int | None, keyword: str, n: int) -> tuple[list[dict], bool, bool]:
+            """取第 n 页,返回(第 n 页 items, 是否还有下一页, 是否调用失败)。
+
+            优先用缓存游标一发直达;缓存未命中/过期才从最近的已知页往后重放,并把沿途游标补进缓存。
+            """
+            key = mt_search_cursor.route_key(lat, lon, platform, keyword)
+            start, sid = mt_search_cursor.lookup(key, n)
+            data, has_next, failed = _replay(platform, biz_line, keyword, key, start, sid, n)
+            # 用缓存游标却打不通,多半是上游 searchId 过期:作废整条路线,回到第 1 页重放一次。
+            if failed and start > 1:
+                mt_search_cursor.drop(key)
+                data, has_next, failed = _replay(platform, biz_line, keyword, key, 1, None, n)
+            return data, has_next, failed
+
         with ThreadPoolExecutor(max_workers=2) as pool:
             f_wm = pool.submit(_search_page_n, 1, None, "外卖", req.page)
             f_dd = pool.submit(_search_page_n, 2, 1, "美食", req.page)
@@ -194,39 +302,25 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
             return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
         PAGE = 20
         try:
-            base = select(MeituanCoupon).where(
-                MeituanCoupon.commission_percent >= 3.0,
-                MeituanCoupon.city_id == city_id,
-            )
-            deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
-                MeituanCoupon.dedup_key,
-                MeituanCoupon.commission_percent.desc(),
-            ).subquery()
-            m = aliased(MeituanCoupon, deduped)
-            start = (req.page - 1) * PAGE
-            rows = db.execute(
-                select(m)
+            ids, has_next = _paged_dedup_ids(
+                db,
+                conds=[
+                    MeituanCoupon.commission_percent >= 3.0,
+                    MeituanCoupon.city_id == city_id,
+                ],
+                # 同一去重键留佣金最高那条
+                dedup_order=[MeituanCoupon.commission_percent.desc()],
                 # 销量高的优先(无销量档排后),同档佣金高优先,id 兜底稳定分页
-                .order_by(nullslast(m.sale_volume_num.desc()), m.commission_percent.desc(), m.id)
-                .offset(start)
-                .limit(PAGE + 1)
-            ).scalars().all()
+                page_order=lambda c: [
+                    nullslast(c.sale_volume_num.desc()), c.commission_percent.desc(), c.id,
+                ],
+                page=req.page, page_size=PAGE,
+            )
+            raws = _load_raws(db, ids)
         except Exception:  # noqa: BLE001
             logger.exception("[feed] rec 库查询失败,降级返空")
             return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
-        has_next = len(rows) > PAGE
-        cards: list[CouponCard] = []
-        for row in rows[:PAGE]:
-            try:
-                card = CouponCard.from_raw(row.raw or {})
-            except Exception:  # noqa: BLE001
-                continue
-            if card.product_view_sign:
-                # 智能推荐不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
-                # 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
-                card.distance_text = None
-                card.distance_meters = None
-                cards.append(card)
+        cards = _cards_from_raws(raws, hide_distance=True)
         if not cards and req.page == 1:
             # 命中城市却 0 券:该城确无 ≥3% 券,或 ETL 灌的 city_id 与 city_dict 口径不一致。
             logger.info("[feed] rec city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)
@@ -282,51 +376,36 @@ def top_sales(req: TopSalesRequest, db: Session = Depends(get_db)) -> CouponList
     if not city_id:
         return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
 
-    # 去重 + 排序 + 分页全在 SQL 做,每页只取并解析当前页 ~20 条。
-    # (之前实现每翻一页都全表拉取 + 全量 from_raw 解析,翻页慢 → 客户端滑动卡顿/翻不动。)
+    # 去重 + 排序 + 分页全在 SQL 做,每页只回表并解析当前页 ~20 条(见 _paged_dedup_ids 的性能说明)。
     # 库为空(prod 刚部署 / ETL 未跑完)时返空 + status=empty,不崩;库查询异常降级 degraded。
+    conds = [
+        MeituanCoupon.sale_volume_num.isnot(None),
+        MeituanCoupon.city_id == city_id,
+    ]
+    if req.platform is not None:
+        conds.append(MeituanCoupon.platform == req.platform)
     try:
-        # 1) DISTINCT ON (dedup_key):每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
-        base = select(MeituanCoupon).where(
-            MeituanCoupon.sale_volume_num.isnot(None),
-            MeituanCoupon.city_id == city_id,
-        )
-        if req.platform is not None:
-            base = base.where(MeituanCoupon.platform == req.platform)
-        deduped = base.distinct(MeituanCoupon.dedup_key).order_by(
-            MeituanCoupon.dedup_key,
-            MeituanCoupon.sale_volume_num.desc(),
-            MeituanCoupon.commission_percent.desc(),
-        ).subquery()
-
-        # 2) 对去重结果按销量降序分页;多取 1 条判断 has_next,只对本页做 from_raw
-        m = aliased(MeituanCoupon, deduped)
-        start = (req.page - 1) * req.page_size
-        rows = db.execute(
-            select(m)
+        ids, has_next = _paged_dedup_ids(
+            db,
+            conds=conds,
+            # 每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
+            dedup_order=[
+                MeituanCoupon.sale_volume_num.desc(),
+                MeituanCoupon.commission_percent.desc(),
+            ],
             # 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
-            .order_by(m.sale_volume_num.desc(), m.commission_percent.desc(), m.id)
-            .offset(start)
-            .limit(req.page_size + 1)
-        ).scalars().all()
+            page_order=lambda c: [
+                c.sale_volume_num.desc(), c.commission_percent.desc(), c.id,
+            ],
+            page=req.page, page_size=req.page_size,
+        )
+        raws = _load_raws(db, ids)
     except Exception:  # noqa: BLE001
         logger.exception("[top-sales] 库查询失败,降级返空")
         return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
 
-    has_next = len(rows) > req.page_size
-    cards: list[CouponCard] = []
-    for row in rows[:req.page_size]:
-        try:
-            card = CouponCard.from_raw(row.raw or {})
-        except Exception:  # noqa: BLE001
-            continue
-        if card.product_view_sign:
-            # 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导)。
-            # 置空后前端"距离 店名"那行只剩店名、自动顶到最左(店名移到原距离的位置)。
-            # 逻辑与推荐流保持一致
-            card.distance_text = None
-            card.distance_meters = None
-            cards.append(card)
+    # 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导),与推荐流口径一致。
+    cards = _cards_from_raws(raws, hide_distance=True)
     if not cards and req.page == 1:
         # 命中城市却 0 券:可能该城确无券,也可能 ETL 灌的 city_id 与 city_dict 口径不一致(静默降级的隐患)。
         logger.info("[top-sales] city_id=%s 命中 0 券(该城确无券?或 ETL/city_dict 的 city_id 口径不一致)", city_id)

app/core/config.py

@@ -384,6 +384,9 @@ class Settings(BaseSettings):
     MEDIA_ROOT: str = "./data/media"
     MEDIA_URL_PREFIX: str = "/media"
     AVATAR_MAX_BYTES: int = 5 * 1024 * 1024  # 头像最大 5MB
+    # 运营后台上传的新手引导视频上限。视频比图片大一个量级,单独一档;
+    # ⚠️ 改大时同步放宽网关 client_max_body_size(实测 QA 4MiB / prod 32MiB),否则 nginx 先挡下。
+    GUIDE_VIDEO_MAX_BYTES: int = 100 * 1024 * 1024  # 引导视频最大 100MB
 
     # ===== 邀请好友 =====
     # 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。

app/core/media.py

@@ -77,6 +77,35 @@ def save_feedback_qr(data: bytes) -> str:
     return _save_named("feedback_qr", "qr", data)
 
 
+def _sniff_video_ext(data: bytes) -> str | None:
+    """按魔数判定视频类型,返回扩展名;非支持类型返回 None。
+
+    只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
+    都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
+    """
+    if len(data) >= 12 and data[4:8] == b"ftyp":
+        return ".mp4"
+    return None
+
+
+def save_guide_video(data: bytes) -> str:
+    """保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。
+
+    与图片分开一套校验:体积上限走 [settings.GUIDE_VIDEO_MAX_BYTES],类型只认 MP4。
+    """
+    if not data:
+        raise MediaError("空文件")
+    limit = settings.GUIDE_VIDEO_MAX_BYTES
+    if len(data) > limit:
+        raise MediaError(f"视频过大(上限 {limit // (1024 * 1024)}MB)")
+    if _sniff_video_ext(data) is None:
+        raise MediaError("仅支持 MP4 视频(H.264 编码)")
+
+    fname = f"guide_{secrets.token_hex(8)}.mp4"
+    (_media_dir("guide_video") / fname).write_bytes(data)
+    return f"{settings.MEDIA_URL_PREFIX}/guide_video/{fname}"
+
+
 def save_cps_image(admin_id: int, data: bytes) -> str:
     """保存 CPS 活动落地页图,返回相对 URL(`/media/cps/<file>`)。admin_id 入文件名便于追溯。"""
     return _save_image("cps", admin_id, data)
@@ -116,3 +145,8 @@ def delete_avatar(url: str | None) -> None:
 def delete_feedback_qr(url: str | None) -> None:
     """删除本服务托管的旧反馈页二维码文件;外部 URL 或空值不处理。"""
     _delete_managed("feedback_qr", url)
+
+
+def delete_guide_video(url: str | None) -> None:
+    """删除本服务托管的旧新手引导视频文件;外部 URL 或空值不处理。"""
+    _delete_managed("guide_video", url)

app/integrations/meituan.py

@@ -11,6 +11,7 @@ import hashlib
 import hmac
 import json
 import logging
+import threading
 import time
 from typing import Any
 
@@ -25,6 +26,44 @@ class MeituanCpsError(Exception):
     """美团 CPS 接口调用失败。"""
 
 
+# ────────────────────── 共享 httpx.Client(连接池 + keep-alive) ──────────────────────
+# 原实现每次 _call 都 `with httpx.Client(...)` 新建再关掉,等于每发一次请求就:
+#   ① 重建一套 SSL 上下文(加载 certifi CA,实测几百 ms) ② 重做一次 TLS 握手 ③ 用完即弃连接。
+# 首页 feed 一次翻页要向美团发好几次请求(外卖/到店两路,「距离最近」还要续页),这份固定开销被成倍放大,
+# 是「滑到底部加载很慢」的一大块。改成进程内单例:握手一次、后续走 keep-alive 复用。
+# httpx.Client 本身线程安全,可被 feed 的 ThreadPoolExecutor 两条抓取线程共用。
+_client: httpx.Client | None = None
+_client_lock = threading.Lock()
+
+
+def get_client() -> httpx.Client:
+    """取美团 CPS 共享 client(幂等,懒建)。lifespan 启动时预热,把建 SSL 的成本摊到启动。"""
+    global _client
+    if _client is None:
+        with _client_lock:
+            if _client is None:
+                # 走 MT_CPS_PROXY(本机开发直连会 SSL EOF,必须走代理;线上留空=直连)。
+                # trust_env=False:不读进程环境的 HTTP_PROXY,只认配置,避免被错误/失效代理误导。
+                _client = httpx.Client(
+                    proxy=settings.MT_CPS_PROXY or None,
+                    trust_env=False,
+                    timeout=settings.MT_CPS_TIMEOUT_SEC,
+                    # 池子留足:feed 每个请求会起 2 条抓取线程,并发用户多时别在池上排队
+                    # (排满会等到 MT_CPS_TIMEOUT_SEC 抛 PoolTimeout,表现成"又慢又降级")。
+                    limits=httpx.Limits(max_keepalive_connections=16, max_connections=32),
+                )
+    return _client
+
+
+def close_client() -> None:
+    """lifespan 关停时调,优雅关连接池。"""
+    global _client
+    with _client_lock:
+        if _client is not None:
+            _client.close()
+            _client = None
+
+
 def _content_md5(body: bytes) -> str:
     return base64.b64encode(hashlib.md5(body).digest()).decode()
 
@@ -62,12 +101,8 @@ 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:
-        with httpx.Client(proxy=proxy, trust_env=False, timeout=settings.MT_CPS_TIMEOUT_SEC) as client:
-            resp = client.post(url, content=body, headers=headers)
+        resp = get_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/main.py

@@ -29,6 +29,7 @@ from app.api.v1.coupon import router as coupon_router
 from app.api.v1.cps_redirect import router as cps_redirect_router
 from app.api.v1.device import router as device_router
 from app.api.v1.feedback import router as feedback_router
+from app.api.v1.guide_video import router as guide_video_router
 from app.api.v1.invite import router as invite_router
 from app.api.v1.meituan import router as meituan_router
 from app.api.v1.notifications import router as notifications_router
@@ -70,6 +71,7 @@ from app.core.withdraw_reconcile_worker import (
     start_withdraw_reconcile_worker,
     stop_withdraw_reconcile_worker,
 )
+from app.integrations import meituan as mt_meituan
 
 setup_logging(debug=settings.APP_DEBUG)
 logger = logging.getLogger("shagua.main")
@@ -86,6 +88,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         settings.DATABASE_URL.split("://", 1)[0],
     )
     get_pricebot_client()  # 预热透传 client:把建 SSL 上下文的一次性成本付在启动,首个领券请求即热
+    if settings.mt_cps_configured:
+        # 同理预热美团 CPS client(TLS 上下文 + 连接池建一次,后续 keep-alive 复用)
+        mt_meituan.get_client()
     try:
         # 预热离线地理库:首次加载 ~2.5M 行 CSV + 建 KDTree,摊到启动、不砸首个按城市过滤的请求
         from app.utils import geo
@@ -108,6 +113,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
         await aclose_pricebot_client()
+        mt_meituan.close_client()
         logger.info("shutting down")
 
 
@@ -154,6 +160,8 @@ app.include_router(signin_router)
 app.include_router(tasks_router)
 app.include_router(savings_router)
 app.include_router(ad_router)
+# 新手引导视频(领券等候浮层前 N 次替代广告;运营后台传片,见 repositories/guide_video.py)
+app.include_router(guide_video_router)
 app.include_router(order_router)
 app.include_router(report_router)
 # 消息通知中心(PRD;数据落库 notification 表,见 repositories/notification.py)

app/models/__init__.py

@@ -28,6 +28,7 @@ from app.models.coupon_state import (  # noqa: F401
     CouponSession,
 )
 from app.models.feedback import Feedback  # noqa: F401
+from app.models.guide_video import GuideVideoPlay  # noqa: F401
 from app.models.inactivity import (  # noqa: F401
     InactivityNotificationLog,
     InactivityResetLog,

app/models/guide_video.py

@@ -0,0 +1,73 @@
+"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
+
+产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
+改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。
+
+口径:
+- **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
+- **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
+  写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill
+  App 也算用掉一次(产品选定口径,防反复进出刷金币)。
+- **发币幂等**靠 play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次
+  改到行(另一次 rowcount=0),所以只发一次币。光有 play_token 唯一键挡不住 —— 发币走的是
+  UPDATE,不 INSERT,撞不到任何唯一键。
+- **次数上限**靠 (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)。
+
+与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class GuideVideoPlay(Base):
+    """一次引导视频播放一行。开播时建(status='playing'),发币后置 'granted'。"""
+
+    __tablename__ = "guide_video_play"
+    __table_args__ = (
+        # 客户端幂等键:同一次播放重复上报奖励只发一次。
+        UniqueConstraint("play_token", name="uq_guide_video_play_token"),
+        # 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插),
+        # N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
+        # 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
+        # start_play 捕获 IntegrityError 降级成"这次不放视频"。
+        # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
+        # 要整表重建),autogenerate 才不会每次报一条假 diff。
+        Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("user.id"), index=True, nullable=False
+    )
+    # 服务端生成下发给客户端的幂等键(uuid hex)。
+    play_token: Mapped[str] = mapped_column(String(64), nullable=False)
+    # 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
+    scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
+    # 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
+    # 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
+    seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    # 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
+    video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    # 实发金币;未发时 0。
+    coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # playing(已开播未发币) / granted(已发币)。
+    status: Mapped[str] = mapped_column(String(16), nullable=False, default="playing")
+    # 客户端上报时是否播完(true=自然播完 / false=中途关闭)。仅留痕:两者都发币。
+    completed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+    granted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<GuideVideoPlay user={self.user_id} seq={self.seq} "
+            f"{self.status} coin={self.coin}>"
+        )

app/models/meituan_coupon.py

@@ -23,7 +23,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import JSON, DateTime, Float, Integer, String, UniqueConstraint, func
+from sqlalchemy import JSON, DateTime, Float, Index, Integer, String, UniqueConstraint, func
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -95,3 +95,23 @@ class MeituanCoupon(Base):
             f"<MeituanCoupon id={self.id} source={self.source} "
             f"name={self.name!r} sale={self.sale_volume} comm={self.commission_percent}>"
         )
+
+
+# 首页「销量最高 / 智能推荐」两个 tab 的分页索引(见 api/v1/meituan.py 的 _paged_dedup_ids)。
+# 两条 tab 都是 `WHERE city_id=? [+过滤] ORDER BY dedup_key, <排序键> DESC` 的 DISTINCT ON 去重,
+# 列顺序对齐后 Postgres 可以顺着索引流式去重,免掉整城数据的排序 —— 否则每翻一页都要把该城
+# 全部券重排一遍(下滑到底越来越慢的根因之一)。
+# 定义放在类外:__table_args__ 里拿不到还没建好的类属性,写不了 .desc()。
+Index(
+    "ix_meituan_coupon_city_dedup_sales",
+    MeituanCoupon.city_id,
+    MeituanCoupon.dedup_key,
+    MeituanCoupon.sale_volume_num.desc(),
+    MeituanCoupon.commission_percent.desc(),
+)
+Index(
+    "ix_meituan_coupon_city_dedup_comm",
+    MeituanCoupon.city_id,
+    MeituanCoupon.dedup_key,
+    MeituanCoupon.commission_percent.desc(),
+)

app/repositories/guide_video.py

@@ -0,0 +1,290 @@
+"""新手引导视频:运营配置读写 + 播放计次 + 发币。
+
+**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config 表
+(key=coupon_guide_video),写法完全对齐 feedback_qr —— 不进 CONFIG_DEFS,所以不会污染
+系统配置页的通用列表,由本模块独占维护。
+
+**计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
+`/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
+已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)。
+COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个。
+
+**发币**幂等键是 play_token,落地方式是 `status='playing' → 'granted'` 的**条件更新**:
+同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)。
+中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
+
+两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释。
+"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import func, select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.core import rewards
+from app.models.app_config import AppConfig
+from app.models.guide_video import GuideVideoPlay
+from app.repositories import wallet as crud_wallet
+
+_KEY = "coupon_guide_video"
+
+#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
+BIZ_TYPE = "guide_video"
+
+# 默认值 = 「运营还没配」时的行为:video_url 为空 → 一律不下发引导视频,浮层维持现状(放广告)。
+# 所以本功能上线后**不配视频就等于没上线**,不会影响存量用户。
+_DEFAULTS: dict[str, Any] = {
+    "enabled": True,
+    "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
+    "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
+    "reward_coin": 120,  # 每次固定金币
+}
+
+_FIELDS = tuple(_DEFAULTS.keys())
+
+# 后台可配范围的护栏:防手滑把次数/金币填成天文数字(配置直接决定发币)。
+MAX_PLAYS_LIMIT = 50
+REWARD_COIN_LIMIT = 10_000
+
+
+# ===== 配置 =====
+
+
+def _merge(raw: Any) -> dict[str, Any]:
+    """DB 里(可能不全的)dict 叠加到默认上,得到完整配置(4 个字段,无 updated_at)。"""
+    out = dict(_DEFAULTS)
+    if isinstance(raw, dict):
+        for k in _FIELDS:
+            v = raw.get(k)
+            if v is not None:
+                out[k] = v
+    return out
+
+
+def get_config(db: Session) -> dict[str, Any]:
+    """完整配置 + updated_at(admin 读 / 业务读共用)。"""
+    row = db.get(AppConfig, _KEY)
+    cfg = _merge(row.value if row is not None else None)
+    cfg["updated_at"] = row.updated_at.isoformat() if row is not None and row.updated_at else None
+    return cfg
+
+
+def _write(db: Session, value: dict[str, Any], *, admin_id: int, commit: bool) -> dict[str, Any]:
+    """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
+    row = db.get(AppConfig, _KEY)
+    if row is None:
+        row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
+        db.add(row)
+    else:
+        row.value = value  # 整体重新赋值,SQLAlchemy 才侦测得到变更
+        row.updated_by_admin_id = admin_id
+    if commit:
+        db.commit()
+        db.refresh(row)
+    else:
+        db.flush()
+    out = _merge(row.value)
+    out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
+    return out
+
+
+def update_config(
+    db: Session,
+    *,
+    enabled: bool | None = None,
+    max_plays: int | None = None,
+    reward_coin: int | None = None,
+    admin_id: int,
+    commit: bool = True,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
+    row = db.get(AppConfig, _KEY)
+    before = _merge(row.value if row is not None else None)
+    new_value = {k: before[k] for k in _FIELDS}
+    if enabled is not None:
+        new_value["enabled"] = enabled
+    if max_plays is not None:
+        new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
+    if reward_coin is not None:
+        new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    return before, after
+
+
+def set_video(
+    db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    """设置/清空引导视频地址。返回 (before, after);before['video_url'] 供调用方删旧文件。"""
+    row = db.get(AppConfig, _KEY)
+    before = _merge(row.value if row is not None else None)
+    new_value = {k: before[k] for k in _FIELDS}
+    new_value["video_url"] = video_url
+    after = _write(db, new_value, admin_id=admin_id, commit=commit)
+    return before, after
+
+
+# ===== 播放计次 =====
+
+
+def used_plays(db: Session, user_id: int) -> int:
+    """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
+    return int(
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.user_id == user_id
+            )
+        ).scalar_one()
+    )
+
+
+def play_stats(db: Session) -> dict[str, int]:
+    """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
+    total = int(
+        db.execute(select(func.count()).select_from(GuideVideoPlay)).scalar_one()
+    )
+    granted = int(
+        db.execute(
+            select(func.count()).select_from(GuideVideoPlay).where(
+                GuideVideoPlay.status == "granted"
+            )
+        ).scalar_one()
+    )
+    return {"total_plays": total, "granted_plays": granted}
+
+
+def start_play(
+    db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
+) -> dict[str, Any]:
+    """决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
+
+    返回 dict:
+      should_play  是否放引导视频(False → 客户端照旧放广告)
+      video_url    相对地址(/media/...);客户端自行拼 BASE_URL
+      play_token   发币幂等键(should_play=False 时为空串)
+      reward_coin  播完/中途关闭都发的固定金币
+      seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
+    """
+    cfg = get_config(db)
+    video_url = (cfg.get("video_url") or "").strip()
+    max_plays = int(cfg.get("max_plays") or 0)
+    reward_coin = int(cfg.get("reward_coin") or 0)
+    used = used_plays(db, user_id)
+
+    def _miss(used_now: int) -> dict[str, Any]:
+        return {
+            "should_play": False,
+            "video_url": None,
+            "play_token": "",
+            "reward_coin": reward_coin,
+            "seq": used_now,
+            "remaining": max(0, max_plays - used_now),
+        }
+
+    if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
+        return _miss(used)
+
+    seq = used + 1
+    play = GuideVideoPlay(
+        user_id=user_id,
+        play_token=uuid.uuid4().hex,
+        scene=scene,
+        seq=seq,
+        video_url=video_url,
+        coin=0,
+        status="playing",
+        completed=0,
+        started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
+    )
+    db.add(play)
+    # 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
+    # (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
+    # 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
+    try:
+        if commit:
+            db.commit()
+        else:
+            db.flush()
+    except IntegrityError:
+        db.rollback()
+        return _miss(used_plays(db, user_id))
+    return {
+        "should_play": True,
+        "video_url": video_url,
+        "play_token": play.play_token,
+        "reward_coin": reward_coin,
+        "seq": seq,
+        "remaining": max(0, max_plays - seq),
+    }
+
+
+def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
+    """按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
+    return db.execute(
+        select(GuideVideoPlay).where(
+            GuideVideoPlay.play_token == token,
+            GuideVideoPlay.user_id == user_id,
+        )
+    ).scalar_one_or_none()
+
+
+def grant_play(
+    db: Session, user_id: int, *, play_token: str, completed: bool
+) -> dict[str, Any]:
+    """按 play_token 发这次引导视频的金币(幂等)。播完 / 中途关闭都发。
+
+    返回 {granted, coin, status}:granted=True 表示**本次调用真的入账了**;
+    重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
+    """
+    token = (play_token or "").strip()
+    # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
+    coin = int(get_config(db).get("reward_coin") or 0)
+
+    # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
+    # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
+    # 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
+    # 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
+    # (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
+    #
+    # 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
+    # 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
+    # (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
+    won = db.execute(
+        update(GuideVideoPlay)
+        .where(
+            GuideVideoPlay.play_token == token,
+            GuideVideoPlay.user_id == user_id,
+            GuideVideoPlay.status == "playing",
+        )
+        .values(
+            status="granted",
+            coin=coin,
+            completed=1 if completed else 0,
+            granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
+        )
+        .execution_options(synchronize_session=False)
+    ).rowcount
+
+    if not won:
+        # 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
+        # (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
+        db.rollback()
+        play = _find_play(db, user_id, token)
+        if play is None:
+            return {"granted": False, "coin": 0, "status": "not_found"}
+        return {"granted": False, "coin": play.coin, "status": "already_granted"}
+
+    if coin > 0:
+        crud_wallet.grant_coins(
+            db,
+            user_id,
+            coin,
+            biz_type=BIZ_TYPE,
+            ref_id=token,
+            remark="新手引导视频奖励",
+        )
+    db.commit()
+    return {"granted": True, "coin": coin, "status": "granted"}

app/schemas/guide_video.py

@@ -0,0 +1,36 @@
+"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class GuideVideoStartIn(BaseModel):
+    """开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
+
+    scene: str = Field(default="coupon", max_length=16)
+
+
+class GuideVideoStartOut(BaseModel):
+    """should_play=False 时客户端照旧走广告链路,其余字段无意义。"""
+
+    should_play: bool
+    video_url: str | None = None   # 相对地址 /media/...;客户端自行拼 BASE_URL
+    play_token: str = ""           # 发奖幂等键
+    reward_coin: int = 0           # 播完/中途关闭都发的固定金币
+    seq: int = 0                   # 本账号第几次
+    remaining: int = 0             # 发完这次还剩几次
+
+
+class GuideVideoRewardIn(BaseModel):
+    """播完或中途关闭都调这个;completed 只做留痕,两者都发币。"""
+
+    play_token: str = Field(min_length=1, max_length=64)
+    completed: bool = False
+
+
+class GuideVideoRewardOut(BaseModel):
+    """granted=True 表示本次调用真的入账(重复上报为 False,coin 是已发金额)。"""
+
+    granted: bool
+    coin: int
+    status: str

app/utils/mt_search_cursor.py

@@ -0,0 +1,97 @@
+"""美团搜索翻页游标(searchId)缓存 —— 「距离最近」tab 翻页提速。
+
+**问题**:美团 query_coupon 的搜索结果只能靠 searchId 续页(pageNo 翻不动),而我们的 HTTP 接口
+是无状态的、客户端只传页码。原实现因此每次都从第 1 页顺序重放到第 N 页 —— 取第 N 页要向美团发
+N 次请求,用户越往下滑越慢(第 5 页 ≈ 第 1 页的 5 倍耗时,且每次调用都可能撞 402 限流)。
+
+**做法**:把「翻到第 n 页所需的 searchId」按 (量化坐标, platform, 关键词) 记下来,
+下次请求第 n 页直接一发命中;未命中才从缓存里最深的那一页往后重放,并把沿途游标补进缓存。
+稳态下每翻一页恒定 1 次请求。
+
+**坐标量化**到 2 位小数(~1km),与 `utils/meituan_city` 同口径:原始 GPS 每次抖动到小数点后 5~6 位,
+不量化几乎不命中缓存;而同一路搜索的排序原点相差 1km 以内,对券列表的实际顺序无感知差别。
+
+**TTL** 取 10 分钟:searchId 属于上游会话态,放太久会翻到过期游标(调用失败 → 调用方回退重放)。
+缓存是**进程内**的,多 worker 各存一份、不共享,这没问题:命中率只影响快慢,不影响正确性。
+"""
+from __future__ import annotations
+
+import threading
+import time
+
+# 缓存键:(量化纬度, 量化经度, platform, 搜索词)
+RouteKey = tuple[float, float, int, str]
+
+_TTL_SEC = 10 * 60
+_MAX_ROUTES = 512          # 最多缓存多少条搜索路线(每条 = 一个坐标×平台×关键词)
+_MAX_PAGES_PER_ROUTE = 60  # 单条路线最多记多少页游标,防某个坐标被无限翻页撑爆
+
+# {route_key: {page_no: (search_id, 写入时刻)}};page_no 表示「用这个 searchId 能取到第几页」。
+# 第 1 页不需要 searchId(直接 pageNo=1),故永远不入表。
+_cursors: dict[RouteKey, dict[int, tuple[str, float]]] = {}
+_lock = threading.Lock()
+
+
+def route_key(latitude: float, longitude: float, platform: int, keyword: str) -> RouteKey:
+    """构造缓存键(坐标量化到 ~1km)。"""
+    return (round(latitude, 2), round(longitude, 2), platform, keyword)
+
+
+def lookup(key: RouteKey, page: int) -> tuple[int, str | None]:
+    """找到能最省调用地拿到第 `page` 页的起点。
+
+    返回 (起始页, 该页要用的 search_id):
+      - (page, "xxx")  缓存命中 → 调用方一发直达第 page 页
+      - (m, "xxx")     命中更浅的第 m 页(1 < m < page)→ 从第 m 页重放到第 page 页
+      - (1, None)      全未命中 → 从第 1 页(pageNo=1)重放
+    """
+    if page <= 1:
+        return 1, None
+    now = time.time()
+    with _lock:
+        pages = _cursors.get(key)
+        if not pages:
+            return 1, None
+        for p in range(page, 1, -1):
+            hit = pages.get(p)
+            if hit is None:
+                continue
+            search_id, wrote_at = hit
+            if now - wrote_at > _TTL_SEC:
+                pages.pop(p, None)   # 过期即清,继续往浅了找
+                continue
+            return p, search_id
+    return 1, None
+
+
+def remember(key: RouteKey, page: int, search_id: str) -> None:
+    """记下「用 search_id 可取到第 page 页」。page<=1 无意义(第 1 页不用游标)。"""
+    if page <= 1 or not search_id:
+        return
+    now = time.time()
+    with _lock:
+        pages = _cursors.get(key)
+        if pages is None:
+            if len(_cursors) >= _MAX_ROUTES:
+                _evict_oldest_route_locked()
+            pages = _cursors[key] = {}
+        pages[page] = (search_id, now)
+        if len(pages) > _MAX_PAGES_PER_ROUTE:
+            # 越浅的页越容易被重新走到,优先丢最深的那些(它们下次多半也过期了)
+            for p in sorted(pages, reverse=True)[: len(pages) - _MAX_PAGES_PER_ROUTE]:
+                pages.pop(p, None)
+
+
+def drop(key: RouteKey) -> None:
+    """整条路线作废。用在「拿缓存游标去请求却失败」时(多半是上游 searchId 过期)。"""
+    with _lock:
+        _cursors.pop(key, None)
+
+
+def _evict_oldest_route_locked() -> None:
+    """淘汰最久没更新过的一条路线(调用方须已持锁)。"""
+    oldest_key = min(
+        _cursors,
+        key=lambda k: max((w for _, w in _cursors[k].values()), default=0.0),
+    )
+    _cursors.pop(oldest_key, None)