Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2004e8c910 |
@@ -1,26 +0,0 @@
|
||||
"""merge notification/comparison_user_idx/monitoring_audit heads
|
||||
|
||||
Revision ID: 8e04cc13a211
|
||||
Revises: comparison_user_created_idx, monitoring_audit_rbac, notification_table
|
||||
Create Date: 2026-07-23 15:37:26.967540
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = '8e04cc13a211'
|
||||
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:
|
||||
pass
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
pass
|
||||
@@ -1,26 +0,0 @@
|
||||
"""merge guide_video and main alembic heads
|
||||
|
||||
Revision ID: d9c03cc3ea07
|
||||
Revises: 8e04cc13a211, guide_video_play_table
|
||||
Create Date: 2026-07-23 22:57:40.998161
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'd9c03cc3ea07'
|
||||
down_revision: Union[str, Sequence[str], None] = ('8e04cc13a211', 'guide_video_play_table')
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
pass
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
pass
|
||||
@@ -1,50 +0,0 @@
|
||||
"""新手引导视频播放记录表(领券浮层前 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")
|
||||
@@ -1,52 +0,0 @@
|
||||
"""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")
|
||||
@@ -30,7 +30,6 @@ 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
|
||||
@@ -41,7 +40,6 @@ 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")
|
||||
@@ -55,8 +53,6 @@ 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")
|
||||
|
||||
|
||||
@@ -105,7 +101,6 @@ 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)
|
||||
|
||||
@@ -1,101 +0,0 @@
|
||||
"""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)
|
||||
@@ -1,25 +0,0 @@
|
||||
"""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)
|
||||
@@ -1,64 +0,0 @@
|
||||
"""新手引导视频(领券等候浮层前 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)
|
||||
+75
-154
@@ -5,12 +5,11 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from sqlalchemy import nullslast, select
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy.orm import Session, aliased
|
||||
|
||||
from app.core.config import settings
|
||||
from app.db.session import get_db
|
||||
@@ -26,14 +25,8 @@ 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")
|
||||
|
||||
|
||||
@@ -116,86 +109,6 @@ 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
|
||||
@@ -220,21 +133,17 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
|
||||
return [], True
|
||||
|
||||
# 距离最近:搜索召回(外卖搜"外卖" + 到店搜"美食",都 sortField=6 离我最近)一页页拉。
|
||||
# 搜索翻页必须用 searchId(pageNo 翻不动),而接口是无状态的(客户端只传页码)—— 原实现因此
|
||||
# 每次都从第 1 页顺序重放到第 N 页,取第 N 页要向美团发 N 次请求,越往下滑越慢。
|
||||
# 现在把沿途 searchId 记进 [mt_search_cursor],稳态下每翻一页恒定 1 次请求;两路仍并行。
|
||||
# 按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
|
||||
# 搜索翻页必须用 searchId(pageNo 翻不动),所以每个 feed 页顺序翻到第 N 页;两路并行、page 1 最快。
|
||||
# 无状态、不改 APP(传页码即可);按你位置实时算距离(库里没存 POI 经纬度,只能实时)。
|
||||
if tab == "distance":
|
||||
lon_i, lat_i = int(lon * 1_000_000), int(lat * 1_000_000)
|
||||
|
||||
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)。"""
|
||||
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
|
||||
data: list[dict] = []
|
||||
has_next = False
|
||||
for pg in range(start, n + 1):
|
||||
for pg in range(1, n + 1):
|
||||
body: dict = {
|
||||
"platform": platform, "searchText": keyword, "sortField": 6,
|
||||
"longitude": lon_i, "latitude": lat_i, "pageSize": 20,
|
||||
@@ -252,27 +161,10 @@ 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)
|
||||
@@ -302,25 +194,39 @@ def feed(req: FeedRequest, db: Session = Depends(get_db)) -> FeedResponse:
|
||||
return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
|
||||
PAGE = 20
|
||||
try:
|
||||
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 兜底稳定分页
|
||||
page_order=lambda c: [
|
||||
nullslast(c.sale_volume_num.desc()), c.commission_percent.desc(), c.id,
|
||||
],
|
||||
page=req.page, page_size=PAGE,
|
||||
base = select(MeituanCoupon).where(
|
||||
MeituanCoupon.commission_percent >= 3.0,
|
||||
MeituanCoupon.city_id == city_id,
|
||||
)
|
||||
raws = _load_raws(db, ids)
|
||||
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)
|
||||
# 销量高的优先(无销量档排后),同档佣金高优先,id 兜底稳定分页
|
||||
.order_by(nullslast(m.sale_volume_num.desc()), m.commission_percent.desc(), m.id)
|
||||
.offset(start)
|
||||
.limit(PAGE + 1)
|
||||
).scalars().all()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("[feed] rec 库查询失败,降级返空")
|
||||
return FeedResponse(items=[], has_next=False, page=req.page, status="degraded")
|
||||
cards = _cards_from_raws(raws, hide_distance=True)
|
||||
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)
|
||||
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)
|
||||
@@ -376,36 +282,51 @@ 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 条(见 _paged_dedup_ids 的性能说明)。
|
||||
# 去重 + 排序 + 分页全在 SQL 做,每页只取并解析当前页 ~20 条。
|
||||
# (之前实现每翻一页都全表拉取 + 全量 from_raw 解析,翻页慢 → 客户端滑动卡顿/翻不动。)
|
||||
# 库为空(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:
|
||||
ids, has_next = _paged_dedup_ids(
|
||||
db,
|
||||
conds=conds,
|
||||
# 每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
|
||||
dedup_order=[
|
||||
MeituanCoupon.sale_volume_num.desc(),
|
||||
MeituanCoupon.commission_percent.desc(),
|
||||
],
|
||||
# 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
|
||||
page_order=lambda c: [
|
||||
c.sale_volume_num.desc(), c.commission_percent.desc(), c.id,
|
||||
],
|
||||
page=req.page, page_size=req.page_size,
|
||||
# 1) DISTINCT ON (dedup_key):每个去重键(品牌|名|价)只留销量最高那条(同销量再按佣金)
|
||||
base = select(MeituanCoupon).where(
|
||||
MeituanCoupon.sale_volume_num.isnot(None),
|
||||
MeituanCoupon.city_id == city_id,
|
||||
)
|
||||
raws = _load_raws(db, ids)
|
||||
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)
|
||||
# 加 id 作稳定 tiebreaker:同销量同佣金的并列项排序确定,避免跨页重复/漏项
|
||||
.order_by(m.sale_volume_num.desc(), m.commission_percent.desc(), m.id)
|
||||
.offset(start)
|
||||
.limit(req.page_size + 1)
|
||||
).scalars().all()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("[top-sales] 库查询失败,降级返空")
|
||||
return CouponListResponse(items=[], has_next=False, search_id=None, status="degraded")
|
||||
|
||||
# 不显示距离:库里的距离是相对城市默认点的(对用户无意义、且误导),与推荐流口径一致。
|
||||
cards = _cards_from_raws(raws, hide_distance=True)
|
||||
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)
|
||||
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)
|
||||
|
||||
@@ -384,9 +384,6 @@ 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)。
|
||||
|
||||
@@ -77,35 +77,6 @@ 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)
|
||||
@@ -145,8 +116,3 @@ 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)
|
||||
|
||||
@@ -11,7 +11,6 @@ import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
@@ -26,44 +25,6 @@ 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()
|
||||
|
||||
@@ -101,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 = get_client().post(url, content=body, headers=headers)
|
||||
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
|
||||
|
||||
@@ -29,7 +29,6 @@ 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
|
||||
@@ -71,7 +70,6 @@ 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")
|
||||
@@ -88,9 +86,6 @@ 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
|
||||
@@ -113,7 +108,6 @@ 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")
|
||||
|
||||
|
||||
@@ -160,8 +154,6 @@ 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)
|
||||
|
||||
@@ -27,7 +27,6 @@ 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,
|
||||
|
||||
@@ -1,62 +0,0 @@
|
||||
"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
|
||||
|
||||
产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
|
||||
改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。
|
||||
|
||||
口径:
|
||||
- **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
|
||||
- **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
|
||||
写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill
|
||||
App 也算用掉一次(产品选定口径,防反复进出刷金币)。
|
||||
- **发币幂等**靠 play_token 唯一键:同一次播放重复上报只发一次。
|
||||
|
||||
与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, 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"),
|
||||
)
|
||||
|
||||
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 为准。
|
||||
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}>"
|
||||
)
|
||||
@@ -23,7 +23,7 @@ from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import JSON, DateTime, Float, Index, Integer, String, UniqueConstraint, func
|
||||
from sqlalchemy import JSON, DateTime, Float, Integer, String, UniqueConstraint, func
|
||||
from sqlalchemy.dialects.postgresql import JSONB
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
@@ -95,23 +95,3 @@ 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(),
|
||||
)
|
||||
|
||||
@@ -1,257 +0,0 @@
|
||||
"""新手引导视频:运营配置读写 + 播放计次 + 发币。
|
||||
|
||||
**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 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 后不再下发,客户端改放广告(原逻辑)。
|
||||
|
||||
**发币**幂等键是 play_token:同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发
|
||||
都靠它挡住)。中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
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)
|
||||
miss = {
|
||||
"should_play": False,
|
||||
"video_url": None,
|
||||
"play_token": "",
|
||||
"reward_coin": reward_coin,
|
||||
"seq": used,
|
||||
"remaining": max(0, max_plays - used),
|
||||
}
|
||||
if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
|
||||
return miss
|
||||
|
||||
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)
|
||||
if commit:
|
||||
db.commit()
|
||||
else:
|
||||
db.flush()
|
||||
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 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()
|
||||
play = db.execute(
|
||||
select(GuideVideoPlay).where(
|
||||
GuideVideoPlay.play_token == token,
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if play is None:
|
||||
return {"granted": False, "coin": 0, "status": "not_found"}
|
||||
if play.status == "granted":
|
||||
return {"granted": False, "coin": play.coin, "status": "already_granted"}
|
||||
|
||||
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
|
||||
coin = int(get_config(db).get("reward_coin") or 0)
|
||||
play.completed = 1 if completed else 0
|
||||
play.status = "granted"
|
||||
play.coin = coin
|
||||
play.granted_at = datetime.now(rewards.CN_TZ).replace(tzinfo=None)
|
||||
if coin > 0:
|
||||
crud_wallet.grant_coins(
|
||||
db,
|
||||
user_id,
|
||||
coin,
|
||||
biz_type=BIZ_TYPE,
|
||||
ref_id=token,
|
||||
remark="新手引导视频奖励",
|
||||
)
|
||||
try:
|
||||
db.commit()
|
||||
except IntegrityError:
|
||||
# 并发双发(播完 + ✕ 同时到)时其中一条会撞唯一键/行锁,回滚后按已发返回。
|
||||
db.rollback()
|
||||
again = db.execute(
|
||||
select(GuideVideoPlay).where(
|
||||
GuideVideoPlay.play_token == token,
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
return {"granted": False, "coin": again.coin if again else 0, "status": "already_granted"}
|
||||
return {"granted": True, "coin": coin, "status": "granted"}
|
||||
@@ -1,36 +0,0 @@
|
||||
"""新手引导视频(领券等候浮层前 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
|
||||
@@ -1,97 +0,0 @@
|
||||
"""美团搜索翻页游标(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)
|
||||
@@ -30,8 +30,7 @@
|
||||
## 各 tab 行为
|
||||
- **`rec` 智能推荐**:走【离线库 `meituan_coupon`】筛佣金率 ≥ 3%,`DISTINCT ON(dedup_key)` 去重后按销量降序分页。**纯库查询、不打美团、不依赖 MT 凭证**。库为空(prod 刚部署 / ETL 未跑完)→ `empty`;库异常 → `degraded`。**不显示距离**(库里距离相对城市默认点,对用户无意义)。
|
||||
- 为何不实时:实测同城热销榜中位佣金 ~0.8%,筛佣金≥3% 后每页剩 0–1 条,既撞 402 又填不满,故从库出;库空时也**不回退实时**(回退同样填不满)。
|
||||
- **`distance` 距离最近**:外卖搜「外卖」+ 到店搜「美食」(均 `sortField=6`),两路并行翻页,实时按你坐标算距离、由近及远。两路**都失败**且无结果 → `degraded`;否则有结果 `ok` / 无结果 `empty`。
|
||||
- 翻页成本:美团搜索只能靠 `searchId` 续页,本接口又是无状态的(客户端只传页码)。服务端把沿途 `searchId` 按「量化坐标(~1km)+平台+关键词」缓存 10 分钟(`utils/mt_search_cursor`),**稳态下每翻一页恒定 1 次上游请求**(改前是「取第 N 页 = 发 N 次」);缓存冷/过期才从最近的已知页往后重放。缓存是进程内的,多 worker 不共享 —— 只影响快慢,不影响结果。
|
||||
- **`distance` 距离最近**:外卖搜「外卖」+ 到店搜「美食」(均 `sortField=6`),两路并行顺序翻页,实时按你坐标算距离、由近及远。两路**都失败**且无结果 → `degraded`;否则有结果 `ok` / 无结果 `empty`。
|
||||
- **默认(空 tab)**:逐轮分页的混合 feed(2 外卖 + 1 到店交叉,写死 3 页:爆款 / 今日必推 / 精选+限时),第 4 页返空。两路榜单都失败 → `degraded`。
|
||||
|
||||
## 错误码
|
||||
|
||||
@@ -17,7 +17,6 @@
|
||||
|
||||
## 说明
|
||||
- 从 `meituan_coupon` 取 `sale_volume_num` 非空 **且 `city_id` = 反查城市** 的券(#116,同城销量榜),`DISTINCT ON(dedup_key)` 跨源去重(每个「品牌|名|价」只留销量最高一条,同销量再按佣金),按销量降序分页;每页只对当前 ~20 条做 `from_raw` 解析(翻页快,不全表拉取)。
|
||||
- 分页查询分两步(与 `rec` 共用 `_paged_dedup_ids`):**① 只在 `id + 排序键` 这几个小列上去重/排序/分页,② 再按 id 回表取本页 `raw`**。`raw` 是整条美团原始返回(JSONB,每行数 KB),让它参与排序会把整城数据推过 `work_mem`、落盘做外部归并,而且每翻一页都重来一遍 —— 这是此前「滑到底越来越慢」的主因之一。配套索引 `ix_meituan_coupon_city_dedup_sales` / `..._comm`(见 `alembic/versions/meituan_coupon_feed_indexes.py`)让 `DISTINCT ON` 顺着索引流式去重,免掉排序。
|
||||
- **不依赖 MT 凭证**(纯库查询)。库为空(prod 刚部署 / ETL 未跑完)→ `status=empty`;库查询异常 → `status=degraded`。均返 `200`、不抛 5xx。
|
||||
- **仅 PostgreSQL**(`DISTINCT ON` 为 PG 专用)。
|
||||
|
||||
|
||||
@@ -42,12 +42,8 @@
|
||||
## 索引与约束
|
||||
- 唯一约束 `uq_meituan_coupon_source_sign`(`source`, `product_view_sign`)—— ETL upsert 的冲突目标。
|
||||
- 单列 index:`source`、`city_id`、`brand_name`、`sale_volume_num`、`commission_percent`、`dedup_key`、`last_seen`。
|
||||
- 复合 index(`meituan_coupon_feed_indexes` 迁移,2026-07-23 加):
|
||||
- `ix_meituan_coupon_city_dedup_sales`(`city_id`, `dedup_key`, `sale_volume_num DESC`, `commission_percent DESC`)——「销量最高」。
|
||||
- `ix_meituan_coupon_city_dedup_comm`(`city_id`, `dedup_key`, `commission_percent DESC`)——「智能推荐」。
|
||||
- 列顺序对齐 `WHERE city_id=? … ORDER BY dedup_key, <排序键> DESC`,让 `DISTINCT ON` 顺着索引流式去重、免掉排序。
|
||||
- **字段核对(2026-06-10)**:模型 ↔ 迁移(`meituan_coupon_table`)↔ ETL 解析三处一致;解析截断宽度均 ≤ 列宽(name256 / brand128 / poi128 / head512 / sign128,`dedup_key` = md5 32 字符 ≤ 64)。
|
||||
- **索引评估修订(2026-07-23)**:此前判断「单城几千行 seq scan 即亚毫秒,复合索引留作放量时再加」——**漏算了 `raw`**。原查询用 `select(MeituanCoupon)` 做子查询,`raw`(JSONB,每行数 KB)被一起拖进 `DISTINCT ON` 与分页两轮排序,整城体量轻松推过 `work_mem` → 落盘外部归并,且**每翻一页重来一次**,是首页「滑到底越来越慢」的主因之一。现已双管齐下:接口侧改成「先在小列上排出本页 id,再按 id 回表取 `raw`」(`api/v1/meituan.py` 的 `_paged_dedup_ids`),库侧补上上面两条复合索引。
|
||||
- **索引评估**:`rec` / `sales` 查询的理想索引是复合 `(dedup_key, sale_volume_num DESC)` 与 `(dedup_key, commission_percent DESC)`(匹配 `DISTINCT ON`);但当前**单城几千行,seq scan 即亚毫秒**,现有单列索引已够用,复合索引留作多城 / 放量时再加,避免过早优化。
|
||||
|
||||
## 过期清理策略
|
||||
- ETL 每轮(每小时)末尾按 **`last_seen` 的 TTL** 清理:`DELETE WHERE last_seen < now() - prune_hours`(默认 24h,`--prune-hours` 可调,0=不清)。
|
||||
|
||||
@@ -1,216 +0,0 @@
|
||||
"""重置指定账号的「领券引导视频」已播次数,让领券等候浮层重新放引导视频,方便反复看这支片。
|
||||
|
||||
原理:浮层放不放引导视频只由两件事决定(见 app/repositories/guide_video.py `start_play`):
|
||||
1. 运营配置 app_config(key=coupon_guide_video):enabled / video_url / max_plays;
|
||||
2. 该账号 guide_video_play 的**行数**(开播即计次)—— 行数 >= max_plays 就不再下发,
|
||||
`/api/v1/guide-video/start` 返 should_play=false,客户端照旧放广告。
|
||||
所以「想再看一遍」= 删掉这个账号的 guide_video_play 行。配置本脚本**只读不改**(要换片子 /
|
||||
开关去运营后台「配置 - 领券引导视频」;次数 3 / 金币 120 是服务端默认值,后台不开放调整),
|
||||
这样不会把别的账号的线上行为一起动了。
|
||||
|
||||
默认连**当时发的金币一起退**(每次 reward_coin,现配置 120):这些流水 biz_type='guide_video',
|
||||
不退的话每重置一轮余额就白涨一轮,收益明细里还会堆出一串「新手引导视频奖励」。想留着用 --keep-coins。
|
||||
例外:金币已被兑换成现金、余额兜不住时**整笔跳过不退** —— coin_balance 必须恒等于流水总和,
|
||||
硬退会退成负数,夹到 0 又会吃掉别处赚的金币,两种做法都会让账对不上(同 reset_signin_today.py)。
|
||||
|
||||
用法(在项目根、已 pip install -e . 的环境里跑):
|
||||
python scripts/reset_guide_video.py # 默认测试号 11111111111
|
||||
python scripts/reset_guide_video.py 13800138000 # 指定手机号
|
||||
python scripts/reset_guide_video.py --user-id 5 # 直接指定 user_id
|
||||
python scripts/reset_guide_video.py --dry-run # 预览(照常执行再回滚),不落库
|
||||
python scripts/reset_guide_video.py --keep-coins # 只清次数,已发金币不退
|
||||
|
||||
走 SessionLocal 连 DATABASE_URL(SQLite / Postgres 都行),因此**只允许 APP_ENV=dev 改库**
|
||||
(--dry-run 只读,任何环境都能跑)。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.core.config import settings
|
||||
from app.db.session import SessionLocal, engine
|
||||
from app.models.guide_video import GuideVideoPlay
|
||||
from app.models.user import User
|
||||
from app.models.wallet import CoinAccount, CoinTransaction
|
||||
from app.repositories import guide_video as crud_guide
|
||||
|
||||
# Windows 控制台默认 GBK,强制 UTF-8 否则中文输出乱码。stderr 也要设:
|
||||
# SystemExit(如"用户不存在")的中文提示走的是 stderr。
|
||||
for _stream in (sys.stdout, sys.stderr):
|
||||
if hasattr(_stream, "reconfigure"):
|
||||
_stream.reconfigure(encoding="utf-8")
|
||||
|
||||
# dev 下 engine 是 echo=True(APP_DEBUG),几十行 SQL 会把前后对比刷没。echo 走 SQLAlchemy 自己的
|
||||
# InstanceLogger,不吃 logging.setLevel,只能改 engine.echo。
|
||||
engine.echo = False
|
||||
|
||||
DEFAULT_PHONE = "11111111111"
|
||||
|
||||
|
||||
def resolve_user(db, phone: str, user_id: int | None) -> User:
|
||||
if user_id is not None:
|
||||
user = db.get(User, user_id)
|
||||
if user is None:
|
||||
raise SystemExit(f"user_id={user_id} 不存在")
|
||||
return user
|
||||
user = db.execute(select(User).where(User.phone == phone)).scalar_one_or_none()
|
||||
if user is None:
|
||||
raise SystemExit(f"手机号 {phone} 没有对应用户(注意 phone 才是登录账号,username 是展示 ID)")
|
||||
return user
|
||||
|
||||
|
||||
def load_plays(db, user_id: int) -> list[GuideVideoPlay]:
|
||||
return list(db.execute(
|
||||
select(GuideVideoPlay)
|
||||
.where(GuideVideoPlay.user_id == user_id)
|
||||
.order_by(GuideVideoPlay.id)
|
||||
).scalars().all())
|
||||
|
||||
|
||||
def print_config(db) -> dict:
|
||||
"""打印运营配置 + 片子是否真在盘上(配了地址但文件丢了,客户端会黑屏/加载失败)。"""
|
||||
cfg = crud_guide.get_config(db)
|
||||
video_url = (cfg.get("video_url") or "").strip()
|
||||
print("--- 运营配置(app_config: coupon_guide_video,本脚本不改) ---")
|
||||
print(f" enabled={cfg['enabled']} max_plays={cfg['max_plays']} reward_coin={cfg['reward_coin']}")
|
||||
print(f" video_url={video_url or '(未配片)'}")
|
||||
if video_url.startswith(settings.MEDIA_URL_PREFIX + "/"):
|
||||
rel = video_url[len(settings.MEDIA_URL_PREFIX) + 1:]
|
||||
f = Path(settings.MEDIA_ROOT) / rel
|
||||
if f.is_file():
|
||||
print(f" 文件: {f} ({f.stat().st_size / 1024 / 1024:.1f} MB) ✓")
|
||||
else:
|
||||
print(f" ⚠️ 文件不存在: {f} —— 客户端会加载失败,请去运营后台重新上传")
|
||||
return cfg
|
||||
|
||||
|
||||
def print_state(db, user: User, cfg: dict, label: str) -> None:
|
||||
plays = load_plays(db, user.id)
|
||||
max_plays = int(cfg.get("max_plays") or 0)
|
||||
used = len(plays)
|
||||
print(f"--- {label} ---")
|
||||
print(f" guide_video_play: {used} 行(已用次数,开播即计),上限 {max_plays}")
|
||||
for p in plays:
|
||||
print(f" #{p.id} 第{p.seq}次 {p.status} +{p.coin}金币 "
|
||||
f"completed={p.completed} 开播于 {p.started_at} token={p.play_token[:12]}…")
|
||||
|
||||
# 用仓储层原样复算一遍"下次会不会放",而不是脚本里自己判规则 —— 这就是客户端会拿到的结果
|
||||
enabled = bool(cfg.get("enabled"))
|
||||
has_video = bool((cfg.get("video_url") or "").strip())
|
||||
will_play = enabled and has_video and max_plays > 0 and used < max_plays
|
||||
reason = (
|
||||
"会放引导视频" if will_play
|
||||
else "开关关着" if not enabled
|
||||
else "没配片" if not has_video
|
||||
else "max_plays=0" if max_plays <= 0
|
||||
else f"次数已用完({used}/{max_plays})"
|
||||
)
|
||||
print(f" [下次点一键领取] should_play={will_play} —— {reason}")
|
||||
|
||||
acc = db.get(CoinAccount, user.id)
|
||||
if acc is None:
|
||||
print(" coin_account: (无)")
|
||||
else:
|
||||
print(f" coin_account: coin={acc.coin_balance} earned={acc.total_coin_earned}")
|
||||
|
||||
|
||||
def refund_plays(db, user_id: int, tokens: list[str]) -> None:
|
||||
"""退回这些播放发的金币:删流水 + 扣余额。余额兜不住的整笔跳过(见模块 docstring)。"""
|
||||
if not tokens:
|
||||
return
|
||||
rows = list(db.execute(
|
||||
select(CoinTransaction).where(
|
||||
CoinTransaction.user_id == user_id,
|
||||
CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
|
||||
CoinTransaction.ref_id.in_(tokens),
|
||||
)
|
||||
).scalars().all())
|
||||
if not rows:
|
||||
print(" 没有可退的金币流水(可能是 reward_coin=0,或本来就没发过)")
|
||||
return
|
||||
acc = db.get(CoinAccount, user_id)
|
||||
if acc is None:
|
||||
return
|
||||
|
||||
# 从最近一笔往回退,退到余额兜不住为止 —— 保证"最新发的那笔"一定被退掉。
|
||||
rows.sort(key=lambda r: r.id, reverse=True)
|
||||
refundable: list[CoinTransaction] = []
|
||||
total = 0
|
||||
for r in rows:
|
||||
if total + r.amount > acc.coin_balance:
|
||||
break
|
||||
refundable.append(r)
|
||||
total += r.amount
|
||||
|
||||
for r in refundable:
|
||||
db.delete(r)
|
||||
if total:
|
||||
acc.coin_balance -= total
|
||||
acc.total_coin_earned = max(0, acc.total_coin_earned - total)
|
||||
print(f" 已退回 {total} 金币({len(refundable)}/{len(rows)} 笔引导视频奖励)")
|
||||
|
||||
stuck = len(rows) - len(refundable)
|
||||
if stuck:
|
||||
print(f" ⚠️ 还有 {stuck} 笔退不掉(金币已被兑换/花掉,余额 {acc.coin_balance} 兜不住),原样保留 —— "
|
||||
"硬退会让余额和流水总和对不上。收益明细里会多留几条,不影响再看视频。")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="重置账号的领券引导视频次数,让浮层重新放引导视频")
|
||||
parser.add_argument("phone", nargs="?", default=DEFAULT_PHONE,
|
||||
help=f"手机号(默认 {DEFAULT_PHONE})")
|
||||
parser.add_argument("--user-id", type=int, default=None, help="直接按 user_id 定位,优先于 phone")
|
||||
parser.add_argument("--keep-coins", action="store_true",
|
||||
help="不退已发金币(余额会越测越高,收益明细堆重复流水)")
|
||||
parser.add_argument("--dry-run", action="store_true", help="预览,最后回滚不落库")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.dry_run and settings.APP_ENV != "dev":
|
||||
raise SystemExit(f"APP_ENV={settings.APP_ENV},拒绝改库(只有 dev 能改;--dry-run 可任意环境)")
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
user = resolve_user(db, args.phone, args.user_id)
|
||||
print(f"DB: {settings.DATABASE_URL} APP_ENV: {settings.APP_ENV}")
|
||||
print(f"用户: id={user.id} phone={user.phone} username={user.username} keep_coins: {args.keep_coins}")
|
||||
cfg = print_config(db)
|
||||
print_state(db, user, cfg, "before")
|
||||
|
||||
plays = load_plays(db, user.id)
|
||||
if not plays:
|
||||
print("该账号没有任何播放记录,次数本来就是满的,无需重置。")
|
||||
db.rollback()
|
||||
return
|
||||
|
||||
tokens = [p.play_token for p in plays]
|
||||
for p in plays:
|
||||
db.delete(p)
|
||||
db.flush()
|
||||
|
||||
if args.keep_coins:
|
||||
print("(--keep-coins:保留已发金币,流水和余额不动)")
|
||||
else:
|
||||
refund_plays(db, user.id, tokens)
|
||||
# 注:更早流水的 balance_after 是当时的快照,不回改 —— 收益明细里历史行的余额列
|
||||
# 会与现余额对不上,dev 测试库无妨。
|
||||
|
||||
print_state(db, user, cfg, "after")
|
||||
|
||||
if args.dry_run:
|
||||
db.rollback()
|
||||
print(f"(dry-run:以上 after 为预览,已回滚,库没动;真跑会删 {len(plays)} 条播放记录)")
|
||||
return
|
||||
db.commit()
|
||||
print(f"完成:删掉 {len(plays)} 条播放记录,{user.phone} 下次点「一键自动领取」浮层会重新放引导视频。")
|
||||
print("提醒:客户端是在浮层建窗时调 /guide-video/start 的,不用重装、不用重登,直接再点一次一键领取即可;"
|
||||
"但「广告显示」开关关掉时整块浮层不建窗(连引导视频也不放),测之前先确认它是开的。")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,222 +0,0 @@
|
||||
"""本地 mock:往 meituan_coupon 灌一批假券,让首页「智能推荐 / 销量最高」在无美团凭证时也有数据。
|
||||
|
||||
为什么需要:这两个 tab 不实时打美团,只查 `meituan_coupon` 表,数据靠 scripts/pull_meituan_coupons.py
|
||||
定时抓。本地既没 MT_CPS 凭证、也没线上导出的 TSV 时,表是空的 → 接口返 status=empty,页面永远空白,
|
||||
分页 / 左右滑动 / 触底加载全都没法验。本脚本纯造数,不联网、不需要任何凭证。
|
||||
|
||||
覆盖不到的:「距离最近」tab 必须实时打美团搜索接口(库里没存 POI 经纬度),假数据救不了;
|
||||
点「抢」换推广链接也会失败(productViewSign 是假的)。要验这两条只能配 MT_CPS_APP_KEY/SECRET。
|
||||
|
||||
图片:生成纯色 PNG 落到 data/media/mock_coupon/,由后端 /media 静态服务出图。
|
||||
文件名**带 FEED_THUMB_PARAM 后缀**是故意的 —— schemas/meituan.py 的 feed_image_url 会给每个
|
||||
headUrl 无条件拼上该后缀(美团 CDN 的缩放参数),本地静态服务不认参数、只会当成文件名的一部分,
|
||||
所以磁盘上的文件必须叫 `xxx.png@375w_375h_1e_1c.webp` 才取得到。
|
||||
|
||||
city_id 必须与端上定位反查出的一致:rec/top-sales 都按 city_id 过滤,而 city_id 由经纬度经
|
||||
app/utils/meituan_city.py 反查。默认灌北京,测试机的模拟定位也要设成北京,否则照样是空。
|
||||
|
||||
幂等:重跑先按 MOCK_SIGN_PREFIX 清掉旧 mock 行再重建,不碰真实抓取的数据。
|
||||
|
||||
python -m scripts.seed_meituan_coupon_mock # 默认北京 400 条
|
||||
python -m scripts.seed_meituan_coupon_mock --count 800
|
||||
python -m scripts.seed_meituan_coupon_mock --city-id 2QSF6IG3KMDXWO5VP7FXHMMKXA # 上海
|
||||
python -m scripts.seed_meituan_coupon_mock --clean-only # 只清 mock 数据
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import random
|
||||
import struct
|
||||
import sys
|
||||
import zlib
|
||||
from pathlib import Path
|
||||
|
||||
from sqlalchemy import delete
|
||||
|
||||
from app.core.config import settings
|
||||
from app.db.session import SessionLocal
|
||||
from app.models.meituan_coupon import MeituanCoupon
|
||||
from app.schemas.meituan import FEED_THUMB_PARAM
|
||||
|
||||
if hasattr(sys.stdout, "reconfigure"):
|
||||
sys.stdout.reconfigure(encoding="utf-8") # Windows 控制台输出中文
|
||||
|
||||
# mock 行的 productViewSign 前缀:清理时按此删,保证不误伤真实抓取的券。
|
||||
MOCK_SIGN_PREFIX = "MOCK-"
|
||||
|
||||
# 北京。其他城市的 id 见 app/integrations/data/meituan_cities.json。
|
||||
DEFAULT_CITY_ID = "WKV2HMXUEK634WP64CUCUQGM64"
|
||||
|
||||
_IMG_DIR = Path(settings.MEDIA_ROOT) / "mock_coupon"
|
||||
_IMG_COUNT = 16
|
||||
|
||||
_BRANDS = [
|
||||
"蜜雪冰城", "瑞幸咖啡", "肯德基", "麦当劳", "华莱士", "塔斯汀",
|
||||
"沪上阿姨", "古茗", "茶百道", "霸王茶姬", "正新鸡排", "杨国福麻辣烫",
|
||||
"绝味鸭脖", "喜茶", "奈雪的茶", "汉堡王", "德克士", "必胜客",
|
||||
"老乡鸡", "南城香",
|
||||
]
|
||||
_WAIMAI_ITEMS = [
|
||||
"单人套餐", "双人超值餐", "3选1套餐", "招牌奶茶券", "全场通用券",
|
||||
"汉堡可乐套餐", "炸鸡拼盘", "麻辣烫单人餐", "早餐组合", "夜宵拼盘",
|
||||
]
|
||||
_DAODIAN_ITEMS = [
|
||||
"10元代金券", "20元代金券", "50元代金券", "双人自助餐", "四人聚餐套餐",
|
||||
"下午茶双人套餐", "火锅双人餐", "烤肉四人餐",
|
||||
]
|
||||
# (saleVolume 文案, 排序用下界) —— 与 pull 脚本的 _sale_volume_num 解析口径一致
|
||||
_SALE_VOLUMES = [
|
||||
("月售99+", 99), ("月售199+", 199), ("月售500+", 500), ("月售999+", 999),
|
||||
("热销2000+", 2000), ("热销5000+", 5000), ("热销1w+", 10000), ("热销3w+", 30000),
|
||||
]
|
||||
_PRICE_LABELS = [None, "15天低价", "30天低价", "近期低价"]
|
||||
|
||||
|
||||
def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
|
||||
"""生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。"""
|
||||
def _chunk(typ: bytes, data: bytes) -> bytes:
|
||||
body = typ + data
|
||||
return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)
|
||||
|
||||
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) # 8bit/通道, color type 2 = RGB
|
||||
row = b"\x00" + bytes(rgb) * width # 每行前缀 filter byte 0
|
||||
idat = zlib.compress(row * height, 9)
|
||||
return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")
|
||||
|
||||
|
||||
def _hsv_rgb(i: int, total: int) -> tuple[int, int, int]:
|
||||
"""色相均匀铺开,出一组肉眼可区分的高饱和色(免得整屏一个色、看不出卡片边界)。"""
|
||||
h = 6.0 * i / total
|
||||
f = h - int(h)
|
||||
v, p, q, t = 235, 90, int(235 - 145 * f), int(90 + 145 * f)
|
||||
return [(v, t, p), (q, v, p), (p, v, t), (p, q, v), (t, p, v), (v, p, q)][int(h) % 6]
|
||||
|
||||
|
||||
def _write_images() -> list[str]:
|
||||
"""写 mock 头图,返回可直接塞进 raw.headUrl 的 URL 列表(不含缩放后缀)。"""
|
||||
_IMG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
urls: list[str] = []
|
||||
for i in range(_IMG_COUNT):
|
||||
name = f"mock_coupon_{i:02d}.png"
|
||||
# 落盘名带后缀,原因见模块 docstring
|
||||
(_IMG_DIR / f"{name}{FEED_THUMB_PARAM}").write_bytes(_solid_png(375, 375, _hsv_rgb(i, _IMG_COUNT)))
|
||||
urls.append(name)
|
||||
return urls
|
||||
|
||||
|
||||
def _clean(db, city_id: str | None) -> int:
|
||||
"""删掉本脚本造的 mock 行(可限定城市),返回删除条数。"""
|
||||
stmt = delete(MeituanCoupon).where(MeituanCoupon.product_view_sign.like(f"{MOCK_SIGN_PREFIX}%"))
|
||||
if city_id:
|
||||
stmt = stmt.where(MeituanCoupon.city_id == city_id)
|
||||
n = db.execute(stmt).rowcount or 0
|
||||
db.commit()
|
||||
for f in _IMG_DIR.glob(f"mock_coupon_*.png{FEED_THUMB_PARAM}"):
|
||||
f.unlink()
|
||||
return n
|
||||
|
||||
|
||||
def _build_row(i: int, rng: random.Random, city_id: str, img_urls: list[str], base_url: str) -> MeituanCoupon:
|
||||
to_store = rng.random() < 0.35 # 三成半到店,其余外卖
|
||||
platform = 2 if to_store else 1
|
||||
biz_line = rng.choice([1, 2]) if to_store else None
|
||||
source = "store_supply" if to_store else rng.choice(["search_waimai", "search_meishi"])
|
||||
|
||||
brand = rng.choice(_BRANDS)
|
||||
item = rng.choice(_DAODIAN_ITEMS if to_store else _WAIMAI_ITEMS)
|
||||
# 名字带序号:保证 dedup_key(brand|name|price) 唯一,不会被 DISTINCT ON 折叠掉,分页才铺得开
|
||||
name = f"{brand}{item}#{i:04d}"
|
||||
|
||||
sell_cents = rng.randrange(590, 8900, 10)
|
||||
orig_cents = int(sell_cents * rng.uniform(1.35, 2.6))
|
||||
sell, orig = f"{sell_cents / 100:.2f}", f"{orig_cents / 100:.2f}"
|
||||
|
||||
# 六成券佣金 ≥3%(智能推荐的门槛),其余低佣金:两个 tab 的结果集才有明显区别
|
||||
pct = round(rng.uniform(3.0, 8.5), 1) if rng.random() < 0.6 else round(rng.uniform(0.3, 2.9), 1)
|
||||
comm_cents = int(sell_cents * pct / 100)
|
||||
|
||||
sv_text, sv_num = rng.choice(_SALE_VOLUMES)
|
||||
head_url = f"{base_url}{settings.MEDIA_URL_PREFIX}/mock_coupon/{rng.choice(img_urls)}"
|
||||
poi_name = f"{brand}(mock{rng.randrange(1, 60):02d}店)"
|
||||
dist_km = round(rng.uniform(0.2, 8.0), 2) # 外卖侧单位是 km,到店是 m(见 CouponCard.from_raw)
|
||||
sign = f"{MOCK_SIGN_PREFIX}{i:06d}"
|
||||
|
||||
raw = {
|
||||
"couponPackDetail": {
|
||||
"productViewSign": sign,
|
||||
"skuViewId": f"{sign}-SKU",
|
||||
"platform": platform,
|
||||
"bizLine": biz_line,
|
||||
"name": name,
|
||||
"headUrl": head_url,
|
||||
"sellPrice": sell,
|
||||
"originalPrice": orig,
|
||||
"saleVolume": sv_text,
|
||||
"couponNum": rng.choice([1, 1, 1, 3, 5]),
|
||||
"productLabel": {
|
||||
"pricePowerLabel": {"historyPriceLabel": rng.choice(_PRICE_LABELS)},
|
||||
"productRankLabel": f"2小时北京销量榜第{rng.randrange(1, 30)}名" if rng.random() < 0.25 else None,
|
||||
"dianPingRankLabel": f"{rng.uniform(3.8, 4.9):.1f}分" if rng.random() < 0.5 else None,
|
||||
},
|
||||
},
|
||||
"brandInfo": {"brandName": brand, "brandLogoUrl": head_url},
|
||||
# commissionPercent 是「百分比 ×100」(140 → 1.4%),与 pull 脚本的换算保持一致
|
||||
"commissionInfo": {"commissionPercent": int(pct * 100), "commission": f"{comm_cents / 100:.2f}"},
|
||||
"deliverablePoiInfo": {"poiName": poi_name, "deliveryDistance": dist_km if platform == 1 else dist_km * 1000},
|
||||
"availablePoiInfo": {"availablePoiNum": rng.randrange(1, 200)},
|
||||
"couponValidTimeInfo": {"couponValidDay": rng.choice([7, 15, 30, 90])},
|
||||
}
|
||||
|
||||
return MeituanCoupon(
|
||||
source=source, platform=platform, biz_line=biz_line, city_id=city_id,
|
||||
product_view_sign=sign, sku_view_id=f"{sign}-SKU",
|
||||
name=name, brand_name=brand,
|
||||
sell_price_cents=sell_cents, original_price_cents=orig_cents,
|
||||
head_url=head_url, image_size=None, image_type="image/png",
|
||||
sale_volume=sv_text, sale_volume_num=sv_num,
|
||||
commission_percent=pct, commission_amount_cents=comm_cents,
|
||||
poi_name=poi_name, available_poi_num=raw["availablePoiInfo"]["availablePoiNum"],
|
||||
delivery_distance_m=dist_km * 1000,
|
||||
dedup_key=hashlib.md5(f"{brand}|{name}|{sell_cents}".encode()).hexdigest(),
|
||||
raw=raw,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description="往 meituan_coupon 灌 mock 券(本地无凭证时用)")
|
||||
ap.add_argument("--city-id", default=DEFAULT_CITY_ID, help=f"美团 cityId,默认北京 {DEFAULT_CITY_ID}")
|
||||
ap.add_argument("--count", type=int, default=400, help="造多少条(默认 400,约六成过智能推荐的佣金门槛)")
|
||||
ap.add_argument("--base-url", default="http://127.0.0.1:8770",
|
||||
help="头图用的后端地址,须为测试机可达(默认 http://127.0.0.1:8770,对齐 local.properties)")
|
||||
ap.add_argument("--clean-only", action="store_true", help="只清 mock 数据,不重建")
|
||||
ap.add_argument("--seed", type=int, default=20260723, help="随机种子(固定则每次造出同一批)")
|
||||
args = ap.parse_args()
|
||||
|
||||
db = SessionLocal()
|
||||
try:
|
||||
removed = _clean(db, None if args.clean_only else args.city_id)
|
||||
print(f"清理旧 mock: {removed} 条")
|
||||
if args.clean_only:
|
||||
return
|
||||
|
||||
rng = random.Random(args.seed)
|
||||
img_urls = _write_images()
|
||||
print(f"头图: {_IMG_COUNT} 张 -> {_IMG_DIR}")
|
||||
|
||||
rows = [_build_row(i, rng, args.city_id, img_urls, args.base_url.rstrip("/"))
|
||||
for i in range(args.count)]
|
||||
db.add_all(rows)
|
||||
db.commit()
|
||||
|
||||
rec = sum(1 for r in rows if (r.commission_percent or 0) >= 3.0)
|
||||
print(f"入库: {len(rows)} 条 city_id={args.city_id}")
|
||||
print(f" 智能推荐(佣金≥3%): {rec} 条 ≈ {(rec + 19) // 20} 页")
|
||||
print(f" 销量最高(有销量): {len(rows)} 条 ≈ {(len(rows) + 19) // 20} 页")
|
||||
print("\n测试机的模拟定位记得设成对应城市,否则 city_id 对不上仍然是空。")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user