Compare commits

..

5 Commits

Author SHA1 Message Date
zuochenyong 92e47c9d50 更新 run.sh 2026-07-29 22:54:24 +08:00
zuochenyong 591971301d 更新 run.bat 2026-07-29 22:53:15 +08:00
exinglang a1cf1da231 feat(新手引导): 支持十圈进度与逐圈发奖 2026-07-29 22:40:01 +08:00
zuochenyong 53c3b7f60f feat(guide-video): 支持领券和比价独立视频奖励配置 (#196)
Co-authored-by: exinglang <exinglang@qq.com>
Reviewed-on: #196
Co-authored-by: zuochenyong <zuochenyong@wonderable.ai>
Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
2026-07-29 16:12:05 +08:00
marco 4bd4e66678 重构比价结果页取数据逻辑 (#195)
Reviewed-on: #195
2026-07-29 01:59:31 +08:00
29 changed files with 1113 additions and 1174 deletions
-5
View File
@@ -64,11 +64,6 @@ CHUANGLAN_SMS_TEMPLATE_ID=1022457679
CHUANGLAN_SMS_SIGNATURE=
CHUANGLAN_SMS_ENDPOINT=https://smssh.253.com/msg/sms/v2/tpl/send
CHUANGLAN_SMS_TIMEOUT_SEC=10
# --- 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值 ---
ALIYUN_ONEKEY_ACCESS_KEY_ID=
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
ALIYUN_ONEKEY_TIMEOUT_SEC=15
# ===== 测试账号(release 包全流程联调用)=====
# 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
@@ -0,0 +1,26 @@
"""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
@@ -0,0 +1,47 @@
"""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")
@@ -0,0 +1,31 @@
"""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,
)
@@ -0,0 +1,164 @@
"""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 = "旧播放次数不符合 150 约束"
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
View File
@@ -1,6 +1,6 @@
"""Admin 后台 FastAPI app(独立进程)。
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771
启动:uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8773
复用 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:3001",
"http://127.0.0.1:3001",
"http://localhost:3002",
"http://127.0.0.1:3002",
"http://localhost:3000",
"http://127.0.0.1:3000",
]
+60 -32
View File
@@ -9,7 +9,7 @@ client_max_body_size,见 shaguabijia-admin-web/deploy/nginx/admin.shaguabijia.co
"""
from __future__ import annotations
from typing import Annotated
from typing import Annotated, Literal
from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
@@ -27,14 +27,16 @@ router = APIRouter(
)
def _out(db: AdminDb) -> GuideVideoConfigOut:
"""配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
return GuideVideoConfigOut(**guide_video.get_config(db), **guide_video.play_stats(db))
GuideScene = Literal["coupon", "comparison"]
def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
return GuideVideoConfigOut(scene=scene, **guide_video.get_config(db, scene))
@router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
def get_config(db: AdminDb) -> GuideVideoConfigOut:
return _out(db)
def get_config(db: AdminDb, scene: GuideScene = "coupon") -> GuideVideoConfigOut:
return _out(db, scene)
@router.patch("", response_model=GuideVideoConfigOut, summary="改开关/次数/金币(带审计)")
@@ -43,21 +45,26 @@ def update_config(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
scene: GuideScene = "coupon",
) -> 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,
)
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
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,
detail={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
)
db.commit()
return _out(db)
return _out(db, scene)
@router.post("/video", response_model=GuideVideoConfigOut, summary="上传新手引导视频(MP4,带审计)")
@@ -65,23 +72,40 @@ async def upload_video(
request: Request,
admin: Annotated[AdminUser, Depends(require_role("operator"))],
db: AdminDb,
file: UploadFile = File(...),
file: Annotated[UploadFile, File()],
scene: GuideScene = "coupon",
) -> GuideVideoConfigOut:
data = await file.read()
url: str | None = None
try:
url = media.save_guide_video(data)
url, analysis = 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)
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)
@router.delete("/video", response_model=GuideVideoConfigOut, summary="移除新手引导视频(带审计)")
@@ -89,13 +113,17 @@ 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, admin_id=admin.id, commit=False)
before, after = guide_video.set_video(
db, None, scene=scene, 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,
detail={"scene": scene, "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)
# 旧文件由保护期清理任务处理,避免破坏有效计划与已起播快照。
guide_video.cleanup_old_videos(db)
return _out(db, scene)
+30 -11
View File
@@ -1,25 +1,44 @@
"""admin 新手引导视频配置 schemas(开关 / 视频地址 / 前几次 / 每次金币)"""
"""后台引导视频配置协议"""
from __future__ import annotations
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from app.repositories.guide_video import MAX_PLAYS_LIMIT, REWARD_COIN_LIMIT
from app.repositories.guide_video import (
MAX_PLAYS_LIMIT,
MIN_PLAYS,
MIN_REWARD_COIN,
REWARD_COIN_LIMIT,
)
class GuideVideoConfigOut(BaseModel):
scene: str
enabled: bool
video_url: str | None = None # 相对地址 /media/guide_video/xxx.mp4;未配片 = None
video_url: str | None = 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):
"""部分更新:只改传入(非 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)
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
)
@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
+12 -13
View File
@@ -33,7 +33,7 @@ from app.core.security import (
issue_token_pair,
)
from app.integrations import wxpay
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.sms import SmsError, send_code, verify_code
from app.repositories import onboarding as onboarding_repo
from app.repositories import phone_rebind as rebind_repo
@@ -110,15 +110,14 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
):
raise HTTPException(status_code=403, detail="当前设备环境异常,暂无法登录")
logger.info(
"jverify_login provider=%s operator=%s token_len=%d",
req.provider or "-",
"jverify_login operator=%s token_len=%d",
req.operator or "-",
len(req.login_token),
)
try:
phone = verify_and_get_phone(req.provider, req.login_token)
except OneClickError as e:
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
@@ -129,11 +128,11 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
client_ip=_client_ip(request),
outcome="failed",
reason=str(e),
details={"operator": req.operator or None, "provider": req.provider or None},
details={"operator": req.operator or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="jverify")
risk_repo.record_behavior_event(
@@ -147,7 +146,7 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
phone=phone,
client_ip=_client_ip(request),
outcome="success",
details={"operator": req.operator or None, "provider": req.provider or None},
details={"operator": req.operator or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
if user.status != "active":
@@ -493,10 +492,10 @@ def wechat_bind_phone_jverify(
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
try:
phone = verify_and_get_phone("jiguang", req.login_token)
except OneClickError as e:
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
phone = verify_and_get_phone(req.login_token)
except JiguangError as e:
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
return _finish_wechat_bind(
db,
+42 -27
View File
@@ -1,22 +1,16 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)。
路由前缀 `/api/v1/guide-video`(均需 Bearer):
POST /start 这次浮层放引导视频还是放广告?命中则**当场计次**并下发 play_token
POST /reward 播完 / 中途关闭都调,按 play_token 幂等发固定金币
发币额度以**服务端配置**为准(运营后台可改),客户端只报"播完/关闭",报不了金额,
所以被破解也刷不到超额金币;次数上限由 guide_video_play 行数(按账号)硬卡。
"""
"""引导视频 prepare/start/reward 客户端 API。"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, HTTPException
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,
@@ -24,25 +18,41 @@ 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:
"""开播即计数:返回 should_play=True 时服务端已写下这一次,客户端必须真的播。
没配视频 / 开关关 / 次数用完 → should_play=False,客户端照旧走广告链路(行为不变)。
"""
result = crud_guide.start_play(db, user.id, scene=payload.scene or "coupon")
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
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"],
"guide video start user_id=%d token=%s status=%s seq=%d",
user.id, payload.play_token[:12], result["status"], result["seq"],
)
return GuideVideoStartOut(**result)
@@ -50,15 +60,20 @@ def start(payload: GuideVideoStartIn, user: CurrentUser, db: DbSession) -> Guide
@router.post(
"/reward",
response_model=GuideVideoRewardOut,
summary="引导视频发金币(播完/中途关闭都发,play_token 幂等)",
dependencies=[Depends(rate_limit(60, 60, "guide-video-reward"))],
dependencies=[Depends(rate_limit(120, 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
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,
)
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"],
"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"],
)
return GuideVideoRewardOut(**result)
+1 -15
View File
@@ -165,14 +165,6 @@ class Settings(BaseSettings):
CHUANGLAN_SMS_ENDPOINT: str = "https://smssh.253.com/msg/sms/v2/tpl/send"
CHUANGLAN_SMS_TIMEOUT_SEC: int = 10 # httpx 读/连超时秒
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
# 与短信同属 dypnsapi 产品:同一阿里云账号可复用 ALIYUN_SMS_ACCESS_KEY_*,默认独立字段解耦。
# 缺凭证 → provider=aliyun 换号抛错→502,不启动崩(见 aliyun_oneclick_configured)。
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15 # 阿里云 API 读/连超时秒
@property
def aliyun_sms_configured(self) -> bool:
"""阿里云短信凭证齐全(缺则 SMS_PROVIDER=aliyun 时 /sms/* 返 503,而非启动崩)。"""
@@ -183,13 +175,6 @@ class Settings(BaseSettings):
and self.ALIYUN_SMS_TEMPLATE_CODE
)
@property
def aliyun_oneclick_configured(self) -> bool:
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
return bool(
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
)
@property
def chuanglan_sms_configured(self) -> bool:
"""创蓝短信凭证齐全(缺则 SMS_PROVIDER=chuanglan 时 /sms/send 返 503,而非启动崩)。"""
@@ -451,6 +436,7 @@ 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)。
+76 -9
View File
@@ -10,8 +10,13 @@
"""
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
@@ -81,29 +86,91 @@ def _sniff_video_ext(data: bytes) -> str | None:
"""按魔数判定视频类型,返回扩展名;非支持类型返回 None。
只认 MP4 家族(ISO BMFF):`....ftyp` 在偏移 4。Android ExoPlayer 与浏览器 <video>
都稳吃 H.264/AAC 的 mp4;放开 mkv/avi 只会让端上放不出来,不如在入口就挡掉。
支持 H.264 或 HEVC/H.265 视频及 AAC 音频;放开 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>`)。
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("视频时长必须在 30180 秒之间")
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,
}
与图片分开一套校验:体积上限走 [settings.GUIDE_VIDEO_MAX_BYTES],类型只认 MP4。
"""
def save_guide_video(data: bytes) -> tuple[str, dict[str, Any]]:
"""临时落盘、ffprobe 校验后原子发布,返回 URL 与分析结果。"""
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 编码)")
raise MediaError("仅支持 MP4 视频(H.264 或 HEVC/H.265 编码)")
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}"
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
def save_cps_image(admin_id: int, data: bytes) -> str:
-89
View File
@@ -1,89 +0,0 @@
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
链路:
Android 阿里云 SDK getLoginToken → spToken(access_token)
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
与 jiguang 对齐:对外暴露 verify_and_get_phone(login_token)->str,失败抛 AliyunOneClickError,
由 oneclick.py 门面统一 catch。
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
单测 monkeypatch 它即可,不触真 SDK/网络。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
"""
from __future__ import annotations
import logging
from app.core.config import settings
logger = logging.getLogger("shagua.aliyun.onekey")
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
class AliyunOneClickError(Exception):
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
def verify_and_get_phone(login_token: str) -> str:
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
if not settings.aliyun_oneclick_configured:
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
result = _call_get_mobile(login_token)
if not (result["success"] and result["code"] == "OK"):
logger.error(
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
result["code"], result["message"],
)
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
phone = (result["mobile"] or "").strip()
if not (phone.isdigit() and len(phone) == 11):
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
return phone
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
def _get_client():
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
global _client
if _client is None:
from alibabacloud_dypnsapi20170525.client import Client
from alibabacloud_tea_openapi import models as open_api_models
cfg = open_api_models.Config(
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
)
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
_client = Client(cfg)
return _client
def _call_get_mobile(login_token: str) -> dict:
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。
⚠️ GetMobile 响应体字段名以实际 SDK 为准(code=="OK"、get_mobile_result_dto.mobile)。接真号联调时
若字段不同,只需改本函数末尾归一化,verify_and_get_phone 及单测不动。
"""
try:
from alibabacloud_dypnsapi20170525 import models as dypns_models
req = dypns_models.GetMobileRequest(access_token=login_token)
body = _get_client().get_mobile(req).body
except Exception as e:
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
dto = getattr(body, "get_mobile_result_dto", None)
mobile = getattr(dto, "mobile", None) if dto else None
return {
"success": (body.code == "OK"),
"code": body.code,
"message": body.message,
"mobile": mobile,
}
-41
View File
@@ -1,41 +0,0 @@
"""一键登录换号门面:按 provider 分派到极光/阿里云, 统一异常与手机号脱敏。
为什么要门面:一键登录 token 与「拉授权页的那家 SDK」强绑定 —— 客户端用极光 SDK 拉的
token 只能用极光换号, 阿里云的只能用阿里云换号。所以 provider 必须由客户端如实上报, 服务端
按此分派、不能猜。老客户端不带 provider → 默认极光(向后兼容)。
用法(api 层):
from app.integrations import oneclick
try:
phone = oneclick.verify_and_get_phone(req.provider, req.login_token)
except oneclick.OneClickError as e:
raise HTTPException(502, ...) from e
"""
from __future__ import annotations
# 用「模块属性访问」而非 from ... import 函数:保证 monkeypatch 各家实现时门面能拿到替身。
from app.integrations import aliyun_onekey, jiguang
from app.integrations.jiguang import mask_phone # 复用脱敏, 从门面转出供 api 层用
__all__ = ["OneClickError", "mask_phone", "verify_and_get_phone"]
PROVIDER_JIGUANG = "jiguang"
PROVIDER_ALIYUN = "aliyun"
class OneClickError(Exception):
"""换号失败统一异常(不区分厂商), api 层 catch → 502。"""
def verify_and_get_phone(provider: str, login_token: str) -> str:
"""按 provider 换取明文手机号。失败(任一厂商)抛 OneClickError。
provider 大小写不敏感;空 / 未知值兜底走极光(主家), 不因客户端传错值而拒登。
"""
p = (provider or PROVIDER_JIGUANG).strip().lower()
try:
if p == PROVIDER_ALIYUN:
return aliyun_onekey.verify_and_get_phone(login_token)
return jiguang.verify_and_get_phone(login_token)
except (jiguang.JiguangError, aliyun_onekey.AliyunOneClickError) as e:
raise OneClickError(f"[{p}] {e}") from e
+4
View File
@@ -113,6 +113,10 @@ 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 全量),未来取数兜底
+25 -41
View File
@@ -1,20 +1,4 @@
"""新手引导视频播放记录(领券浮层前 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
@@ -26,48 +10,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"),
# 次数上限的**硬约束**: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),
Index(
"uq_guide_video_play_user_scene_seq",
"user_id",
"scene",
"seq",
unique=True,
),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=False
)
# 服务端生成下发给客户端的幂等键(uuid hex)。
play_token: Mapped[str] = mapped_column(String(64), nullable=False)
# 触发场景:目前只有 coupon(领券等候浮层);留字段以便日后比价等场景复用。
scene: Mapped[str] = mapped_column(String(16), nullable=False, default="coupon")
# 本账号第几次(1-based),= 建行时已有行数 + 1。日常判定仍以 COUNT 为准,但 (user_id, seq)
# 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
# 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
# prepare 不占次数,seq=NULLstart 成功才写入 1-based seq
seq: Mapped[int | None] = mapped_column(Integer, nullable=True)
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=中途关闭)。仅留痕:两者都发币。
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")
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
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
)
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}>"
f"<GuideVideoPlay user={self.user_id} scene={self.scene} seq={self.seq} "
f"{self.status} circles={self.settled_circles}>"
)
+9
View File
@@ -60,6 +60,15 @@ 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)
+101 -12
View File
@@ -164,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
is_source_best = best.is_source if best is not None else None
# status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
status = payload.status
# status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
# → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
status = payload.record_status or payload.status
if status is None:
has_valid_target = any(
(not r.is_source) and r.price is not None for r in results
@@ -197,16 +198,35 @@ def upsert_record(
灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
"""
derived = _derive(payload)
# 单源派生: 与 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)
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),
source_platform_id=payload.source_platform_id,
source_platform_name=payload.source_platform_name,
source_package=payload.source_package,
# store_name / source_platform_id / source_platform_name / source_package 统一由
# derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
information=payload.information,
best_deeplink=payload.best_deeplink,
trace_url=payload.trace_url,
@@ -214,6 +234,7 @@ 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,
@@ -286,7 +307,8 @@ def upsert_record(
def _derive_from_results(
results: list[dict], platform_results: dict | None = None
results: list[dict], platform_results: dict | None = None,
record_status: str | None = None,
) -> dict:
"""从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -334,7 +356,53 @@ 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,
"status": "success" if has_valid_target else "failed",
# 记录级结局: 优先用 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"),
}
@@ -493,7 +561,29 @@ def harvest_done(
返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
行不存在(理论上帧0已建;防御)则新建。"""
results = done_params.get("comparison_results") or []
derived = _derive_from_results(results, done_params.get("platform_results"))
# 展示模型统一数组(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, 统一在此算
fail_reason = (
_derive_fail_display(
done_params.get("information"), done_params.get("platform_results")
@@ -501,8 +591,6 @@ 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,
@@ -514,6 +602,7 @@ 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,
+390 -184
View File
@@ -1,95 +1,130 @@
"""新手引导视频:运营配置读写 + 播放计次 + 发币。
**配置**(开关 / 视频地址 / 前几次 / 每次金币)整体作为一个 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` 上的并发注释
"""
"""双场景引导视频配置与 prepare/start/reward 状态机。"""
from __future__ import annotations
import uuid
from datetime import datetime
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
from sqlalchemy import func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from app.core import rewards
from app.core import media, rewards
from app.core.config import settings
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
_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, # 每次固定金币
SCENES = ("coupon", "comparison")
_KEY_BY_SCENE = {
"coupon": "coupon_guide_video",
"comparison": "comparison_guide_video",
}
_FIELDS = tuple(_DEFAULTS.keys())
# 后台可配范围的护栏:防手滑把次数/金币填成天文数字(配置直接决定发币)。
BIZ_TYPE = "guide_video"
CIRCLE_COUNT = 10
PLAN_TTL = timedelta(minutes=10)
MIN_PLAYS = 1
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 k in _FIELDS:
v = raw.get(k)
if v is not None:
out[k] = v
for key in _FIELDS:
if key in raw and raw[key] is not None:
out[key] = raw[key]
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)
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))
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)
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)
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 # 整体重新赋值,SQLAlchemy 才侦测得到变更
row.value = value
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
return get_config(db, scene)
def update_config(
@@ -98,193 +133,364 @@ 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]]:
"""改开关 / 次数 / 金币(只改传了的字段;视频走 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}
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("请先上传并通过分析的视频,再启用")
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)
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)
return before, after
def set_video(
db: Session, video_url: str | None, *, admin_id: int, commit: bool = True
db: Session,
video_url: str | None,
*,
analysis: dict[str, Any] | None = None,
scene: str = "coupon",
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)
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)
return before, after
# ===== 播放计次 =====
def used_plays(db: Session, user_id: int) -> int:
"""该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
def used_plays(db: Session, user_id: int, scene: str = "coupon") -> int:
return int(
db.execute(
select(func.count()).select_from(GuideVideoPlay).where(
GuideVideoPlay.user_id == user_id
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.scene == scene,
GuideVideoPlay.status != "prepared",
)
).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 _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 start_play(
db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
) -> dict[str, Any]:
"""决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
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)
返回 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
now = _now()
play = GuideVideoPlay(
user_id=user_id,
play_token=uuid.uuid4().hex,
scene=scene,
seq=seq,
seq=None,
video_url=video_url,
coin=0,
status="playing",
coin=int(cfg["reward_coin"]),
duration_ms=duration,
config_version=int(cfg["config_version"]),
settled_circles=0,
status="prepared",
completed=0,
started_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
prepared_at=now,
expires_at=now + PLAN_TTL,
started_at=None,
)
db.add(play)
# 上面的 COUNT 判定是无锁 check-then-insert:并发 /start 会都算出同一个 seq。
# (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
# 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
try:
if commit:
db.commit()
else:
db.flush()
except IntegrityError:
db.rollback()
return _miss(used_plays(db, user_id))
db.commit()
return {
"should_play": True,
"reason": "ready",
"scene": scene,
"video_url": video_url,
"play_token": play.play_token,
"reward_coin": reward_coin,
"seq": seq,
"remaining": max(0, max_plays - seq),
"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(),
}
def _find_play(db: Session, user_id: int, token: str) -> GuideVideoPlay | None:
"""按 (play_token, user_id) 取播放行 —— 带 user_id 是防拿别人的 token 来兑。"""
return db.execute(
select(GuideVideoPlay).where(
GuideVideoPlay.play_token == token,
GuideVideoPlay.play_token == token.strip(),
GuideVideoPlay.user_id == user_id,
)
).scalar_one_or_none()
def grant_play(
db: Session, user_id: int, *, play_token: str, completed: bool
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
) -> dict[str, Any]:
"""按 play_token 发这次引导视频的金币(幂等)。播完 / 中途关闭都发。
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,
}
返回 {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%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
final = circle == CIRCLE_COUNT
won = db.execute(
update(GuideVideoPlay)
.where(
GuideVideoPlay.play_token == token,
GuideVideoPlay.user_id == user_id,
GuideVideoPlay.status == "playing",
GuideVideoPlay.id == play.id,
GuideVideoPlay.status == "started",
GuideVideoPlay.settled_circles == settled,
)
.values(
status="granted",
coin=coin,
completed=1 if completed else 0,
granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
settled_circles=circle,
status="completed" if final else "started",
completed=1 if final else 0,
granted_at=_now() if final else None,
)
.execution_options(synchronize_session=False)
).rowcount
if not won:
# 没抢到:token 不存在 / 不是本人的 / 已被另一次上报发过。回滚拿干净快照再区分两者
# (对手此时必然已提交 —— 我们就是被它挡下的,所以读得到它写的 coin)。
db.rollback()
play = _find_play(db, user_id, token)
if play is None:
return {"granted": False, "coin": 0, "status": "not_found"}
return {"granted": False, "coin": play.coin, "status": "already_granted"}
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),
}
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"}
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
+20 -7
View File
@@ -132,10 +132,20 @@ def get_or_create_account(
) -> CoinAccount:
"""取用户金币账户,不存在则建一个空账户。
lock=True 时对已存在的账户行加 SELECT FOR UPDATE(--写余额的调用方串行化,防并发
双写余额错位, admin set 模式连点);默认 False 不改 C 端发奖行为SQLite 下为 no-op
lock=True 时对已存在的账户行加 SELECT FOR UPDATE并强制刷新 identity map 中可能
缓存的旧余额所有金币写入口都会使用它SQLite FOR UPDATE no-op
"""
acc = db.get(CoinAccount, user_id, with_for_update=True) if lock else db.get(CoinAccount, user_id)
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)
if acc is None:
acc = CoinAccount(
user_id=user_id,
@@ -165,7 +175,9 @@ def grant_coins(
返回 (account, transaction)调用方负责 commit
"""
acc = get_or_create_account(db, user_id, commit=False)
# 所有金币来源统一遵守账户行锁协议;否则 guide 锁了账户,广告/签到等未锁路径
# 仍可拿旧余额做 ORM 读改写,最终余额会小于流水累计。
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.coin_balance += amount
if amount > 0:
acc.total_coin_earned += amount
@@ -198,7 +210,7 @@ def grant_cash(
[grant_coins] 同模式(运营手动调现金 / 测试发现金用)返回 (account, transaction),
调用方负责 commit不在此校验扣成负由调用方(admin router)按业务保护
"""
acc = get_or_create_account(db, user_id, commit=False)
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.cash_balance_cents += amount_cents
txn = CashTransaction(
@@ -228,7 +240,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)
acc = get_or_create_account(db, user_id, commit=False, lock=True)
acc.invite_cash_balance_cents += amount_cents
txn = InviteCashTransaction(
@@ -286,7 +298,8 @@ 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)
# 余额充足校验必须和扣减遵守同一账户锁,否则并发消费都可能通过旧余额校验。
acc = get_or_create_account(db, user_id, commit=False, lock=True)
if acc.coin_balance < coin_amount:
raise InsufficientCoinError
-5
View File
@@ -67,11 +67,6 @@ class JverifyLoginRequest(BaseModel):
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
)
provider: str = Field(
"jiguang",
description="一键登录厂商:jiguang(默认)/aliyun。决定后端用哪家换号,"
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
)
# ===== 短信验证码 =====
+10
View File
@@ -107,6 +107,13 @@ 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 定——是
@@ -177,6 +184,9 @@ 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。
+56 -20
View File
@@ -1,36 +1,72 @@
"""新手引导视频(领券等候浮层前 N 次替代广告)的客户端请求/响应契约"""
from __future__ import annotations
"""客户端引导视频三阶段协议"""
from typing import Literal
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):
"""开播询问。scene 目前只有 coupon(领券浮层);预留给日后比价等场景。"""
scene: str = Field(default="coupon", max_length=16)
play_token: str = Field(min_length=1, max_length=64)
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 # 发完这次还剩几次
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
class GuideVideoRewardIn(BaseModel):
"""播完或中途关闭都调这个;completed 只做留痕,两者都发币。"""
play_token: str = Field(min_length=1, max_length=64)
completed: bool = False
circle: int = Field(ge=1, le=10)
class GuideVideoRewardOut(BaseModel):
"""granted=True 表示本次调用真的入账(重复上报为 False,coin 是已发金额)。"""
granted: bool
coin: int
status: str
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
@@ -1,479 +0,0 @@
# 阿里云一键登录 · 后端实施计划 (M1)
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 让 `POST /api/v1/auth/jverify-login` 支持 `provider=aliyun`,用客户端 loginToken 调阿里云 Dypnsapi `GetMobile` 换明文手机号完成登录;极光/创蓝不受影响。
**Architecture:** 复用已存在的 `app/integrations/oneclick.py` 门面(按 provider 分派)。新增 `aliyun_onekey.py` 换号 provider(结构仿 `chuanglan_onekey.py`SDK 调用仿 `sms/aliyun.py` 的惰性 client)。把端点从直连 `jiguang` 改走门面并读 `req.provider`(这一步会让已存在但未接线的 `tests/test_jverify_login_endpoint.py` 转绿)。
**Tech Stack:** FastAPI + Pydantic v2 + pytest + `alibabacloud_dypnsapi20170525`(短信侧已引)。
**参考规范:** `../specs/2026-07-28-aliyun-oneclick-login-design.md`(在 android 仓)。
**运行测试:** `.venv/Scripts/python -m pytest tests/test_oneclick.py tests/test_aliyun_onekey.py tests/test_jverify_login_endpoint.py -q`。(conftest 用临时文件 SQLite,端点测试**无需 PG**`SGB_TEST_SKIP_DB` 未被 conftest 使用。)
> **执行状态(2026-07-28):M1 已实施并验证通过(未提交,待授权)。**
> 实施中发现原计划前提有误:`oneclick.py` 门面 / `chuanglan_onekey.py` / `test_oneclick.py` / `test_jverify_login_endpoint.py` 均为**未提交的创蓝评估 WIP,已清除,git 全程无记录**——门面并不存在。遂按「建小 facadejiguang+aliyun,弃创蓝)」重做,顺序:配置 → aliyun_onekey → **新建** oneclick 门面 → 接线 auth.py 两处 + provider 字段 → **新建**端点测试。
> 验证:新增 15 单测全绿(aliyun_onekey 4 / oneclick 8 / 端点 3)+ 微信绑号回归绿;全量 `pytest` 与改动前逐一对照——同样 8 个既有失败(compare/coupon/invite/withdraw_tiers/notification,均与本次无关:SQLite/顺序依赖),本次 **0 新增失败**617→632 passed)。
> 下方 B1–B5 为实现参考(代码与落地一致);「已存在的门面/测试转绿」等措辞按本节修正理解,实际是先建 aliyun_onekey/facade 再接线。
---
### Task B1: 端点接入 oneclick 门面 + 加 `provider` 字段
已存在的 `tests/test_jverify_login_endpoint.py` 已按「门面已接线」写(monkeypatch `auth.verify_and_get_phone` 为 2 参、POST 带 `provider`),但线上 `auth.py` 仍直连 jiguang。本任务把它接上,使这些测试转绿。
**Files:**
- Modify: `app/schemas/auth.py:60-69``JverifyLoginRequest``provider`
- Modify: `app/api/v1/auth.py:36`import 改门面)、`:112-157``jverify_login` 调用/异常/埋点)、`:495-498`(微信 `bind-phone/jverify` 端点同一 `verify_and_get_phone`/`JiguangError`,一并改)
- Test: `tests/test_jverify_login_endpoint.py`(已存在,勿改,作为验收);回归 `tests/test_wechat_login.py``tests/test_wechat_conflict.py`(都覆盖微信绑号极光换号路径)
> ⚠️ `auth.py` 有**两处**用 `verify_and_get_phone` + `JiguangError``jverify_login`(119) 和微信 `bind-phone/jverify`(495)。换 import 会同时影响两处,必须都改;否则微信绑号编译/运行报错。
- [ ] **Step 1: 跑现有端点测试确认基线**
Run: `pytest tests/test_jverify_login_endpoint.py -v`
Expected: 3 条用例 **FAIL**(当前端点用 1 参 `verify_and_get_phone(req.login_token)`,被 monkeypatch 成 2 参后 `TypeError`)或在无 PG 时 **SKIP**。任一情况都说明「尚未接线」。
- [ ] **Step 2: schema 加 `provider` 字段**
`app/schemas/auth.py``JverifyLoginRequest`(当前 60-69 行)末尾加一个字段:
```python
class JverifyLoginRequest(BaseModel):
login_token: str = Field(..., description="客户端 loginAuth 拿到的 loginToken", min_length=1)
operator: str = Field("", description="CM/CU/CT,用于日志,可选")
device_id: str = Field(
"", max_length=64,
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
)
device_model: str = Field(
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
)
provider: str = Field(
"jiguang",
description="一键登录厂商:jiguang(默认)/aliyun/chuanglan。决定后端用哪家换号,"
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
)
```
- [ ] **Step 3: 端点改走门面**
`app/api/v1/auth.py` 第 36 行的 import 从 jiguang 换成门面:
```python
# 旧: from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
```
`jverify_login`(当前 102-157 行)里的换号调用与异常分支改为:
```python
try:
phone = verify_and_get_phone(req.provider, req.login_token)
except OneClickError as e:
risk_repo.record_behavior_event(
db,
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
subject_type="device",
subject_id=subject_id,
device_id=req.device_id or None,
device_model=req.device_model or None,
client_ip=_client_ip(request),
outcome="failed",
reason=str(e),
details={"operator": req.operator or None, "provider": req.provider or None},
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
)
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
```
并把该函数里成功埋点的 `details={"operator": req.operator or None}`(约 149 行)同样补上 `"provider": req.provider or None`;日志行(约 112-116)改成打印 `req.provider`
换 import 后 `JiguangError` 不再被引用,从第 36 行 import 去掉(只留 `OneClickError, mask_phone, verify_and_get_phone`)。
- [ ] **Step 3b: 微信 `bind-phone/jverify` 端点同步改(否则编译/运行报错)**
`auth.py` 第 495-498 行的第二处换号(微信登录·本机号极光绑定)也走门面。该请求体 `WechatBindPhoneJverifyRequest` **无** `provider` 字段,M1 保持极光不变,显式传字面量 `"jiguang"`
```python
try:
phone = verify_and_get_phone("jiguang", req.login_token)
except OneClickError as e:
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
```
> 说明:当客户端 `WECHAT_BIND` 皮肤改用阿里云后,此端点 + `WechatBindPhoneJverifyRequest` 也要加 `provider`(否则阿里云 token 被极光换号必失败)。这属**后续增强**,不在 M1;已在 spec 待办登记。
- [ ] **Step 4: 跑端点测试转绿**
Run: `pytest tests/test_jverify_login_endpoint.py -v`(有 PG
Expected: 3 条全 **PASS**(默认 provider 建号、provider 原样透传、OneClickError→502)。
- [ ] **Step 5: 跑门面 + 微信绑号测试不回归**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`(门面本就 2 参,未动 → PASS)
Run(有 PG: `pytest tests/test_wechat_login.py tests/test_wechat_conflict.py -v`
Expected: 全 **PASS**(微信绑号极光换号路径改走门面后行为不变)。
- [ ] **Step 6: Commit**
```bash
git add app/schemas/auth.py app/api/v1/auth.py
git commit -m "feat(auth): 一键登录/微信绑号接入 oneclick 门面并支持 provider 字段"
```
---
### Task B2: 阿里云一键登录配置项
**Files:**
- Modify: `app/core/config.py`(极光段 68-70 之后、或阿里云短信段 149-158 附近加一段)
- Modify: `.env.example`
- [ ] **Step 1: 加配置字段 + 已配置属性**
`app/core/config.py``Settings` 类里新增(放在阿里云短信段落之后,与 `aliyun_sms_configured` 属性相邻):
```python
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
# 与短信同属 dypnsapi 产品:同一阿里云账号时可复用 ALIYUN_SMS_ACCESS_KEY_*,
# 默认独立字段解耦(见 spec §2 决策)。缺凭证 → provider=aliyun 换号抛错→502,不启动崩。
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15
```
并在 `aliyun_sms_configured` 属性下方加:
```python
@property
def aliyun_oneclick_configured(self) -> bool:
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
return bool(
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
)
```
- [ ] **Step 2: 补 `.env.example`**
`.env.example` 里对应位置追加(值留空,注释说明可复用短信 AK/SK):
```dotenv
# 阿里云号码认证·一键登录(Dypnsapi GetMobile)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值。
ALIYUN_ONEKEY_ACCESS_KEY_ID=
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
ALIYUN_ONEKEY_TIMEOUT_SEC=15
```
- [ ] **Step 3: 冒烟导入不报错**
Run: `SGB_TEST_SKIP_DB=1 python -c "from app.core.config import settings; print(settings.aliyun_oneclick_configured)"`
Expected: 打印 `False`(默认空凭证)。
- [ ] **Step 4: Commit**
```bash
git add app/core/config.py .env.example
git commit -m "feat(config): 增加阿里云一键登录换号配置项"
```
---
### Task B3: 实现 `aliyun_onekey` 换号 providerTDD
**Files:**
- Create: `app/integrations/aliyun_onekey.py`
- Test: `tests/test_aliyun_onekey.py`
- [ ] **Step 1: 写失败测试**
Create `tests/test_aliyun_onekey.py`
```python
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
"""
from __future__ import annotations
import pytest
from app.core.config import settings
from app.integrations import aliyun_onekey
def _configure(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
def test_get_phone_success(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
)
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
def test_api_failure_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_non_phone_result_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_not_configured_raises(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
```
- [ ] **Step 2: 跑测试确认失败**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
Expected: FAIL`ModuleNotFoundError: app.integrations.aliyun_onekey`
- [ ] **Step 3: 实现 provider**
Create `app/integrations/aliyun_onekey.py`
```python
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
链路:
Android 阿里云 SDK getLoginToken → spToken(access_token)
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
与 jiguang/chuanglan 对齐:对外暴露 verify_and_get_phone(login_token)->str,
失败抛 AliyunOneClickError,由 oneclick.py 门面统一 catch。
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
单测 monkeypatch 它即可。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
"""
from __future__ import annotations
import logging
from app.core.config import settings
logger = logging.getLogger("shagua.aliyun.onekey")
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
class AliyunOneClickError(Exception):
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
def verify_and_get_phone(login_token: str) -> str:
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
if not settings.aliyun_oneclick_configured:
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
result = _call_get_mobile(login_token)
if not (result["success"] and result["code"] == "OK"):
logger.error(
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
result["code"], result["message"],
)
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
phone = (result["mobile"] or "").strip()
if not (phone.isdigit() and len(phone) == 11):
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
return phone
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
def _get_client():
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
global _client
if _client is None:
from alibabacloud_dypnsapi20170525.client import Client
from alibabacloud_tea_openapi import models as open_api_models
cfg = open_api_models.Config(
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
)
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
_client = Client(cfg)
return _client
def _call_get_mobile(login_token: str) -> dict:
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。"""
try:
from alibabacloud_dypnsapi20170525 import models as dypns_models
req = dypns_models.GetMobileRequest(access_token=login_token)
body = _get_client().get_mobile(req).body
except Exception as e:
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
dto = getattr(body, "get_mobile_result_dto", None)
mobile = getattr(dto, "mobile", None) if dto else None
return {
"success": (body.code == "OK"),
"code": body.code,
"message": body.message,
"mobile": mobile,
}
```
> 注:`GetMobile` 响应体字段名以实际 SDK 为准(`code`=="OK"、`get_mobile_result_dto.mobile`)。接真号联调时若字段名不同(如 `mobile` 在别的 dto 下),只需改 `_call_get_mobile` 的最后归一化,`verify_and_get_phone` 及测试不动。
- [ ] **Step 4: 跑测试转绿**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
Expected: 4 条全 **PASS**
- [ ] **Step 5: Commit**
```bash
git add app/integrations/aliyun_onekey.py tests/test_aliyun_onekey.py
git commit -m "feat(auth): 新增阿里云一键登录换号 provider(Dypnsapi GetMobile)"
```
---
### Task B4: 门面注册 aliyun 分派(TDD
**Files:**
- Modify: `app/integrations/oneclick.py`
- Test: `tests/test_oneclick.py`(加两条)
- [ ] **Step 1: 加失败测试**
`tests/test_oneclick.py` 顶部 import 加 `aliyun_onekey`
```python
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang, oneclick
```
文件末尾追加:
```python
def test_dispatch_aliyun(monkeypatch):
seen = {}
def fake_al(t):
seen["al"] = t
return "13855555555"
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13855555555"
assert seen["al"] == "tok"
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise aliyun_onekey.AliyunOneClickError("aliyun down")
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("aliyun", "tok")
```
- [ ] **Step 2: 跑确认失败**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py::test_dispatch_aliyun -v`
Expected: FAIL`aliyun` 未注册 → 走默认极光 → `13855555555` 不匹配 / `seen``al`)。
- [ ] **Step 3: 注册 aliyun 分派**
`app/integrations/oneclick.py` 三处改动:
```python
# import(第 17 行)加入 aliyun_onekey:
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang
# 常量段加:
PROVIDER_ALIYUN = "aliyun"
# verify_and_get_phone 分派与异常 catch:
def verify_and_get_phone(provider: str, login_token: str) -> str:
p = (provider or PROVIDER_JIGUANG).strip().lower()
try:
if p == PROVIDER_CHUANGLAN:
return chuanglan_onekey.verify_and_get_phone(login_token)
if p == PROVIDER_ALIYUN:
return aliyun_onekey.verify_and_get_phone(login_token)
return jiguang.verify_and_get_phone(login_token)
except (
jiguang.JiguangError,
chuanglan_onekey.ChuanglanError,
aliyun_onekey.AliyunOneClickError,
) as e:
raise OneClickError(f"[{p}] {e}") from e
```
- [ ] **Step 4: 跑门面测试全绿**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`
Expected: 全 **PASS**(含新增两条 + 原有创蓝/极光分派不回归)。
- [ ] **Step 5: Commit**
```bash
git add app/integrations/oneclick.py tests/test_oneclick.py
git commit -m "feat(auth): oneclick 门面注册阿里云 provider 分派"
```
---
### Task B5: 全量回归 + 真号联调冒烟
- [ ] **Step 1: 全量单测**
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py tests/test_aliyun_onekey.py -v` 及(有 PG`pytest tests/test_jverify_login_endpoint.py -v`
Expected: 全 PASS。
- [ ] **Step 2: 填真实凭证冒烟(联调期)**
`.env``ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET`(阿里云控制台号码认证方案对应账号 AK/SK),起服务,用客户端真机取到的 loginToken:
Run:
```bash
curl -s -X POST http://127.0.0.1:8770/api/v1/auth/jverify-login \
-H 'Content-Type: application/json' \
-d '{"login_token":"<真机取到的token>","provider":"aliyun","device_id":"smoke"}'
```
Expected: 200 + `user.phone` 为真实手机号 + `access_token` 非空。失败看日志 `shagua.aliyun.onekey``GetMobile failed code=...`)。
- [ ] **Step 3: 无凭证降级冒烟**
清空 `ALIYUN_ONEKEY_*` 重启,`provider=aliyun` 请求应 **502**`OneClickError [aliyun] ... not configured`),且极光 `provider` 省略仍 200。
---
## 自检清单(实现者跑完对照)
- [ ] `provider` 缺省 = jiguang,老客户端不改也能登(B1)。
- [ ] `provider=aliyun` 端到端换号 200B5 Step2)。
- [ ] 任一厂商换号失败 → 502,不建号(B1 tests)。
- [ ] 缺阿里云凭证不 crash 启动,仅该 provider 502B2 property + B5 Step3)。
- [ ] 极光/创蓝分派与端点原行为不回归(B4 Step4 / B1 Step5)。
-51
View File
@@ -1,51 +0,0 @@
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
"""
from __future__ import annotations
import pytest
from app.core.config import settings
from app.integrations import aliyun_onekey
def _configure(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
def test_get_phone_success(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
)
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
def test_api_failure_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_non_phone_result_raises(monkeypatch):
_configure(monkeypatch)
monkeypatch.setattr(
aliyun_onekey, "_call_get_mobile",
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
)
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
def test_not_configured_raises(monkeypatch):
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
with pytest.raises(aliyun_onekey.AliyunOneClickError):
aliyun_onekey.verify_and_get_phone("tok")
+1 -1
View File
@@ -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: 0)
monkeypatch.setattr(crud_guide, "used_plays", lambda db, user_id, scene="coupon": 0)
with SessionLocal() as db:
result = crud_guide.start_play(db, uid)
-50
View File
@@ -1,50 +0,0 @@
"""jverify-login 端点:provider 分派 + 错误映射(端到端:换号 → 建号 → 签 JWT)。
需要 DB(建号)本机无 PG client fixture 会自动 skip(SGB_TEST_SKIP_DB=1);
PG 环境(CI 或本地起 Docker)完整跑换号一步 monkeypatch , 只验端点编排:
provider 是否原样传给门面成功建号OneClickError 502
"""
from __future__ import annotations
from app.api.v1 import auth
from app.integrations.oneclick import OneClickError
def test_jverify_login_default_provider_builds_account(client, monkeypatch):
"""不带 provider 的老客户端 → 门面按默认(极光)换号, 建号登录成功。"""
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: "13500000000")
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok-old"})
assert r.status_code == 200
body = r.json()
assert body["user"]["phone"] == "13500000000"
assert body["access_token"]
def test_jverify_login_passes_provider_to_facade(client, monkeypatch):
"""provider=aliyun 必须原样传给门面(它决定用哪家换号)。"""
captured: dict = {}
def fake(provider, token):
captured["provider"] = provider
captured["token"] = token
return "13500000001"
monkeypatch.setattr(auth, "verify_and_get_phone", fake)
r = client.post(
"/api/v1/auth/jverify-login",
json={"login_token": "tok-al", "provider": "aliyun"},
)
assert r.status_code == 200
assert captured["provider"] == "aliyun"
assert captured["token"] == "tok-al"
def test_jverify_login_oneclick_error_maps_to_502(client, monkeypatch):
"""任一厂商换号失败(OneClickError) → 502, 不建号。"""
def boom(provider, token):
raise OneClickError("verify failed")
monkeypatch.setattr(auth, "verify_and_get_phone", boom)
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok"})
assert r.status_code == 502
-74
View File
@@ -1,74 +0,0 @@
"""一键登录换号门面 oneclick 单测:按 provider 分派到极光/阿里云, 统一异常。
纯函数(monkeypatch 掉两家的 verify_and_get_phone), 不碰 DB本机可跑:
SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py
"""
from __future__ import annotations
import pytest
from app.integrations import aliyun_onekey, jiguang, oneclick
def test_dispatch_defaults_to_jiguang_when_blank(monkeypatch):
"""provider 为空 = 老客户端 → 走极光(向后兼容)。"""
seen = {}
def fake_jg(t):
seen["jg"] = t
return "13800000000"
monkeypatch.setattr(jiguang, "verify_and_get_phone", fake_jg)
assert oneclick.verify_and_get_phone("", "tok") == "13800000000"
assert seen["jg"] == "tok"
def test_dispatch_jiguang_explicit(monkeypatch):
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13811111111")
assert oneclick.verify_and_get_phone("jiguang", "tok") == "13811111111"
def test_dispatch_aliyun(monkeypatch):
seen = {}
def fake_al(t):
seen["al"] = t
return "13822222222"
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13822222222"
assert seen["al"] == "tok"
def test_provider_is_case_insensitive(monkeypatch):
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", lambda t: "13844444444")
assert oneclick.verify_and_get_phone("Aliyun", "tok") == "13844444444"
def test_unknown_provider_falls_back_to_jiguang(monkeypatch):
"""未知 provider 兜底走极光(主家), 不因客户端传错值而拒登。"""
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13833333333")
assert oneclick.verify_and_get_phone("weird-value", "tok") == "13833333333"
def test_jiguang_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise jiguang.JiguangError("jg down")
monkeypatch.setattr(jiguang, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("jiguang", "tok")
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
def boom(_t):
raise aliyun_onekey.AliyunOneClickError("aliyun down")
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
with pytest.raises(oneclick.OneClickError):
oneclick.verify_and_get_phone("aliyun", "tok")
def test_mask_phone_reexported():
"""auth 层只依赖 oneclick, 脱敏函数从门面转出。"""
assert oneclick.mask_phone("13800138000") == "138****00"
+5 -5
View File
@@ -147,8 +147,8 @@ def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
"""本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
phone = "13900139005"
# loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...oneclick import verify_and_get_phone,2 参 provider/token)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: phone)
# 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
ticket = client.post(
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
@@ -181,11 +181,11 @@ def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) ->
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
"""换号失败(OneClickError)→ 502。"""
"""极光核验失败(JiguangError)→ 502。"""
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
def _raise(provider: str, token: str) -> str:
raise auth.OneClickError("mock oneclick failure")
def _raise(token: str) -> str:
raise auth.JiguangError("mock jg failure")
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
ticket = client.post(