Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 93a942dff1 | |||
| 34798ed7b5 |
@@ -1,26 +0,0 @@
|
||||
"""merge comparison platforms + fail_reason heads
|
||||
|
||||
Revision ID: 6d2309208549
|
||||
Revises: comparison_platforms_col, comparison_record_fail_reason
|
||||
Create Date: 2026-07-29 01:48:41.868083
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = '6d2309208549'
|
||||
down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
|
||||
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,47 +0,0 @@
|
||||
"""add platforms unified array column to comparison_record
|
||||
|
||||
展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
|
||||
status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
|
||||
纯新增列, 老记录为空 → 前端回退老 comparison_results。
|
||||
|
||||
Revision ID: comparison_platforms_col
|
||||
Revises: user_manual_risk_fields
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "comparison_platforms_col"
|
||||
down_revision: str | None = "user_manual_risk_fields"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
|
||||
# `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
|
||||
# NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
|
||||
# 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
|
||||
bind = op.get_bind()
|
||||
cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
|
||||
if "platforms" in cols:
|
||||
return
|
||||
with op.batch_alter_table("comparison_record") as batch_op:
|
||||
batch_op.add_column(
|
||||
sa.Column(
|
||||
"platforms", _JSON, nullable=False,
|
||||
server_default=sa.text("'[]'"),
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("comparison_record") as batch_op:
|
||||
batch_op.drop_column("platforms")
|
||||
@@ -1,31 +0,0 @@
|
||||
"""guide video play count is independent for coupon and comparison
|
||||
|
||||
Revision ID: guide_video_scene_unique
|
||||
Revises: 6d2309208549
|
||||
"""
|
||||
from alembic import op
|
||||
|
||||
revision = "guide_video_scene_unique"
|
||||
down_revision = "6d2309208549"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.drop_index("uq_guide_video_play_user_seq", table_name="guide_video_play")
|
||||
op.create_index(
|
||||
"uq_guide_video_play_user_scene_seq",
|
||||
"guide_video_play",
|
||||
["user_id", "scene", "seq"],
|
||||
unique=True,
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_index("uq_guide_video_play_user_scene_seq", table_name="guide_video_play")
|
||||
op.create_index(
|
||||
"uq_guide_video_play_user_seq",
|
||||
"guide_video_play",
|
||||
["user_id", "seq"],
|
||||
unique=True,
|
||||
)
|
||||
@@ -1,164 +0,0 @@
|
||||
"""guide video prepare/start/ten-circle reward state machine
|
||||
|
||||
Revision ID: guide_video_ten_circle_v2
|
||||
Revises: guide_video_scene_unique
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision = "guide_video_ten_circle_v2"
|
||||
down_revision = "guide_video_scene_unique"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def _backfill_configs(connection) -> None:
|
||||
from app.core import media
|
||||
from app.core.config import settings
|
||||
|
||||
for key in ("coupon_guide_video", "comparison_guide_video"):
|
||||
row = connection.execute(
|
||||
sa.text("SELECT value FROM app_config WHERE key = :key"), {"key": key}
|
||||
).first()
|
||||
if not row:
|
||||
continue
|
||||
raw = row[0]
|
||||
value = json.loads(raw) if isinstance(raw, str) else dict(raw or {})
|
||||
if value.get("guide_video_v2_migrated") is True:
|
||||
continue
|
||||
maximum = int(value.get("max_plays", 3) or 0)
|
||||
reward = int(value.get("reward_coin", 100) or 0)
|
||||
video_url = str(value.get("video_url") or "")
|
||||
error = None
|
||||
analysis = None
|
||||
if not 1 <= maximum <= 50:
|
||||
error = "旧播放次数不符合 1~50 约束"
|
||||
elif not 10 <= reward <= 10_000 or reward % 10:
|
||||
error = "旧金币总价不符合 10~10000 且为 10 的倍数约束"
|
||||
elif not video_url:
|
||||
error = "尚未上传视频"
|
||||
else:
|
||||
prefix = f"{settings.MEDIA_URL_PREFIX}/guide_video/"
|
||||
if not video_url.startswith(prefix):
|
||||
error = "旧视频不是服务端托管文件,请重新上传"
|
||||
else:
|
||||
path = Path(settings.MEDIA_ROOT) / "guide_video" / video_url[len(prefix):]
|
||||
if not path.is_file():
|
||||
error = "旧视频文件缺失,请重新上传"
|
||||
else:
|
||||
try:
|
||||
analysis = media._probe_guide_video(path)
|
||||
except media.MediaError as exc:
|
||||
error = str(exc)
|
||||
value["config_version"] = int(value.get("config_version", 0) or 0) + 1
|
||||
value["guide_video_v2_migrated"] = True
|
||||
if analysis:
|
||||
value.update(analysis)
|
||||
else:
|
||||
value.update(
|
||||
enabled=False,
|
||||
duration_ms=None,
|
||||
video_codec=None,
|
||||
audio_codec=None,
|
||||
analysis_status="invalid" if video_url else "missing",
|
||||
analysis_error=error,
|
||||
)
|
||||
statement = sa.text(
|
||||
"UPDATE app_config SET value = :value WHERE key = :key"
|
||||
).bindparams(sa.bindparam("value", type_=sa.JSON()))
|
||||
connection.execute(statement, {"key": key, "value": value})
|
||||
|
||||
|
||||
def _seal_legacy_plays(connection) -> None:
|
||||
"""旧整笔发奖 token 永久封口;旧未发奖会话终止但仍保留计次行。"""
|
||||
connection.execute(
|
||||
sa.text(
|
||||
"""
|
||||
UPDATE guide_video_play
|
||||
SET status = CASE WHEN status = 'granted' THEN 'legacy_completed' ELSE 'legacy_closed' END,
|
||||
settled_circles = CASE WHEN status = 'granted' THEN 10 ELSE 0 END,
|
||||
prepared_at = started_at,
|
||||
expires_at = started_at
|
||||
WHERE status IN ('granted', 'playing')
|
||||
"""
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _prepare_downgrade(connection) -> None:
|
||||
"""清除未起播计划,并把所有保留 token 封成旧版不可再领奖的 granted。"""
|
||||
connection.execute(
|
||||
sa.text("DELETE FROM guide_video_play WHERE status = 'prepared'")
|
||||
)
|
||||
connection.execute(
|
||||
sa.text(
|
||||
"""
|
||||
UPDATE guide_video_play
|
||||
SET status = 'granted',
|
||||
completed = CASE WHEN settled_circles >= 10 THEN 1 ELSE completed END,
|
||||
granted_at = CASE
|
||||
WHEN settled_circles >= 10 AND granted_at IS NULL THEN started_at
|
||||
ELSE granted_at
|
||||
END
|
||||
WHERE status IN (
|
||||
'started', 'completed', 'legacy_completed', 'legacy_closed'
|
||||
)
|
||||
"""
|
||||
)
|
||||
)
|
||||
# 防御未知/中间状态:宁可删除,也绝不让旧应用把它当 playing 整笔发奖。
|
||||
connection.execute(
|
||||
sa.text(
|
||||
"""
|
||||
DELETE FROM guide_video_play
|
||||
WHERE seq IS NULL OR started_at IS NULL OR status != 'granted'
|
||||
"""
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table("guide_video_play") as batch:
|
||||
batch.alter_column("seq", existing_type=sa.Integer(), nullable=True)
|
||||
batch.alter_column("started_at", existing_type=sa.DateTime(timezone=True), nullable=True)
|
||||
batch.add_column(sa.Column("duration_ms", sa.Integer(), nullable=False, server_default="0"))
|
||||
batch.add_column(sa.Column("config_version", sa.Integer(), nullable=False, server_default="0"))
|
||||
batch.add_column(sa.Column("settled_circles", sa.Integer(), nullable=False, server_default="0"))
|
||||
batch.add_column(
|
||||
sa.Column("prepared_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now())
|
||||
)
|
||||
batch.add_column(sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True))
|
||||
batch.alter_column("status", existing_type=sa.String(length=16), type_=sa.String(length=24))
|
||||
op.create_index("ix_guide_video_play_expires_at", "guide_video_play", ["expires_at"])
|
||||
op.create_index(
|
||||
"ux_coin_transaction_guide_video_ref",
|
||||
"coin_transaction",
|
||||
["user_id", "biz_type", "ref_id"],
|
||||
unique=True,
|
||||
sqlite_where=sa.text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
|
||||
postgresql_where=sa.text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
|
||||
)
|
||||
connection = op.get_bind()
|
||||
_seal_legacy_plays(connection)
|
||||
_backfill_configs(connection)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
_prepare_downgrade(op.get_bind())
|
||||
op.drop_index("ux_coin_transaction_guide_video_ref", table_name="coin_transaction")
|
||||
op.drop_index("ix_guide_video_play_expires_at", table_name="guide_video_play")
|
||||
with op.batch_alter_table("guide_video_play") as batch:
|
||||
batch.alter_column("status", existing_type=sa.String(length=24), type_=sa.String(length=16))
|
||||
batch.drop_column("expires_at")
|
||||
batch.drop_column("prepared_at")
|
||||
batch.drop_column("settled_circles")
|
||||
batch.drop_column("config_version")
|
||||
batch.drop_column("duration_ms")
|
||||
batch.alter_column("started_at", existing_type=sa.DateTime(timezone=True), nullable=False)
|
||||
batch.alter_column("seq", existing_type=sa.Integer(), nullable=False)
|
||||
+3
-3
@@ -1,6 +1,6 @@
|
||||
"""Admin 后台 FastAPI app(独立进程)。
|
||||
|
||||
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8773
|
||||
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
|
||||
复用 App 的 DB/models/repositories/integrations;鉴权独立(admin JWT,见 app/admin/security.py)。
|
||||
现有 app.main:app 不 import 本模块,两进程互不影响。
|
||||
"""
|
||||
@@ -71,8 +71,8 @@ admin_app = FastAPI(
|
||||
|
||||
# admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
|
||||
_dev_origins = [
|
||||
"http://localhost:3002",
|
||||
"http://127.0.0.1:3002",
|
||||
"http://localhost:3001",
|
||||
"http://127.0.0.1:3001",
|
||||
"http://localhost:3000",
|
||||
"http://127.0.0.1:3000",
|
||||
]
|
||||
|
||||
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Literal
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
|
||||
|
||||
@@ -27,16 +27,14 @@ router = APIRouter(
|
||||
)
|
||||
|
||||
|
||||
GuideScene = Literal["coupon", "comparison"]
|
||||
|
||||
|
||||
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
|
||||
return GuideVideoConfigOut(scene=scene, **guide_video.get_config(db, scene))
|
||||
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, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
|
||||
return _out(db, scene)
|
||||
def get_config(db: AdminDb) -> GuideVideoConfigOut:
|
||||
return _out(db)
|
||||
|
||||
|
||||
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
|
||||
@@ -45,26 +43,21 @@ def update_config(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
try:
|
||||
before, after = guide_video.update_config(
|
||||
db,
|
||||
enabled=body.enabled,
|
||||
max_plays=body.max_plays,
|
||||
reward_coin=body.reward_coin,
|
||||
scene=scene,
|
||||
admin_id=admin.id,
|
||||
commit=False,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
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={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
detail={"before": before, "after": after}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
return _out(db, scene)
|
||||
return _out(db)
|
||||
|
||||
|
||||
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
|
||||
@@ -72,40 +65,23 @@ async def upload_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
file: Annotated[UploadFile, File()],
|
||||
scene: GuideScene = "coupon",
|
||||
file: UploadFile = File(...),
|
||||
) -> GuideVideoConfigOut:
|
||||
data = await file.read()
|
||||
url: str | None = None
|
||||
try:
|
||||
url, analysis = media.save_guide_video(data)
|
||||
url = media.save_guide_video(data)
|
||||
except media.MediaError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e)) from e
|
||||
try:
|
||||
before, after = guide_video.set_video(
|
||||
db, url, analysis=analysis, scene=scene, admin_id=admin.id, commit=False
|
||||
)
|
||||
write_audit(
|
||||
db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
|
||||
detail={
|
||||
"scene": scene,
|
||||
"before": before.get("video_url"),
|
||||
"after": url,
|
||||
"bytes": len(data),
|
||||
"duration_ms": analysis["duration_ms"],
|
||||
"video_codec": analysis["video_codec"],
|
||||
"audio_codec": analysis["audio_codec"],
|
||||
},
|
||||
ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
except Exception:
|
||||
db.rollback()
|
||||
media.delete_guide_video(url)
|
||||
raise
|
||||
# 旧文件仍可能被 prepare/已起播快照引用,不能在这里立即删除。
|
||||
guide_video.cleanup_old_videos(db)
|
||||
return _out(db, scene)
|
||||
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="移除新手引导视频(带审计)")
|
||||
@@ -113,17 +89,13 @@ def delete_video(
|
||||
request: Request,
|
||||
admin: Annotated[AdminUser, Depends(require_role("operator"))],
|
||||
db: AdminDb,
|
||||
scene: GuideScene = "coupon",
|
||||
) -> GuideVideoConfigOut:
|
||||
"""移除后 /guide-video/start 一律返回 should_play=false,领券浮层回到「只放广告」。"""
|
||||
before, after = guide_video.set_video(
|
||||
db, None, scene=scene, admin_id=admin.id, commit=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={"scene": scene, "before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
detail={"before": before.get("video_url")}, ip=get_client_ip(request), commit=False,
|
||||
)
|
||||
db.commit()
|
||||
# 旧文件由保护期清理任务处理,避免破坏有效计划与已起播快照。
|
||||
guide_video.cleanup_old_videos(db)
|
||||
return _out(db, scene)
|
||||
media.delete_guide_video(before.get("video_url"))
|
||||
return _out(db)
|
||||
|
||||
@@ -1,44 +1,25 @@
|
||||
"""后台引导视频配置协议。"""
|
||||
"""admin 新手引导视频配置 schemas(开关 / 视频地址 / 前几次 / 每次金币)。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.repositories.guide_video import (
|
||||
MAX_PLAYS_LIMIT,
|
||||
MIN_PLAYS,
|
||||
MIN_REWARD_COIN,
|
||||
REWARD_COIN_LIMIT,
|
||||
)
|
||||
from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
|
||||
|
||||
|
||||
class GuideVideoConfigOut(BaseModel):
|
||||
scene: str
|
||||
enabled: bool
|
||||
video_url: str | None = None
|
||||
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
|
||||
max_plays: int
|
||||
reward_coin: int
|
||||
duration_ms: int | None = None
|
||||
circle_count: int = 10
|
||||
circle_duration_ms: float | None = None
|
||||
reward_per_circle: int
|
||||
video_codec: str | None = None
|
||||
audio_codec: str | None = None
|
||||
analysis_status: str
|
||||
analysis_error: str | None = None
|
||||
config_version: int
|
||||
updated_at: str | None = None
|
||||
# 只读统计,后台展示用:已有多少次播放、其中已发币多少次。
|
||||
total_plays: int = 0
|
||||
granted_plays: int = 0
|
||||
|
||||
|
||||
class GuideVideoConfigUpdate(BaseModel):
|
||||
enabled: bool | None = None
|
||||
max_plays: int | None = Field(default=None, ge=MIN_PLAYS, le=MAX_PLAYS_LIMIT)
|
||||
reward_coin: int | None = Field(
|
||||
default=None, ge=MIN_REWARD_COIN, le=REWARD_COIN_LIMIT
|
||||
)
|
||||
"""部分更新:只改传入(非 None)字段。视频文件走 /video 上传接口。"""
|
||||
|
||||
@field_validator("reward_coin")
|
||||
@classmethod
|
||||
def reward_must_be_multiple_of_ten(cls, value: int | None) -> int | None:
|
||||
if value is not None and value % 10:
|
||||
raise ValueError("金币总价必须是 10 的倍数")
|
||||
return value
|
||||
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)
|
||||
|
||||
+27
-42
@@ -1,16 +1,22 @@
|
||||
"""引导视频 prepare/start/reward 客户端 API。"""
|
||||
"""新手引导视频(领券等候浮层前 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, HTTPException
|
||||
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 (
|
||||
GuideVideoPrepareIn,
|
||||
GuideVideoPrepareOut,
|
||||
GuideVideoRewardIn,
|
||||
GuideVideoRewardOut,
|
||||
GuideVideoStartIn,
|
||||
@@ -18,41 +24,25 @@ from app.schemas.guide_video import (
|
||||
)
|
||||
|
||||
logger = logging.getLogger("shagua.guide_video")
|
||||
|
||||
router = APIRouter(prefix="/api/v1/guide-video", tags=["guide-video"])
|
||||
|
||||
|
||||
@router.post(
|
||||
"/prepare",
|
||||
response_model=GuideVideoPrepareOut,
|
||||
dependencies=[Depends(rate_limit(60, 60, "guide-video-prepare"))],
|
||||
)
|
||||
def prepare(
|
||||
payload: GuideVideoPrepareIn, user: CurrentUser, db: DbSession
|
||||
) -> GuideVideoPrepareOut:
|
||||
result = crud_guide.prepare_play(db, user.id, scene=payload.scene)
|
||||
logger.info(
|
||||
"guide video prepare user_id=%d scene=%s should_play=%s reason=%s",
|
||||
user.id, payload.scene, result["should_play"], result["reason"],
|
||||
)
|
||||
return GuideVideoPrepareOut(**result)
|
||||
|
||||
|
||||
@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:
|
||||
try:
|
||||
result = crud_guide.start_play(db, user.id, play_token=payload.play_token)
|
||||
except crud_guide.PlayStateError as exc:
|
||||
status_code = 404 if exc.code == "play_not_found" else 409
|
||||
raise HTTPException(status_code=status_code, detail=exc.detail()) from exc
|
||||
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 token=%s status=%s seq=%d",
|
||||
user.id, payload.play_token[:12], result["status"], result["seq"],
|
||||
"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)
|
||||
|
||||
@@ -60,20 +50,15 @@ def start(
|
||||
@router.post(
|
||||
"/reward",
|
||||
response_model=GuideVideoRewardOut,
|
||||
dependencies=[Depends(rate_limit(120, 60, "guide-video-reward"))],
|
||||
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_circle(
|
||||
db,
|
||||
user.id,
|
||||
play_token=payload.play_token,
|
||||
circle=payload.circle,
|
||||
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 circle=%d status=%s granted=%s",
|
||||
user.id, payload.play_token[:12], payload.circle,
|
||||
result["status"], result["granted"],
|
||||
"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)
|
||||
|
||||
@@ -436,7 +436,6 @@ class Settings(BaseSettings):
|
||||
# 运营后台上传的新手引导视频上限。视频比图片大一个量级,单独一档;
|
||||
# ⚠️ 改大时同步放宽网关 client_max_body_size(实测 QA 4MiB / prod 32MiB),否则 nginx 先挡下。
|
||||
GUIDE_VIDEO_MAX_BYTES: int = 100 * 1024 * 1024 # 引导视频最大 100MB
|
||||
FFPROBE_BINARY: str = "ffprobe"
|
||||
|
||||
# ===== 邀请好友 =====
|
||||
# 分享落地页(二维码 / 分享链接指向这里;扫码 → 落地页 → 引导浏览器下载 APK)。
|
||||
|
||||
+9
-76
@@ -10,13 +10,8 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import subprocess
|
||||
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
@@ -86,91 +81,29 @@ def _sniff_video_ext(data: bytes) -> str | None:
|
||||
"""按魔数判定视频类型,返回扩展名;非支持类型返回 None。
|
||||
|
||||
只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
|
||||
支持 H.264 或 HEVC/H.265 视频及 AAC 音频;放开 mkv/avi 只会让端上放不出来,
|
||||
不如在入口就挡掉。
|
||||
都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
|
||||
"""
|
||||
if len(data) >= 12 and data[4:8] == b"ftyp":
|
||||
return ".mp4"
|
||||
return None
|
||||
|
||||
|
||||
def _probe_guide_video(path: Path) -> dict[str, Any]:
|
||||
"""用 ffprobe 读取服务端可信的时长/编码信息。"""
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
[
|
||||
settings.FFPROBE_BINARY,
|
||||
"-v", "error",
|
||||
"-show_format",
|
||||
"-show_streams",
|
||||
"-of", "json",
|
||||
str(path),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
check=False,
|
||||
)
|
||||
except FileNotFoundError as exc:
|
||||
raise MediaError("服务器未安装 ffprobe,暂时无法分析视频") from exc
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise MediaError("ffprobe 分析视频失败") from exc
|
||||
if proc.returncode != 0:
|
||||
raise MediaError("无法分析视频,请确认文件是有效的 MP4")
|
||||
try:
|
||||
payload = json.loads(proc.stdout)
|
||||
streams = payload.get("streams") or []
|
||||
video = next(s for s in streams if s.get("codec_type") == "video")
|
||||
audio = next((s for s in streams if s.get("codec_type") == "audio"), None)
|
||||
raw_duration = (
|
||||
(payload.get("format") or {}).get("duration")
|
||||
or video.get("duration")
|
||||
)
|
||||
duration_ms = int(
|
||||
(Decimal(str(raw_duration)) * 1000).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
|
||||
)
|
||||
except (StopIteration, TypeError, ValueError, InvalidOperation) as exc:
|
||||
raise MediaError("视频缺少可识别的视频轨或时长") from exc
|
||||
video_codec = str(video.get("codec_name") or "").lower()
|
||||
audio_codec = str(audio.get("codec_name") or "").lower() if audio else None
|
||||
if not 30_000 <= duration_ms <= 180_000:
|
||||
raise MediaError("视频时长必须在 30~180 秒之间")
|
||||
if video_codec not in ("h264", "hevc", "h265"):
|
||||
raise MediaError("视频编码必须为 H.264 或 HEVC/H.265")
|
||||
if audio_codec not in (None, "aac"):
|
||||
raise MediaError("音频编码必须为 AAC")
|
||||
return {
|
||||
"duration_ms": duration_ms,
|
||||
"video_codec": video_codec,
|
||||
"audio_codec": audio_codec,
|
||||
"analysis_status": "valid",
|
||||
"analysis_error": None,
|
||||
}
|
||||
def save_guide_video(data: bytes) -> str:
|
||||
"""保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。
|
||||
|
||||
|
||||
def save_guide_video(data: bytes) -> tuple[str, dict[str, Any]]:
|
||||
"""临时落盘、ffprobe 校验后原子发布,返回 URL 与分析结果。"""
|
||||
与图片分开一套校验:体积上限走 [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 或 HEVC/H.265 编码)")
|
||||
raise MediaError("仅支持 MP4 视频(H.264 编码)")
|
||||
|
||||
directory = _media_dir("guide_video")
|
||||
stem = f"guide_{secrets.token_hex(8)}"
|
||||
temp_path = directory / f".{stem}.upload"
|
||||
final_path = directory / f"{stem}.mp4"
|
||||
try:
|
||||
temp_path.write_bytes(data)
|
||||
analysis = _probe_guide_video(temp_path)
|
||||
os.replace(temp_path, final_path)
|
||||
except Exception:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
final_path.unlink(missing_ok=True)
|
||||
raise
|
||||
return f"{settings.MEDIA_URL_PREFIX}/guide_video/{final_path.name}", analysis
|
||||
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:
|
||||
|
||||
@@ -113,10 +113,6 @@ class ComparisonRecord(Base):
|
||||
items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
|
||||
# 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
|
||||
comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
|
||||
# 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
|
||||
# status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
|
||||
# + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
|
||||
platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
|
||||
# 目标平台未找到、跳过的菜名
|
||||
skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
|
||||
# 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
|
||||
|
||||
+41
-25
@@ -1,4 +1,20 @@
|
||||
"""引导视频待播放计划、起播快照与逐圈结算状态。"""
|
||||
"""新手引导视频播放记录(领券浮层前 N 次用它替代广告)。
|
||||
|
||||
产品规则(2026-07 拍板):新用户点「一键自动领取」后的等候浮层,**前 3 次**不放广告,
|
||||
改放运营后台上传的引导视频;每次固定 120 金币,中途关闭也算看完照发。
|
||||
|
||||
口径:
|
||||
- **计次按账号**(user_id),与设备无关 —— 换设备不重新送 3 次。
|
||||
- **开播即计数**:客户端每次要展示浮层时调 `/api/v1/guide-video/start`,服务端当场
|
||||
写一行(status='playing')并返回 play_token;`COUNT(*)` 即已用次数。用户中途 kill
|
||||
App 也算用掉一次(产品选定口径,防反复进出刷金币)。
|
||||
- **发币幂等**靠 play_token 定位 + `status='playing'` 条件更新:并发两次上报只有一次
|
||||
改到行(另一次 rowcount=0),所以只发一次币。光有 play_token 唯一键挡不住 —— 发币走的是
|
||||
UPDATE,不 INSERT,撞不到任何唯一键。
|
||||
- **次数上限**靠 (user_id, seq) 唯一键兜底,防并发 /start 绕过 COUNT 判定(见下)。
|
||||
|
||||
与广告收益(ad_feed_reward_record)彻底分离:引导视频不是广告,不该进广告收益报表。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
@@ -10,48 +26,48 @@ 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"),
|
||||
Index(
|
||||
"uq_guide_video_play_user_scene_seq",
|
||||
"user_id",
|
||||
"scene",
|
||||
"seq",
|
||||
unique=True,
|
||||
),
|
||||
# 次数上限的**硬约束**:start_play 是无锁 check-then-insert(读 COUNT 算 seq 再插),
|
||||
# N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
|
||||
# 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
|
||||
# start_play 捕获 IntegrityError 降级成"这次不放视频"。
|
||||
# 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
|
||||
# 要整表重建),autogenerate 才不会每次报一条假 diff。
|
||||
Index("uq_guide_video_play_user_seq", "user_id", "seq", unique=True),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
user_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("user.id"), index=True, nullable=False
|
||||
)
|
||||
# 服务端生成下发给客户端的幂等键(uuid hex)。
|
||||
play_token: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
# 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
|
||||
scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
|
||||
# prepare 不占次数,seq=NULL;start 成功才写入 1-based seq。
|
||||
seq: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
# 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
|
||||
# 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
|
||||
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
|
||||
# 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
|
||||
video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
|
||||
# 实发金币;未发时 0。
|
||||
coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
duration_ms: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
config_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
settled_circles: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
# prepared / started / completed / legacy_completed / legacy_closed
|
||||
status: Mapped[str] = mapped_column(String(24), nullable=False, default="prepared")
|
||||
# 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)
|
||||
|
||||
prepared_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), nullable=False
|
||||
)
|
||||
expires_at: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), index=True, nullable=True
|
||||
)
|
||||
started_at: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), index=True, nullable=True
|
||||
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} scene={self.scene} seq={self.seq} "
|
||||
f"{self.status} circles={self.settled_circles}>"
|
||||
f"<GuideVideoPlay user={self.user_id} seq={self.seq} "
|
||||
f"{self.status} coin={self.coin}>"
|
||||
)
|
||||
|
||||
@@ -60,15 +60,6 @@ class CoinTransaction(Base):
|
||||
sqlite_where=text("biz_type LIKE 'task%' AND ref_id IS NOT NULL"),
|
||||
postgresql_where=text("biz_type LIKE 'task%' AND ref_id IS NOT NULL"),
|
||||
),
|
||||
Index(
|
||||
"ux_coin_transaction_guide_video_ref",
|
||||
"user_id",
|
||||
"biz_type",
|
||||
"ref_id",
|
||||
unique=True,
|
||||
sqlite_where=text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
|
||||
postgresql_where=text("biz_type = 'guide_video' AND ref_id IS NOT NULL"),
|
||||
),
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
||||
|
||||
+12
-101
@@ -164,9 +164,8 @@ def _derive(payload: ComparisonRecordIn) -> dict:
|
||||
|
||||
is_source_best = best.is_source if best is not None else None
|
||||
|
||||
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
|
||||
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
|
||||
status = payload.record_status or payload.status
|
||||
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
|
||||
status = payload.status
|
||||
if status is None:
|
||||
has_valid_target = any(
|
||||
(not r.is_source) and r.price is not None for r in results
|
||||
@@ -198,35 +197,16 @@ def upsert_record(
|
||||
灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
|
||||
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
|
||||
"""
|
||||
# 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
|
||||
# _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
|
||||
if payload.platforms:
|
||||
derived = _derive_from_platforms(payload.platforms, payload.record_status)
|
||||
# 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
|
||||
derived["fail_reason"] = (
|
||||
_derive_fail_display(payload.information, payload.platform_results or {})
|
||||
if derived["status"] == "failed"
|
||||
else None
|
||||
)
|
||||
# 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
|
||||
# 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
|
||||
# 与 _derive_from_platforms 同名键撞键 TypeError)。
|
||||
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
|
||||
if not derived.get(_k):
|
||||
derived[_k] = getattr(payload, _k)
|
||||
else:
|
||||
derived = _derive(payload)
|
||||
# _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
|
||||
# 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
|
||||
for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
|
||||
derived[_k] = getattr(payload, _k)
|
||||
derived = _derive(payload)
|
||||
items = [it.model_dump(exclude_none=True) for it in payload.items]
|
||||
fields = dict(
|
||||
device_id=payload.device_id,
|
||||
business_type=payload.business_type,
|
||||
store_name=payload.store_name,
|
||||
product_names=_product_names_from_items(items),
|
||||
# store_name / source_platform_id / source_platform_name / source_package 统一由
|
||||
# derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
|
||||
source_platform_id=payload.source_platform_id,
|
||||
source_platform_name=payload.source_platform_name,
|
||||
source_package=payload.source_package,
|
||||
information=payload.information,
|
||||
best_deeplink=payload.best_deeplink,
|
||||
trace_url=payload.trace_url,
|
||||
@@ -234,7 +214,6 @@ def upsert_record(
|
||||
skipped_dish_count=payload.skipped_dish_count,
|
||||
items=items,
|
||||
comparison_results=[r.model_dump() for r in payload.comparison_results],
|
||||
platforms=list(payload.platforms or []),
|
||||
skipped_dish_names=list(payload.skipped_dish_names),
|
||||
# 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
|
||||
device_model=payload.device_model,
|
||||
@@ -307,8 +286,7 @@ def upsert_record(
|
||||
|
||||
|
||||
def _derive_from_results(
|
||||
results: list[dict], platform_results: dict | None = None,
|
||||
record_status: str | None = None,
|
||||
results: list[dict], platform_results: dict | None = None
|
||||
) -> dict:
|
||||
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
|
||||
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
|
||||
@@ -356,53 +334,7 @@ def _derive_from_results(
|
||||
"saved_amount_cents": saved_amount_cents,
|
||||
"is_source_best": best.get("is_source") if best else None,
|
||||
"store_name": (src_row or {}).get("store_name") or None,
|
||||
# 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
|
||||
# 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
|
||||
# 回退老的 success/failed 二态派生, 向后兼容。
|
||||
"status": record_status or ("success" if has_valid_target else "failed"),
|
||||
}
|
||||
|
||||
|
||||
def _derive_from_platforms(
|
||||
platforms: list, record_status: str | None = None,
|
||||
) -> dict:
|
||||
"""从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
|
||||
|
||||
best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
|
||||
platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
|
||||
消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
|
||||
platforms 时调用方回退 _derive_from_results(向后兼容)。"""
|
||||
rows = [p for p in (platforms or []) if isinstance(p, dict)]
|
||||
src = next((p for p in rows if p.get("role") == "source"), None)
|
||||
best = next((p for p in rows if p.get("is_best")), None)
|
||||
source_price_cents = _yuan_to_cents(src.get("price")) if src else None
|
||||
best_price_cents = _yuan_to_cents(best.get("price")) if best else None
|
||||
saved_amount_cents = None
|
||||
if source_price_cents is not None and best_price_cents is not None:
|
||||
saved_amount_cents = source_price_cents - best_price_cents
|
||||
has_valid_target = any(
|
||||
p.get("role") != "source" and p.get("price") is not None for p in rows
|
||||
)
|
||||
# store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
|
||||
# _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
|
||||
# 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
|
||||
store_name = (
|
||||
(src or {}).get("store_name")
|
||||
or (best or {}).get("store_name")
|
||||
or next((p.get("store_name") for p in rows if p.get("store_name")), None)
|
||||
)
|
||||
return {
|
||||
"source_platform_id": (src or {}).get("platform_id"),
|
||||
"source_platform_name": (src or {}).get("platform_name"),
|
||||
"source_package": (src or {}).get("package"),
|
||||
"source_price_cents": source_price_cents,
|
||||
"best_platform_id": (best or {}).get("platform_id"),
|
||||
"best_platform_name": (best or {}).get("platform_name"),
|
||||
"best_price_cents": best_price_cents,
|
||||
"saved_amount_cents": saved_amount_cents,
|
||||
"is_source_best": (best.get("role") == "source") if best else None,
|
||||
"store_name": store_name or None,
|
||||
"status": record_status or ("success" if has_valid_target else "failed"),
|
||||
"status": "success" if has_valid_target else "failed",
|
||||
}
|
||||
|
||||
|
||||
@@ -561,29 +493,7 @@ def harvest_done(
|
||||
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
|
||||
行不存在(理论上帧0已建;防御)则新建。"""
|
||||
results = done_params.get("comparison_results") or []
|
||||
# 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
|
||||
# record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
|
||||
platforms = done_params.get("platforms") or []
|
||||
record_status = done_params.get("record_status")
|
||||
# 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
|
||||
# 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
|
||||
# platforms 时回退从 comparison_results 派生(向后兼容)。
|
||||
if platforms:
|
||||
derived = _derive_from_platforms(platforms, record_status)
|
||||
# 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
|
||||
_item_row = (
|
||||
next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
|
||||
or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
|
||||
or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
|
||||
)
|
||||
items = (_item_row or {}).get("items") or []
|
||||
else:
|
||||
derived = _derive_from_results(
|
||||
results, done_params.get("platform_results"), record_status
|
||||
)
|
||||
# pricebot 已把源单菜品塞进 comparison_results[源行].items
|
||||
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
|
||||
# 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
|
||||
derived = _derive_from_results(results, done_params.get("platform_results"))
|
||||
fail_reason = (
|
||||
_derive_fail_display(
|
||||
done_params.get("information"), done_params.get("platform_results")
|
||||
@@ -591,6 +501,8 @@ def harvest_done(
|
||||
if derived["status"] == "failed"
|
||||
else None
|
||||
)
|
||||
# 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
|
||||
items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
|
||||
fields = dict(
|
||||
business_type=business_type or "food",
|
||||
information=done_params.get("information") or None,
|
||||
@@ -602,7 +514,6 @@ def harvest_done(
|
||||
skipped_dish_count=done_params.get("skipped_dish_count"),
|
||||
skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
|
||||
comparison_results=results,
|
||||
platforms=platforms,
|
||||
items=items,
|
||||
product_names=_product_names_from_items(items),
|
||||
raw_payload=done_params,
|
||||
|
||||
+184
-390
@@ -1,130 +1,95 @@
|
||||
"""双场景引导视频配置与 prepare/start/reward 状态机。"""
|
||||
"""新手引导视频:运营配置读写 + 播放计次 + 发币。
|
||||
|
||||
**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 JSON 存进通用 app_config 表
|
||||
(key=coupon_guide_video),写法完全对齐 feedback_qr —— 不进 CONFIG_DEFS,所以不会污染
|
||||
系统配置页的通用列表,由本模块独占维护。
|
||||
|
||||
**计次**按账号(user_id)、**开播即计数**:客户端每次要展示领券等候浮层时调
|
||||
`/api/v1/guide-video/start`,命中则当场写一行 guide_video_play(status='playing')。
|
||||
已用次数 = 该账号的行数,达到 max_plays 后不再下发,客户端改放广告(原逻辑)。
|
||||
COUNT 判定本身无锁,真正卡住次数上限的是 (user_id, seq) 唯一键:并发 /start 只能成一个。
|
||||
|
||||
**发币**幂等键是 play_token,落地方式是 `status='playing' → 'granted'` 的**条件更新**:
|
||||
同一次播放重复上报只入账一次(网络重试 / 关闭与播完同时触发都靠它挡住)。
|
||||
中途关闭也照发 —— 产品拍板「中途关闭也算看完」。
|
||||
|
||||
两处都是直接铸币的路径,改动前先看 `start_play` / `grant_play` 上的并发注释。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select, update
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core import media, rewards
|
||||
from app.core.config import settings
|
||||
from app.core import rewards
|
||||
from app.models.app_config import AppConfig
|
||||
from app.models.guide_video import GuideVideoPlay
|
||||
from app.models.wallet import CoinAccount
|
||||
from app.repositories import wallet as crud_wallet
|
||||
|
||||
SCENES = ("coupon", "comparison")
|
||||
_KEY_BY_SCENE = {
|
||||
"coupon": "coupon_guide_video",
|
||||
"comparison": "comparison_guide_video",
|
||||
}
|
||||
_KEY = "coupon_guide_video"
|
||||
|
||||
#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
|
||||
BIZ_TYPE = "guide_video"
|
||||
CIRCLE_COUNT = 10
|
||||
PLAN_TTL = timedelta(minutes=10)
|
||||
MIN_PLAYS = 1
|
||||
|
||||
# 默认值 = 「运营还没配」时的行为: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
|
||||
MIN_REWARD_COIN = 10
|
||||
REWARD_COIN_LIMIT = 10_000
|
||||
|
||||
_DEFAULTS: dict[str, Any] = {
|
||||
"enabled": False,
|
||||
"video_url": None,
|
||||
"max_plays": 3,
|
||||
"reward_coin": 100,
|
||||
"duration_ms": None,
|
||||
"video_codec": None,
|
||||
"audio_codec": None,
|
||||
"analysis_status": "missing",
|
||||
"analysis_error": None,
|
||||
"config_version": 0,
|
||||
}
|
||||
_FIELDS = tuple(_DEFAULTS)
|
||||
|
||||
|
||||
class PlayStateError(Exception):
|
||||
def __init__(self, code: str, message: str, *, reprepare_required: bool = False):
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.message = message
|
||||
self.reprepare_required = reprepare_required
|
||||
|
||||
def detail(self) -> dict[str, Any]:
|
||||
return {
|
||||
"code": self.code,
|
||||
"message": self.message,
|
||||
"retryable": False,
|
||||
"reprepare_required": self.reprepare_required,
|
||||
}
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(rewards.CN_TZ).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _config_key(scene: str) -> str:
|
||||
if scene not in _KEY_BY_SCENE:
|
||||
raise ValueError(f"unsupported guide video scene: {scene}")
|
||||
return _KEY_BY_SCENE[scene]
|
||||
# ===== 配置 =====
|
||||
|
||||
|
||||
def _merge(raw: Any) -> dict[str, Any]:
|
||||
"""DB 里(可能不全的)dict 叠加到默认上,得到完整配置(4 个字段,无 updated_at)。"""
|
||||
out = dict(_DEFAULTS)
|
||||
if isinstance(raw, dict):
|
||||
for key in _FIELDS:
|
||||
if key in raw and raw[key] is not None:
|
||||
out[key] = raw[key]
|
||||
for k in _FIELDS:
|
||||
v = raw.get(k)
|
||||
if v is not None:
|
||||
out[k] = v
|
||||
return out
|
||||
|
||||
|
||||
def _public_config(cfg: dict[str, Any]) -> dict[str, Any]:
|
||||
duration = int(cfg.get("duration_ms") or 0)
|
||||
reward = int(cfg.get("reward_coin") or 0)
|
||||
return {
|
||||
**cfg,
|
||||
"duration_ms": duration or None,
|
||||
"circle_count": CIRCLE_COUNT,
|
||||
"circle_duration_ms": duration / CIRCLE_COUNT if duration else None,
|
||||
"reward_per_circle": reward // CIRCLE_COUNT,
|
||||
}
|
||||
|
||||
|
||||
def get_config(db: Session, scene: str = "coupon") -> dict[str, Any]:
|
||||
row = db.get(AppConfig, _config_key(scene))
|
||||
cfg = _public_config(_merge(row.value if row is not None else None))
|
||||
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 _validate_settings(max_plays: int, reward_coin: int) -> None:
|
||||
if not MIN_PLAYS <= max_plays <= MAX_PLAYS_LIMIT:
|
||||
raise ValueError(f"播放次数必须在 {MIN_PLAYS}~{MAX_PLAYS_LIMIT} 之间")
|
||||
if not MIN_REWARD_COIN <= reward_coin <= REWARD_COIN_LIMIT:
|
||||
raise ValueError(f"金币总价必须在 {MIN_REWARD_COIN}~{REWARD_COIN_LIMIT} 之间")
|
||||
if reward_coin % CIRCLE_COUNT:
|
||||
raise ValueError("金币总价必须是 10 的倍数")
|
||||
|
||||
|
||||
def _write(
|
||||
db: Session, value: dict[str, Any], *, scene: str, admin_id: int, commit: bool
|
||||
) -> dict[str, Any]:
|
||||
key = _config_key(scene)
|
||||
row = db.get(AppConfig, key)
|
||||
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)
|
||||
row = AppConfig(key=_KEY, value=value, updated_by_admin_id=admin_id)
|
||||
db.add(row)
|
||||
else:
|
||||
row.value = value
|
||||
row.value = value # 整体重新赋值,SQLAlchemy 才侦测得到变更
|
||||
row.updated_by_admin_id = admin_id
|
||||
if commit:
|
||||
db.commit()
|
||||
db.refresh(row)
|
||||
else:
|
||||
db.flush()
|
||||
return get_config(db, scene)
|
||||
out = _merge(row.value)
|
||||
out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
|
||||
return out
|
||||
|
||||
|
||||
def update_config(
|
||||
@@ -133,364 +98,193 @@ def update_config(
|
||||
enabled: bool | None = None,
|
||||
max_plays: int | None = None,
|
||||
reward_coin: int | None = None,
|
||||
scene: str = "coupon",
|
||||
admin_id: int,
|
||||
commit: bool = True,
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
before = get_config(db, scene)
|
||||
raw = {key: before[key] for key in _FIELDS}
|
||||
candidate_plays = int(max_plays if max_plays is not None else raw["max_plays"])
|
||||
candidate_reward = int(reward_coin if reward_coin is not None else raw["reward_coin"])
|
||||
_validate_settings(candidate_plays, candidate_reward)
|
||||
if enabled is True and (
|
||||
not raw.get("video_url")
|
||||
or raw.get("analysis_status") != "valid"
|
||||
or not raw.get("duration_ms")
|
||||
):
|
||||
raise ValueError("请先上传并通过分析的视频,再启用")
|
||||
"""改开关 / 次数 / 金币(只改传了的字段;视频走 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:
|
||||
raw["enabled"] = enabled
|
||||
raw["max_plays"] = candidate_plays
|
||||
raw["reward_coin"] = candidate_reward
|
||||
raw["config_version"] = int(raw.get("config_version") or 0) + 1
|
||||
after = _write(db, raw, scene=scene, admin_id=admin_id, commit=commit)
|
||||
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,
|
||||
*,
|
||||
analysis: dict[str, Any] | None = None,
|
||||
scene: str = "coupon",
|
||||
admin_id: int,
|
||||
commit: bool = True,
|
||||
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
|
||||
) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
before = get_config(db, scene)
|
||||
raw = {key: before[key] for key in _FIELDS}
|
||||
raw["video_url"] = video_url
|
||||
if video_url:
|
||||
if not analysis or analysis.get("analysis_status") != "valid":
|
||||
raise ValueError("视频必须先完成服务端分析")
|
||||
for key in (
|
||||
"duration_ms", "video_codec", "audio_codec",
|
||||
"analysis_status", "analysis_error",
|
||||
):
|
||||
raw[key] = analysis.get(key)
|
||||
else:
|
||||
raw.update(
|
||||
duration_ms=None,
|
||||
video_codec=None,
|
||||
audio_codec=None,
|
||||
analysis_status="missing",
|
||||
analysis_error=None,
|
||||
)
|
||||
raw["enabled"] = False
|
||||
raw["config_version"] = int(raw.get("config_version") or 0) + 1
|
||||
after = _write(db, raw, scene=scene, admin_id=admin_id, commit=commit)
|
||||
"""设置/清空引导视频地址。返回 (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, scene: str = "coupon") -> int:
|
||||
# ===== 播放计次 =====
|
||||
|
||||
|
||||
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,
|
||||
GuideVideoPlay.scene == scene,
|
||||
GuideVideoPlay.status != "prepared",
|
||||
GuideVideoPlay.user_id == user_id
|
||||
)
|
||||
).scalar_one()
|
||||
)
|
||||
|
||||
|
||||
def _prepare_miss(scene: str, reason: str, cfg: dict[str, Any], used: int) -> dict[str, Any]:
|
||||
maximum = int(cfg.get("max_plays") or 0)
|
||||
return {
|
||||
"should_play": False,
|
||||
"reason": reason,
|
||||
"scene": scene,
|
||||
"video_url": None,
|
||||
"play_token": "",
|
||||
"config_version": int(cfg.get("config_version") or 0),
|
||||
"duration_ms": 0,
|
||||
"circle_count": CIRCLE_COUNT,
|
||||
"circle_duration_ms": 0,
|
||||
"reward_coin": int(cfg.get("reward_coin") or 0),
|
||||
"reward_per_circle": int(cfg.get("reward_coin") or 0) // CIRCLE_COUNT,
|
||||
"seq": used,
|
||||
"remaining": max(0, maximum - used),
|
||||
"expires_at": None,
|
||||
}
|
||||
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 prepare_play(db: Session, user_id: int, *, scene: str = "coupon") -> dict[str, Any]:
|
||||
cfg = get_config(db, scene)
|
||||
used = used_plays(db, user_id, scene)
|
||||
video_url = str(cfg.get("video_url") or "").strip()
|
||||
duration = int(cfg.get("duration_ms") or 0)
|
||||
maximum = int(cfg.get("max_plays") or 0)
|
||||
if not cfg.get("enabled"):
|
||||
return _prepare_miss(scene, "disabled", cfg, used)
|
||||
if not video_url or cfg.get("analysis_status") != "valid" or duration <= 0:
|
||||
return _prepare_miss(scene, "video_unavailable", cfg, used)
|
||||
if used >= maximum:
|
||||
return _prepare_miss(scene, "play_limit_reached", cfg, used)
|
||||
def start_play(
|
||||
db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
|
||||
) -> dict[str, Any]:
|
||||
"""决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
|
||||
|
||||
now = _now()
|
||||
返回 dict:
|
||||
should_play 是否放引导视频(False → 客户端照旧放广告)
|
||||
video_url 相对地址(/media/...);客户端自行拼 BASE_URL
|
||||
play_token 发币幂等键(should_play=False 时为空串)
|
||||
reward_coin 播完/中途关闭都发的固定金币
|
||||
seq / remaining 第几次 / 发完这次还剩几次(仅展示与排查用)
|
||||
"""
|
||||
cfg = get_config(db)
|
||||
video_url = (cfg.get("video_url") or "").strip()
|
||||
max_plays = int(cfg.get("max_plays") or 0)
|
||||
reward_coin = int(cfg.get("reward_coin") or 0)
|
||||
used = used_plays(db, user_id)
|
||||
|
||||
def _miss(used_now: int) -> dict[str, Any]:
|
||||
return {
|
||||
"should_play": False,
|
||||
"video_url": None,
|
||||
"play_token": "",
|
||||
"reward_coin": reward_coin,
|
||||
"seq": used_now,
|
||||
"remaining": max(0, max_plays - used_now),
|
||||
}
|
||||
|
||||
if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
|
||||
return _miss(used)
|
||||
|
||||
seq = used + 1
|
||||
play = GuideVideoPlay(
|
||||
user_id=user_id,
|
||||
play_token=uuid.uuid4().hex,
|
||||
scene=scene,
|
||||
seq=None,
|
||||
seq=seq,
|
||||
video_url=video_url,
|
||||
coin=int(cfg["reward_coin"]),
|
||||
duration_ms=duration,
|
||||
config_version=int(cfg["config_version"]),
|
||||
settled_circles=0,
|
||||
status="prepared",
|
||||
coin=0,
|
||||
status="playing",
|
||||
completed=0,
|
||||
prepared_at=now,
|
||||
expires_at=now + PLAN_TTL,
|
||||
started_at=None,
|
||||
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
|
||||
)
|
||||
db.add(play)
|
||||
db.commit()
|
||||
# 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
|
||||
# (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
|
||||
# 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
|
||||
try:
|
||||
if commit:
|
||||
db.commit()
|
||||
else:
|
||||
db.flush()
|
||||
except IntegrityError:
|
||||
db.rollback()
|
||||
return _miss(used_plays(db, user_id))
|
||||
return {
|
||||
"should_play": True,
|
||||
"reason": "ready",
|
||||
"scene": scene,
|
||||
"video_url": video_url,
|
||||
"play_token": play.play_token,
|
||||
"config_version": play.config_version,
|
||||
"duration_ms": duration,
|
||||
"circle_count": CIRCLE_COUNT,
|
||||
"circle_duration_ms": duration / CIRCLE_COUNT,
|
||||
"reward_coin": play.coin,
|
||||
"reward_per_circle": play.coin // CIRCLE_COUNT,
|
||||
"seq": used + 1,
|
||||
"remaining": max(0, maximum - used - 1),
|
||||
"expires_at": play.expires_at.isoformat(),
|
||||
"reward_coin": reward_coin,
|
||||
"seq": seq,
|
||||
"remaining": max(0, max_plays - seq),
|
||||
}
|
||||
|
||||
|
||||
def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
|
||||
"""按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
|
||||
return db.execute(
|
||||
select(GuideVideoPlay).where(
|
||||
GuideVideoPlay.play_token == token.strip(),
|
||||
GuideVideoPlay.play_token == token,
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def _start_out(play: GuideVideoPlay, maximum: int, status: str) -> dict[str, Any]:
|
||||
assert play.started_at is not None and play.seq is not None and play.video_url
|
||||
return {
|
||||
"started": True,
|
||||
"status": status,
|
||||
"play_token": play.play_token,
|
||||
"scene": play.scene,
|
||||
"video_url": play.video_url,
|
||||
"config_version": play.config_version,
|
||||
"duration_ms": play.duration_ms,
|
||||
"circle_count": CIRCLE_COUNT,
|
||||
"circle_duration_ms": play.duration_ms / CIRCLE_COUNT,
|
||||
"reward_coin": play.coin,
|
||||
"reward_per_circle": play.coin // CIRCLE_COUNT,
|
||||
"seq": play.seq,
|
||||
"remaining": max(0, maximum - play.seq),
|
||||
"started_at": play.started_at.isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def start_play(db: Session, user_id: int, *, play_token: str) -> dict[str, Any]:
|
||||
play = _find_play(db, user_id, play_token)
|
||||
if play is None:
|
||||
raise PlayStateError("play_not_found", "播放计划不存在")
|
||||
cfg = get_config(db, play.scene)
|
||||
maximum = int(cfg["max_plays"])
|
||||
if play.status in {"started", "completed"}:
|
||||
return _start_out(play, maximum, "already_started")
|
||||
if play.status != "prepared":
|
||||
raise PlayStateError("play_not_found", "播放计划不可用")
|
||||
now = _now()
|
||||
if play.expires_at is None or now > play.expires_at:
|
||||
raise PlayStateError(
|
||||
"plan_expired", "播放计划已过期,请重新获取",
|
||||
reprepare_required=True,
|
||||
)
|
||||
if int(cfg["config_version"]) != play.config_version:
|
||||
raise PlayStateError(
|
||||
"config_changed", "视频配置已变化,请重新获取",
|
||||
reprepare_required=True,
|
||||
)
|
||||
used = used_plays(db, user_id, play.scene)
|
||||
if used >= maximum:
|
||||
raise PlayStateError("play_limit_reached", "播放次数已用完")
|
||||
play.seq = used + 1
|
||||
play.status = "started"
|
||||
play.started_at = now
|
||||
try:
|
||||
db.commit()
|
||||
db.refresh(play)
|
||||
except IntegrityError as exc:
|
||||
db.rollback()
|
||||
raise PlayStateError("play_limit_reached", "并发起播冲突,请重新获取") from exc
|
||||
return _start_out(play, maximum, "started")
|
||||
|
||||
|
||||
def _coin_balance(db: Session, user_id: int) -> int:
|
||||
account = db.get(CoinAccount, user_id)
|
||||
return int(account.coin_balance if account else 0)
|
||||
|
||||
|
||||
def grant_circle(
|
||||
db: Session, user_id: int, *, play_token: str, circle: int
|
||||
def grant_play(
|
||||
db: Session, user_id: int, *, play_token: str, completed: bool
|
||||
) -> dict[str, Any]:
|
||||
play = _find_play(db, user_id, play_token)
|
||||
balance = _coin_balance(db, user_id)
|
||||
base = {
|
||||
"granted": False,
|
||||
"retryable": False,
|
||||
"retry_after_ms": 0,
|
||||
"circle": circle,
|
||||
"granted_coin": 0,
|
||||
"settled_circles": int(play.settled_circles if play else 0),
|
||||
"coin_balance": balance,
|
||||
}
|
||||
if play is None:
|
||||
return {**base, "status": "not_found"}
|
||||
if play.status not in {"started", "completed"} or play.started_at is None:
|
||||
return {**base, "status": "not_started"}
|
||||
settled = int(play.settled_circles)
|
||||
if circle <= settled:
|
||||
return {
|
||||
**base,
|
||||
"status": "already_granted",
|
||||
"granted_coin": play.coin // CIRCLE_COUNT,
|
||||
"settled_circles": settled,
|
||||
}
|
||||
if settled >= CIRCLE_COUNT:
|
||||
return {**base, "status": "finished", "settled_circles": settled}
|
||||
if circle != settled + 1:
|
||||
return {
|
||||
**base,
|
||||
"status": "out_of_order",
|
||||
"retryable": True,
|
||||
"settled_circles": settled,
|
||||
}
|
||||
# 向上取整到毫秒,绝不让第 N 圈早于精确 N/10 边界发奖。
|
||||
boundary_ms = (circle * play.duration_ms + CIRCLE_COUNT - 1) // CIRCLE_COUNT
|
||||
elapsed_ms = max(0, int((_now() - play.started_at).total_seconds() * 1000))
|
||||
if elapsed_ms < boundary_ms:
|
||||
return {
|
||||
**base,
|
||||
"status": "too_early",
|
||||
"retryable": True,
|
||||
"retry_after_ms": boundary_ms - elapsed_ms,
|
||||
"settled_circles": settled,
|
||||
}
|
||||
"""按 play_token 发这次引导视频的金币(幂等)。播完 / 中途关闭都发。
|
||||
|
||||
final = circle == CIRCLE_COUNT
|
||||
返回 {granted, coin, status}:granted=True 表示**本次调用真的入账了**;
|
||||
重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
|
||||
"""
|
||||
token = (play_token or "").strip()
|
||||
# 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
|
||||
coin = int(get_config(db).get("reward_coin") or 0)
|
||||
|
||||
# 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
|
||||
# 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
|
||||
# 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
|
||||
# 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
|
||||
# (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
|
||||
#
|
||||
# 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
|
||||
# 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
|
||||
# (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
|
||||
won = db.execute(
|
||||
update(GuideVideoPlay)
|
||||
.where(
|
||||
GuideVideoPlay.id == play.id,
|
||||
GuideVideoPlay.status == "started",
|
||||
GuideVideoPlay.settled_circles == settled,
|
||||
GuideVideoPlay.play_token == token,
|
||||
GuideVideoPlay.user_id == user_id,
|
||||
GuideVideoPlay.status == "playing",
|
||||
)
|
||||
.values(
|
||||
settled_circles=circle,
|
||||
status="completed" if final else "started",
|
||||
completed=1 if final else 0,
|
||||
granted_at=_now() if final else None,
|
||||
status="granted",
|
||||
coin=coin,
|
||||
completed=1 if completed else 0,
|
||||
granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
|
||||
)
|
||||
.execution_options(synchronize_session=False)
|
||||
).rowcount
|
||||
|
||||
if not won:
|
||||
# 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
|
||||
# (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
|
||||
db.rollback()
|
||||
fresh = _find_play(db, user_id, play_token)
|
||||
fresh_settled = int(fresh.settled_circles if fresh else 0)
|
||||
return {
|
||||
**base,
|
||||
"status": "already_granted" if circle <= fresh_settled else "out_of_order",
|
||||
"retryable": circle > fresh_settled,
|
||||
"granted_coin": (
|
||||
play.coin // CIRCLE_COUNT if circle <= fresh_settled else 0
|
||||
),
|
||||
"settled_circles": fresh_settled,
|
||||
"coin_balance": _coin_balance(db, user_id),
|
||||
}
|
||||
play = _find_play(db, user_id, token)
|
||||
if play is None:
|
||||
return {"granted": False, "coin": 0, "status": "not_found"}
|
||||
return {"granted": False, "coin": play.coin, "status": "already_granted"}
|
||||
|
||||
per_circle = play.coin // CIRCLE_COUNT
|
||||
account, _ = crud_wallet.grant_coins(
|
||||
db,
|
||||
user_id,
|
||||
per_circle,
|
||||
biz_type=BIZ_TYPE,
|
||||
ref_id=f"{play.play_token}:{circle}",
|
||||
remark=f"新手引导视频第{circle}圈奖励",
|
||||
)
|
||||
try:
|
||||
db.commit()
|
||||
except IntegrityError:
|
||||
# 流水唯一键是第二道幂等保险;若撞键,回滚后返回权威状态。
|
||||
db.rollback()
|
||||
fresh = _find_play(db, user_id, play_token)
|
||||
return {
|
||||
**base,
|
||||
"status": "already_granted",
|
||||
"granted_coin": play.coin // CIRCLE_COUNT,
|
||||
"settled_circles": int(fresh.settled_circles if fresh else settled),
|
||||
"coin_balance": _coin_balance(db, user_id),
|
||||
}
|
||||
return {
|
||||
**base,
|
||||
"granted": True,
|
||||
"status": "granted",
|
||||
"granted_coin": per_circle,
|
||||
"settled_circles": circle,
|
||||
"coin_balance": int(account.coin_balance),
|
||||
}
|
||||
|
||||
|
||||
def cleanup_old_videos(db: Session, *, protection: timedelta = PLAN_TTL) -> int:
|
||||
"""清理超过保护期且未被当前配置/有效计划/近期起播引用的托管旧片。"""
|
||||
now = _now()
|
||||
protected_urls: set[str] = set()
|
||||
for scene in SCENES:
|
||||
url = get_config(db, scene).get("video_url")
|
||||
if url:
|
||||
protected_urls.add(str(url))
|
||||
protected_urls.update(
|
||||
str(url)
|
||||
for url in db.execute(
|
||||
select(GuideVideoPlay.video_url).where(
|
||||
GuideVideoPlay.video_url.is_not(None),
|
||||
(
|
||||
(
|
||||
(GuideVideoPlay.status == "prepared")
|
||||
& (GuideVideoPlay.expires_at >= now)
|
||||
)
|
||||
| (
|
||||
(GuideVideoPlay.status == "started")
|
||||
& (GuideVideoPlay.started_at >= now - protection)
|
||||
)
|
||||
),
|
||||
)
|
||||
).scalars()
|
||||
if url
|
||||
)
|
||||
directory = Path(settings.MEDIA_ROOT) / "guide_video"
|
||||
if not directory.is_dir():
|
||||
return 0
|
||||
cutoff = now.timestamp() - protection.total_seconds()
|
||||
removed = 0
|
||||
for path in directory.glob("guide_*.mp4"):
|
||||
url = f"{settings.MEDIA_URL_PREFIX}/guide_video/{path.name}"
|
||||
try:
|
||||
if url not in protected_urls and path.stat().st_mtime <= cutoff:
|
||||
media.delete_guide_video(url)
|
||||
removed += 1
|
||||
except OSError:
|
||||
continue
|
||||
return removed
|
||||
if coin > 0:
|
||||
crud_wallet.grant_coins(
|
||||
db,
|
||||
user_id,
|
||||
coin,
|
||||
biz_type=BIZ_TYPE,
|
||||
ref_id=token,
|
||||
remark="新手引导视频奖励",
|
||||
)
|
||||
db.commit()
|
||||
return {"granted": True, "coin": coin, "status": "granted"}
|
||||
|
||||
@@ -132,20 +132,10 @@ def get_or_create_account(
|
||||
) -> CoinAccount:
|
||||
"""取用户金币账户,不存在则建一个空账户。
|
||||
|
||||
lock=True 时对已存在的账户行加 SELECT FOR UPDATE,并强制刷新 identity map 中可能
|
||||
缓存的旧余额;所有金币写入口都会使用它。SQLite 下 FOR UPDATE 为 no-op。
|
||||
lock=True 时对已存在的账户行加 SELECT FOR UPDATE(读-算-写余额的调用方串行化,防并发
|
||||
双写余额错位,如 admin set 模式连点);默认 False 不改 C 端发奖行为。SQLite 下为 no-op。
|
||||
"""
|
||||
if lock:
|
||||
# populate_existing 很关键:同一 Session 可能早已缓存旧余额。只加 FOR UPDATE 而
|
||||
# 继续复用 identity map 里的旧对象,仍会在另一笔发奖提交后把余额覆盖回去。
|
||||
acc = db.execute(
|
||||
select(CoinAccount)
|
||||
.where(CoinAccount.user_id == user_id)
|
||||
.with_for_update()
|
||||
.execution_options(populate_existing=True)
|
||||
).scalar_one_or_none()
|
||||
else:
|
||||
acc = db.get(CoinAccount, user_id)
|
||||
acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
|
||||
if acc is None:
|
||||
acc = CoinAccount(
|
||||
user_id=user_id,
|
||||
@@ -175,9 +165,7 @@ def grant_coins(
|
||||
|
||||
返回 (account, transaction)。调用方负责 commit。
|
||||
"""
|
||||
# 所有金币来源统一遵守账户行锁协议;否则 guide 锁了账户,广告/签到等未锁路径
|
||||
# 仍可拿旧余额做 ORM 读改写,最终余额会小于流水累计。
|
||||
acc = get_or_create_account(db, user_id, commit=False, lock=True)
|
||||
acc = get_or_create_account(db, user_id, commit=False)
|
||||
acc.coin_balance += amount
|
||||
if amount > 0:
|
||||
acc.total_coin_earned += amount
|
||||
@@ -210,7 +198,7 @@ def grant_cash(
|
||||
与 [grant_coins] 同模式(运营手动调现金 / 测试发现金用)。返回 (account, transaction),
|
||||
调用方负责 commit。不在此校验扣成负——由调用方(admin router)按业务保护。
|
||||
"""
|
||||
acc = get_or_create_account(db, user_id, commit=False, lock=True)
|
||||
acc = get_or_create_account(db, user_id, commit=False)
|
||||
acc.cash_balance_cents += amount_cents
|
||||
|
||||
txn = CashTransaction(
|
||||
@@ -240,7 +228,7 @@ def grant_invite_cash(
|
||||
invite_cash_transaction,不 commit。与金币兑换的 cash_balance_cents **物理隔离**
|
||||
(产品红线:邀请奖励金 ≠ 金币现金,两本账不可累加)。返回 (account, transaction),
|
||||
调用方负责 commit。不在此校验扣成负——由调用方按业务保护。"""
|
||||
acc = get_or_create_account(db, user_id, commit=False, lock=True)
|
||||
acc = get_or_create_account(db, user_id, commit=False)
|
||||
acc.invite_cash_balance_cents += amount_cents
|
||||
|
||||
txn = InviteCashTransaction(
|
||||
@@ -298,8 +286,7 @@ def exchange_coins_to_cash(
|
||||
if coin_amount < floor_min or coin_amount % COIN_PER_CENT != 0:
|
||||
raise InvalidExchangeAmountError
|
||||
|
||||
# 余额充足校验必须和扣减遵守同一账户锁,否则并发消费都可能通过旧余额校验。
|
||||
acc = get_or_create_account(db, user_id, commit=False, lock=True)
|
||||
acc = get_or_create_account(db, user_id, commit=False)
|
||||
if acc.coin_balance < coin_amount:
|
||||
raise InsufficientCoinError
|
||||
|
||||
|
||||
@@ -107,13 +107,6 @@ class ComparisonRecordIn(BaseModel):
|
||||
# 明细
|
||||
items: list[ComparisonItemIn] = Field(default_factory=list)
|
||||
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
|
||||
# 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
|
||||
# status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
|
||||
# pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
|
||||
platforms: list[dict] = Field(default_factory=list)
|
||||
# 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
|
||||
# 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
|
||||
record_status: str | None = None
|
||||
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
|
||||
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
|
||||
# admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
|
||||
@@ -184,9 +177,6 @@ class ComparisonRecordOut(BaseModel):
|
||||
fail_reason: str | None = None
|
||||
items: list = []
|
||||
comparison_results: list = []
|
||||
# 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
|
||||
# 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
|
||||
platforms: list = []
|
||||
skipped_dish_names: list = []
|
||||
total_ms: int | None = None
|
||||
# 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
|
||||
|
||||
+20
-56
@@ -1,72 +1,36 @@
|
||||
"""客户端引导视频三阶段协议。"""
|
||||
from typing import Literal
|
||||
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约。"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
GuideScene = Literal["coupon", "comparison"]
|
||||
|
||||
|
||||
class GuideVideoPrepareIn(BaseModel):
|
||||
scene: GuideScene = "coupon"
|
||||
|
||||
|
||||
class GuideVideoPrepareOut(BaseModel):
|
||||
should_play: bool
|
||||
reason: str
|
||||
scene: GuideScene
|
||||
video_url: str | None = None
|
||||
play_token: str = ""
|
||||
config_version: int = 0
|
||||
duration_ms: int = 0
|
||||
circle_count: int = 10
|
||||
circle_duration_ms: float = 0
|
||||
reward_coin: int = 0
|
||||
reward_per_circle: int = 0
|
||||
seq: int = 0
|
||||
remaining: int = 0
|
||||
expires_at: str | None = None
|
||||
|
||||
|
||||
class GuideVideoStartIn(BaseModel):
|
||||
play_token: str = Field(min_length=1, max_length=64)
|
||||
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
|
||||
|
||||
scene: str = Field(default="coupon", max_length=16)
|
||||
|
||||
|
||||
class GuideVideoStartOut(BaseModel):
|
||||
started: bool
|
||||
status: Literal["started", "already_started"]
|
||||
play_token: str
|
||||
scene: GuideScene
|
||||
video_url: str
|
||||
config_version: int
|
||||
duration_ms: int
|
||||
circle_count: int = 10
|
||||
circle_duration_ms: float
|
||||
reward_coin: int
|
||||
reward_per_circle: int
|
||||
seq: int
|
||||
remaining: int
|
||||
started_at: str
|
||||
"""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)
|
||||
circle: int = Field(ge=1, le=10)
|
||||
completed: bool = False
|
||||
|
||||
|
||||
class GuideVideoRewardOut(BaseModel):
|
||||
"""granted=True 表示本次调用真的入账(重复上报为 False,coin 是已发金额)。"""
|
||||
|
||||
granted: bool
|
||||
status: Literal[
|
||||
"granted",
|
||||
"already_granted",
|
||||
"too_early",
|
||||
"out_of_order",
|
||||
"not_started",
|
||||
"not_found",
|
||||
"finished",
|
||||
]
|
||||
retryable: bool = False
|
||||
retry_after_ms: int = 0
|
||||
circle: int
|
||||
granted_coin: int = 0
|
||||
settled_circles: int = 0
|
||||
coin_balance: int = 0
|
||||
coin: int
|
||||
status: str
|
||||
|
||||
@@ -201,7 +201,7 @@ def test_start_loses_seq_race_degrades_to_ad(client, guide_configured, monkeypat
|
||||
assert first["should_play"] is True and first["seq"] == 1
|
||||
|
||||
# 本次请求读到的是过期计数 → 仍会算出 seq=1
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
|
||||
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id: 0)
|
||||
with SessionLocal() as db:
|
||||
result = crud_guide.start_play(db, uid)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user