From a2270ee1b2f96dc2a6a99d5101bc09744277a5f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B7=A6=E8=BE=B0=E5=8B=87?= Date: Thu, 23 Jul 2026 22:50:44 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E6=96=B0=E6=89=8B?= =?UTF-8?q?=E5=BC=95=E5=AF=BC=E8=A7=86=E9=A2=91=20+=20=E7=BE=8E=E5=9B=A2?= =?UTF-8?q?=E5=88=B8=E9=A6=96=E9=A1=B5=E5=88=86=E9=A1=B5=E7=B4=A2=E5=BC=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新手引导视频:运营后台上传 MP4(上限 100MB,魔数校验只认 ISO BMFF), App 端在领券等候浮层前 N 次以引导视频替代广告。新增 guide_video 的 model/schema/repository/router(App 侧 + 后台侧)与播放记录表迁移。 美团券:首页「销量最高 / 智能推荐」两个 tab 改游标分页,配套两条 (city_id, dedup_key, 排序键 DESC) 复合索引,让 Postgres 顺着索引流式 去重,免掉每翻一页重排整城券的开销。美团 CPS client 在 lifespan 预热 并在关闭时释放连接池。 Co-Authored-By: Claude Opus 4.8 (1M context) --- alembic/versions/guide_video_play_table.py | 50 ++++ .../versions/meituan_coupon_feed_indexes.py | 52 ++++ app/admin/main.py | 5 + app/admin/routers/guide_video.py | 101 +++++++ app/admin/schemas/guide_video.py | 25 ++ app/api/v1/guide_video.py | 64 +++++ app/api/v1/meituan.py | 229 +++++++++++----- app/core/config.py | 3 + app/core/media.py | 34 +++ app/integrations/meituan.py | 45 ++- app/main.py | 8 + app/models/__init__.py | 1 + app/models/guide_video.py | 62 +++++ app/models/meituan_coupon.py | 22 +- app/repositories/guide_video.py | 257 ++++++++++++++++++ app/schemas/guide_video.py | 36 +++ app/utils/mt_search_cursor.py | 97 +++++++ docs/api/meituan/meituan-feed.md | 3 +- docs/api/meituan/meituan-top-sales.md | 1 + docs/database/meituan_coupon.md | 6 +- scripts/reset_guide_video.py | 216 +++++++++++++++ scripts/seed_meituan_coupon_mock.py | 222 +++++++++++++++ 22 files changed, 1456 insertions(+), 83 deletions(-) create mode 100644 alembic/versions/guide_video_play_table.py create mode 100644 alembic/versions/meituan_coupon_feed_indexes.py create mode 100644 app/admin/routers/guide_video.py create mode 100644 app/admin/schemas/guide_video.py create mode 100644 app/api/v1/guide_video.py create mode 100644 app/models/guide_video.py create mode 100644 app/repositories/guide_video.py create mode 100644 app/schemas/guide_video.py create mode 100644 app/utils/mt_search_cursor.py create mode 100644 scripts/reset_guide_video.py create mode 100644 scripts/seed_meituan_coupon_mock.py diff --git a/alembic/versions/guide_video_play_table.py b/alembic/versions/guide_video_play_table.py new file mode 100644 index 0000000..3c6a62e --- /dev/null +++ b/alembic/versions/guide_video_play_table.py @@ -0,0 +1,50 @@ +"""新手引导视频播放记录表(领券浮层前 N 次替代广告) + +见 app/models/guide_video.py:按账号计次(开播即计数)、play_token 幂等发币。 +配置(开关 / 视频地址 / 次数 / 金币)复用既有 app_config 表,无需建表。 + +Revision ID: guide_video_play_table +Revises: meituan_coupon_feed_indexes +Create Date: 2026-07-23 12:00:00.000000 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "guide_video_play_table" +down_revision: Union[str, Sequence[str], None] = "meituan_coupon_feed_indexes" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "guide_video_play", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("play_token", sa.String(length=64), nullable=False), + sa.Column("scene", sa.String(length=16), nullable=False, server_default="coupon"), + sa.Column("seq", sa.Integer(), nullable=False, server_default="1"), + sa.Column("video_url", sa.String(length=512), nullable=True), + sa.Column("coin", sa.Integer(), nullable=False, server_default="0"), + sa.Column("status", sa.String(length=16), nullable=False, server_default="playing"), + sa.Column("completed", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "started_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False + ), + sa.Column("granted_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["user_id"], ["user.id"]), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("play_token", name="uq_guide_video_play_token"), + ) + op.create_index("ix_guide_video_play_user_id", "guide_video_play", ["user_id"]) + op.create_index("ix_guide_video_play_started_at", "guide_video_play", ["started_at"]) + + +def downgrade() -> None: + op.drop_index("ix_guide_video_play_started_at", table_name="guide_video_play") + op.drop_index("ix_guide_video_play_user_id", table_name="guide_video_play") + op.drop_table("guide_video_play") diff --git a/alembic/versions/meituan_coupon_feed_indexes.py b/alembic/versions/meituan_coupon_feed_indexes.py new file mode 100644 index 0000000..c120ca6 --- /dev/null +++ b/alembic/versions/meituan_coupon_feed_indexes.py @@ -0,0 +1,52 @@ +"""meituan_coupon 首页 feed 分页复合索引(销量最高 / 智能推荐) + +「销量最高」「智能推荐」两个 tab 都是 + WHERE city_id = ? [+ 过滤] → DISTINCT ON (dedup_key) ORDER BY dedup_key, <排序键> DESC +的形状。列顺序对齐后 Postgres 可以顺着索引流式去重,免掉「每翻一页就把该城全部券重排一遍」, +这是首页下滑到底越来越慢的根因之一(另一半在 app 层:见 api/v1/meituan.py 的 _paged_dedup_ids)。 + +⚠️ 本文件同时是一个 **merge 迁移**:主干此前有 3 个并行 head +(comparison_user_created_idx / monitoring_audit_rbac / notification_table), +`alembic upgrade head` 会因 multiple heads 报错。这里一并收敛回单 head。 + +Revision ID: meituan_coupon_feed_indexes +Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table +Create Date: 2026-07-23 10:00:00.000000 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "meituan_coupon_feed_indexes" +down_revision: Union[str, Sequence[str], None] = ( + "comparison_user_created_idx", + "monitoring_audit_rbac", + "notification_table", +) +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # 销量最高:WHERE city_id=? AND sale_volume_num IS NOT NULL + # ORDER BY dedup_key, sale_volume_num DESC, commission_percent DESC + op.create_index( + "ix_meituan_coupon_city_dedup_sales", + "meituan_coupon", + ["city_id", "dedup_key", sa.text("sale_volume_num DESC"), sa.text("commission_percent DESC")], + ) + # 智能推荐:WHERE city_id=? AND commission_percent>=3.0 + # ORDER BY dedup_key, commission_percent DESC + op.create_index( + "ix_meituan_coupon_city_dedup_comm", + "meituan_coupon", + ["city_id", "dedup_key", sa.text("commission_percent DESC")], + ) + + +def downgrade() -> None: + op.drop_index("ix_meituan_coupon_city_dedup_comm", table_name="meituan_coupon") + op.drop_index("ix_meituan_coupon_city_dedup_sales", table_name="meituan_coupon") diff --git a/app/admin/main.py b/app/admin/main.py index c869487..918b238 100644 --- a/app/admin/main.py +++ b/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) diff --git a/app/admin/routers/guide_video.py b/app/admin/routers/guide_video.py new file mode 100644 index 0000000..59f75ae --- /dev/null +++ b/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) diff --git a/app/admin/schemas/guide_video.py b/app/admin/schemas/guide_video.py new file mode 100644 index 0000000..721f0a6 --- /dev/null +++ b/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) diff --git a/app/api/v1/guide_video.py b/app/api/v1/guide_video.py new file mode 100644 index 0000000..c38251b --- /dev/null +++ b/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) diff --git a/app/api/v1/meituan.py b/app/api/v1/meituan.py index c1d07a1..b3cf1c6 100644 --- a/app/api/v1/meituan.py +++ b/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) diff --git a/app/core/config.py b/app/core/config.py index afae3ee..81149db 100644 --- a/app/core/config.py +++ b/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)。 diff --git a/app/core/media.py b/app/core/media.py index 192b186..a6e84e4 100644 --- a/app/core/media.py +++ b/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 与浏览器