更新 run.sh

.env.example

@@ -175,9 +175,3 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
 # APPLOG_MAX_BATCH=500                      # 单批最大条数(超 → 422;导入期常量,改需重启)
 # APPLOG_MAX_BODY_BYTES=2097152            # 请求体上限 2MB(超 → 413;运行期可调)
 # APPLOG_MAX_MSG_BYTES=8192                # 单条 msg 超此字节数截断
-
-# ===== 华为 Push =====
-HUAWEI_PUSH_APP_ID=
-HUAWEI_PUSH_APP_SECRET=
-# 0=正式消息(默认)；1=测试消息(仅开发联调，生产必须保持 0)
-HUAWEI_PUSH_TARGET_USER_TYPE=0

alembic/versions/guide_video_ten_circle_v2.py

@@ -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 = "旧播放次数不符合 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)

app/admin/main.py

@@ -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://localhost:3002",
-    "http://127.0.0.1:3001",
+    "http://127.0.0.1:3002",
     "http://localhost:3000",
     "http://127.0.0.1:3000",
 ]

app/admin/routers/guide_video.py

@@ -31,12 +31,7 @@ GuideScene = Literal["coupon", "comparison"]
 
 
 def _out(db: AdminDb, scene: GuideScene) -> GuideVideoConfigOut:
-    """配置 + 播放统计合成响应(四个写接口都以最新状态返回,前端一次同步到位)。"""
+    return GuideVideoConfigOut(scene=scene, **guide_video.get_config(db, scene))
-    return GuideVideoConfigOut(
-        scene=scene,
-        **guide_video.get_config(db, scene),
-        **guide_video.play_stats(db, scene),
-    )
 
 
 @router.get("", response_model=GuideVideoConfigOut, summary="新手引导视频配置(领券浮层)")
@@ -52,15 +47,18 @@ def update_config(
     db: AdminDb,
     scene: GuideScene = "coupon",
 ) -> GuideVideoConfigOut:
-    before, after = guide_video.update_config(
+    try:
-        db,
+        before, after = guide_video.update_config(
-        enabled=body.enabled,
+            db,
-        max_plays=body.max_plays,
+            enabled=body.enabled,
-        reward_coin=body.reward_coin,
+            max_plays=body.max_plays,
-        scene=scene,
+            reward_coin=body.reward_coin,
-        admin_id=admin.id,
+            scene=scene,
-        commit=False,
+            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={"scene": scene, "before": before, "after": after}, ip=get_client_ip(request), commit=False,
@@ -78,21 +76,35 @@ async def upload_video(
     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(
+    try:
-        db, url, scene=scene, admin_id=admin.id, commit=False
+        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,
+        write_audit(
-        detail={"scene": scene, "before": before.get("video_url"), "after": url, "bytes": len(data)},
+            db, admin, action="guide_video.set_video", target_type="guide_video", target_id=None,
-        ip=get_client_ip(request), commit=False,
+            detail={
-    )
+                "scene": scene,
-    db.commit()
+                "before": before.get("video_url"),
-    # 提交成功后再删旧片,避免新片没落库就把旧片丢了
+                "after": url,
-    media.delete_guide_video(before.get("video_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)
 
 
@@ -112,5 +124,6 @@ def delete_video(
         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"))
+    # 旧文件由保护期清理任务处理，避免破坏有效计划与已起播快照。
+    guide_video.cleanup_old_videos(db)
     return _out(db, scene)

app/admin/schemas/guide_video.py

@@ -1,26 +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)
+    max_plays: int | None = Field(default=None, ge=MIN_PLAYS, le=MAX_PLAYS_LIMIT)
-    reward_coin: int | None = Field(default=None, ge=0, le=REWARD_COIN_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

app/api/v1/guide_video.py

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

app/core/config.py

@@ -88,8 +88,6 @@ class Settings(BaseSettings):
     # (OAuth 换 token 时 client_id 即 AppId)。发送走 v1 messages:send,成功码 80000000。
     HUAWEI_PUSH_APP_ID: str = ""
     HUAWEI_PUSH_APP_SECRET: str = ""
-    # 0=正式消息(默认，受正式消息频控)；1=测试消息(仅开发联调，勿用于生产)。
-    HUAWEI_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
     HUAWEI_PUSH_TOKEN_ENDPOINT: str = "https://oauth-login.cloud.huawei.com/oauth2/v3/token"
     HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
         "https://push-api.cloud.huawei.com/v1/{app_id}/messages:send"
@@ -438,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)。

app/core/media.py

@@ -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:
+def _probe_guide_video(path: Path) -> dict[str, Any]:
-    """保存新手引导视频(运营后台上传的运营素材),返回相对 URL(`/media/guide_video/<file>`)。
+    """用 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,
+    }
 
-    与图片分开一套校验:体积上限走 [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"
+    directory = _media_dir("guide_video")
-    (_media_dir("guide_video") / fname).write_bytes(data)
+    stem = f"guide_{secrets.token_hex(8)}"
-    return f"{settings.MEDIA_URL_PREFIX}/guide_video/{fname}"
+    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:

app/integrations/vendor_push.py

@@ -22,7 +22,7 @@ import uuid
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
-from urllib.parse import quote, urlsplit
+from urllib.parse import quote
 
 import httpx
 
@@ -31,7 +31,6 @@ from app.core.config import settings
 logger = logging.getLogger("shagua.vendor_push")
 
 TYPE_ACCESSIBILITY_DISABLED = "accessibility_disabled"
-DATA_EVENT_NOTIFICATION_CREATED = "notification_created"
 SUPPORTED_VENDORS = frozenset({"honor", "huawei", "vivo", "xiaomi", "oppo"})
 
 # vendor key → 中文名(测试/配置状态接口展示用)
@@ -70,18 +69,6 @@ class _CachedToken:
 
 _token_cache: dict[str, _CachedToken] = {}
 
-_LOG_SUMMARY_MAX_CHARS = 1500
-_LOG_STRING_MAX_CHARS = 200
-_SENSITIVE_LOG_KEYS = {
-    "accesstoken",
-    "appkey",
-    "authtoken",
-    "authorization",
-    "clientsecret",
-    "mastersecret",
-    "sign",
-}
-
 
 def normalize_vendor(push_vendor: str | None) -> str | None:
     if not push_vendor:
@@ -128,16 +115,8 @@ def send_notification(
 
     if mock:
         logger.info(
-            "vendor push mock vendor=%s request=%s",
+            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
-            vendor,
+            vendor, token[:12], title, body, extras,
-            _log_summary(
-                {
-                    "push_token": token,
-                    "title": title,
-                    "body": body,
-                    "extras": extras,
-                }
-            ),
         )
         return {
             "mock": True,
@@ -154,28 +133,7 @@ def send_notification(
         "xiaomi": _send_xiaomi,
         "oppo": _send_oppo,
     }
-    started = time.perf_counter()
+    return dispatch[vendor](token, title, body, extras)
-    try:
-        result = dispatch[vendor](token, title, body, extras)
-    except Exception:
-        logger.exception(
-            "vendor push dispatch failed vendor=%s token=%s notification_type=%s "
-            "elapsed_ms=%.1f",
-            vendor,
-            token,
-            extras.get("type", ""),
-            _elapsed_ms(started),
-        )
-        raise
-    logger.info(
-        "vendor push dispatch succeeded vendor=%s token=%s notification_type=%s "
-        "elapsed_ms=%.1f",
-        vendor,
-        token,
-        extras.get("type", ""),
-        _elapsed_ms(started),
-    )
-    return result
 
 
 def send_accessibility_disabled(
@@ -195,169 +153,19 @@ def send_accessibility_disabled(
     )
 
 
-def send_data_event(
-    push_vendor: str,
-    push_token: str,
-    *,
-    event: str,
-    notification_id: str,
-    mock: bool = False,
-) -> dict[str, Any]:
-    """发送无界面的轻量事件；当前只允许站内消息创建事件。"""
-    vendor = normalize_vendor(push_vendor)
-    token = push_token.strip() if push_token else ""
-    if not vendor or vendor not in SUPPORTED_VENDORS:
-        raise VendorPushError(f"unsupported push vendor: {push_vendor}")
-    if not token:
-        raise VendorPushError("push token is empty")
-    if event != DATA_EVENT_NOTIFICATION_CREATED:
-        raise VendorPushError(f"unsupported data event: {event}")
-    if not notification_id:
-        raise VendorPushError("notification_id is empty")
-
-    payload = {"event": event, "notificationId": str(notification_id)}
-    if mock:
-        return {"mock": True, "vendor": vendor, "payload": payload}
-
-    # vivo 的通知已设置 foregroundShow=false：App 在前台时必走
-    # onForegroundMessageArrived，正好就是本事件需要的刷新信号；不重复占用一次推送配额。
-    if vendor == "vivo":
-        return {"skipped": True, "reason": "foreground notification callback"}
-
-    # OPush 当前只支持通知栏消息，没有服务端透传单推接口。
-    # 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
-    # /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
-    if vendor == "oppo":
-        return {"skipped": True, "reason": "oppo does not support data messages"}
-
-    dispatch: dict[str, Callable[[str, dict[str, str]], dict[str, Any]]] = {
-        "honor": _send_honor_data,
-        "huawei": _send_huawei_data,
-        "xiaomi": _send_xiaomi_data,
-    }
-    return dispatch[vendor](token, payload)
-
-
 def _require(value: str, name: str) -> str:
     if not value:
         raise VendorPushError(f"{name} not configured")
     return value
 
 
-def _elapsed_ms(started: float) -> float:
-    return (time.perf_counter() - started) * 1000
-
-
-def _redacted_value(value: Any) -> str:
-    raw = json.dumps(value, ensure_ascii=False, sort_keys=True, default=str)
-    digest = hashlib.sha256(raw.encode()).hexdigest()[:10]
-    return f"<redacted len={len(raw)} sha256={digest}>"
-
-
-def _sanitize_for_log(value: Any, *, key: str = "", depth: int = 0) -> Any:
-    """生成有排障价值、但不暴露服务端鉴权凭据的紧凑日志摘要。"""
-    normalized_key = "".join(char for char in key.lower() if char.isalnum())
-    if normalized_key in _SENSITIVE_LOG_KEYS or normalized_key.endswith("secret"):
-        return _redacted_value(value)
-    if depth >= 10:
-        return f"<{type(value).__name__}>"
-    if isinstance(value, dict):
-        items = list(value.items())
-        sanitized = {
-            str(item_key): _sanitize_for_log(item_value, key=str(item_key), depth=depth + 1)
-            for item_key, item_value in items[:30]
-        }
-        if len(items) > 30:
-            sanitized["<omitted_keys>"] = len(items) - 30
-        return sanitized
-    if isinstance(value, (list, tuple)):
-        items = list(value)
-        sanitized = [_sanitize_for_log(item, key=key, depth=depth + 1) for item in items[:20]]
-        if len(items) > 20:
-            sanitized.append(f"<omitted_items={len(items) - 20}>")
-        return sanitized
-    if isinstance(value, str):
-        if value.lstrip().startswith(("{", "[")):
-            try:
-                decoded = json.loads(value)
-            except ValueError:
-                pass
-            else:
-                return _sanitize_for_log(decoded, key=key, depth=depth + 1)
-        if len(value) > _LOG_STRING_MAX_CHARS:
-            return f"{value[:_LOG_STRING_MAX_CHARS]}…<len={len(value)}>"
-    return value
-
-
-def _log_summary(value: Any) -> str:
-    rendered = json.dumps(
-        _sanitize_for_log(value),
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-        default=str,
-    )
-    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
-        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
-    return rendered
-
-
-def _raw_log_summary(value: Any) -> str:
-    """厂商响应摘要：保留原始字段和值，仅限制单条日志长度。"""
-    rendered = json.dumps(
-        value,
-        ensure_ascii=False,
-        sort_keys=True,
-        separators=(",", ":"),
-        default=str,
-    )
-    if len(rendered) > _LOG_SUMMARY_MAX_CHARS:
-        return f"{rendered[:_LOG_SUMMARY_MAX_CHARS]}…<len={len(rendered)}>"
-    return rendered
-
-
-def _endpoint_for_log(url: str) -> str:
-    parsed = urlsplit(url)
-    return f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
-
-
-def _request_summary(kwargs: dict[str, Any]) -> str:
-    summary: dict[str, Any] = {}
-    if "json" in kwargs:
-        summary["body_type"] = "json"
-        summary["request_payload"] = kwargs["json"]
-    elif "data" in kwargs:
-        summary["body_type"] = "form"
-        summary["request_payload"] = kwargs["data"]
-    if kwargs.get("params"):
-        summary["params"] = kwargs["params"]
-    if kwargs.get("headers"):
-        summary["headers"] = kwargs["headers"]
-    return _log_summary(summary)
-
-
-def _response_summary(resp: Any, parsed: Any | None = None) -> str:
-    if parsed is None:
-        try:
-            parsed = resp.json()
-        except ValueError:
-            pass
-    if parsed is not None:
-        return _raw_log_summary(parsed)
-    return _raw_log_summary(getattr(resp, "text", ""))
-
-
 def _request_json(
     method: str,
     url: str,
     *,
-    vendor: str,
-    operation: str,
     expected_status: tuple[int, ...] = (200,),
     **kwargs: Any,
 ) -> dict[str, Any]:
-    started = time.perf_counter()
-    request_summary = _request_summary(kwargs)
     try:
         resp = httpx.request(
             method,
@@ -366,82 +174,41 @@ def _request_json(
             **kwargs,
         )
     except httpx.HTTPError as e:
-        logger.error(
-            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
-            "outcome=network_error request=%s response=%s elapsed_ms=%.1f error=%s",
-            vendor,
-            operation,
-            method,
-            _endpoint_for_log(url),
-            request_summary,
-            "<no_response>",
-            _elapsed_ms(started),
-            type(e).__name__,
-        )
         raise VendorPushError(f"push http error: {e}") from e
 
     if resp.status_code not in expected_status:
-        logger.error(
+        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
-            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
-            "outcome=http_error http_status=%s request=%s response=%s elapsed_ms=%.1f",
-            vendor,
-            operation,
-            method,
-            _endpoint_for_log(url),
-            resp.status_code,
-            request_summary,
-            _response_summary(resp),
-            _elapsed_ms(started),
-        )
         raise VendorPushError(f"push http {resp.status_code}")
     try:
-        data = resp.json()
+        return resp.json()
     except ValueError as e:
-        logger.error(
-            "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
-            "outcome=invalid_json http_status=%s request=%s response=%s elapsed_ms=%.1f",
-            vendor,
-            operation,
-            method,
-            _endpoint_for_log(url),
-            resp.status_code,
-            request_summary,
-            _response_summary(resp),
-            _elapsed_ms(started),
-        )
         raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
-    logger.info(
-        "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
-        "outcome=success http_status=%s request=%s response=%s elapsed_ms=%.1f",
-        vendor,
-        operation,
-        method,
-        _endpoint_for_log(url),
-        resp.status_code,
-        request_summary,
-        _response_summary(resp, data),
-        _elapsed_ms(started),
-    )
-    return data
 
 
 def _request_form(
     method: str,
     url: str,
     *,
-    vendor: str,
-    operation: str,
     expected_status: tuple[int, ...] = (200,),
     **kwargs: Any,
 ) -> dict[str, Any]:
-    return _request_json(
+    try:
-        method,
+        resp = httpx.request(
-        url,
+            method,
-        vendor=vendor,
+            url,
-        operation=operation,
+            timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
-        expected_status=expected_status,
+            **kwargs,
-        **kwargs,
+        )
-    )
+    except httpx.HTTPError as e:
+        raise VendorPushError(f"push http error: {e}") from e
+
+    if resp.status_code not in expected_status:
+        logger.error("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
+        raise VendorPushError(f"push http {resp.status_code}")
+    try:
+        return resp.json()
+    except ValueError as e:
+        raise VendorPushError(f"push invalid json: {resp.text[:200]}") from e
 
 
 def _cache_get(key: str) -> str | None:
@@ -467,8 +234,6 @@ def _honor_access_token() -> str:
     data = _request_form(
         "POST",
         settings.HONOR_PUSH_TOKEN_ENDPOINT,
-        vendor="honor",
-        operation="authenticate",
         data={
             "grant_type": "client_credentials",
             "client_id": client_id,
@@ -478,22 +243,16 @@ def _honor_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"honor auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"honor auth failed: {data}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
 def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> dict[str, Any]:
     app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
     access_token = _honor_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只负责打开首页，不保证 data 会变成目标 Activity 的 extras。消息中心通知必须
-        # 用自定义页面(type=1)+ intent URI，把 notificationId/type/feedbackId 等直接带给
-        # MainActivity；否则华为/荣耀系统能展示通知，但用户点击后客户端收不到任何导航参数。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
+        # clickAction type=3(打开应用首页)时,荣耀点击会把 data JSON 的键值对注入启动 intent 的
+        # extras(与 HMS 同机制)→ MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
         "data": json.dumps(_click_extras(extras), ensure_ascii=False),
         "notification": {"title": title, "body": body},
         "android": {
@@ -502,7 +261,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
             "notification": {
                 "title": title,
                 "body": body,
-                "clickAction": click_action,
+                "clickAction": {"type": 3},
                 "importance": "NORMAL",
             },
         },
@@ -511,8 +270,6 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
     data = _request_json(
         "POST",
         settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="honor",
-        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json; charset=UTF-8",
@@ -522,28 +279,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
     )
     code = data.get("code")
     if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"honor push failed: {data}")
-    return data
-
-
-def _send_honor_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HONOR_PUSH_APP_ID, "HONOR_PUSH_APP_ID")
-    access_token = _honor_access_token()
-    data = _request_json(
-        "POST",
-        settings.HONOR_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="honor",
-        operation="send_data_event",
-        json={"data": json.dumps(payload, ensure_ascii=False), "token": [token]},
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-            "timestamp": str(int(time.time() * 1000)),
-        },
-    )
-    code = data.get("code")
-    if code is not None and int(code) != 200:
-        raise VendorPushError(f"honor data push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -558,8 +294,6 @@ def _huawei_access_token() -> str:
     data = _request_form(
         "POST",
         settings.HUAWEI_PUSH_TOKEN_ENDPOINT,
-        vendor="huawei",
-        operation="authenticate",
         data={
             "grant_type": "client_credentials",
             "client_id": app_id,
@@ -569,7 +303,7 @@ def _huawei_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"huawei auth failed: {data}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
@@ -578,24 +312,18 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     '80100000' 为部分成功(单 token 场景仍视为失败,错误里带原始响应便于排障)。"""
     app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
     access_token = _huawei_access_token()
-    click_action: dict[str, Any]
-    if extras.get("notificationId"):
-        # type=3 只打开首页，Mate 20/EMUI 不会把 message.data 自动拆成启动 Intent extras。
-        # type=1 的自定义 intent 才能稳定携带反馈记录 id，且冷启动/onNewIntent 都走同一路由。
-        click_action = {"type": 1, "intent": _click_intent_uri(extras)}
-    else:
-        click_action = {"type": 3}
     payload = {
         "validate_only": False,
         "message": {
+            # click_action type=3(打开应用首页)时,HMS 点击会把 data JSON 的键值对注入启动 intent
+            # 的 extras → MainActivity.consumeNavTarget 读 notif_id/notif_type 直达落地。
             "data": json.dumps(_click_extras(extras), ensure_ascii=False),
             "android": {
-                "target_user_type": settings.HUAWEI_PUSH_TARGET_USER_TYPE,
                 "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
                 "notification": {
                     "title": title,
                     "body": body,
-                    "click_action": click_action,
+                    "click_action": {"type": 3},
                     "importance": "NORMAL",
                 },
             },
@@ -605,8 +333,6 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
     data = _request_json(
         "POST",
         settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="huawei",
-        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json; charset=UTF-8",
@@ -614,32 +340,7 @@ def _send_huawei(token: str, title: str, body: str, extras: dict[str, str]) -> d
         },
     )
     if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"huawei push failed: {data}")
-    return data
-
-
-def _send_huawei_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_id = _require(settings.HUAWEI_PUSH_APP_ID, "HUAWEI_PUSH_APP_ID")
-    access_token = _huawei_access_token()
-    data = _request_json(
-        "POST",
-        settings.HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE.format(app_id=app_id),
-        vendor="huawei",
-        operation="send_data_event",
-        json={
-            "validate_only": False,
-            "message": {
-                "data": json.dumps(payload, ensure_ascii=False),
-                "token": [token],
-            },
-        },
-        headers={
-            "Content-Type": "application/json; charset=UTF-8",
-            "Authorization": f"Bearer {access_token}",
-        },
-    )
-    if str(data.get("code", "")) != "80000000":
-        raise VendorPushError(f"huawei data push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -656,8 +357,6 @@ def _vivo_auth_token() -> str:
     data = _request_json(
         "POST",
         settings.VIVO_PUSH_AUTH_ENDPOINT,
-        vendor="vivo",
-        operation="authenticate",
         json={
             "appId": app_id,
             "appKey": app_key,
@@ -667,10 +366,10 @@ def _vivo_auth_token() -> str:
         headers={"Content-Type": "application/json"},
     )
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo auth failed: {data}")
     token = data.get("authToken")
     if not token:
-        raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo auth missing authToken: {data}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -686,22 +385,15 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "timeToLive": settings.PUSH_TIME_TO_LIVE_SEC,
         "requestId": uuid.uuid4().hex,
         "pushMode": settings.VIVO_PUSH_MODE,
-        # vivo SDK 4.1.5 只有在前台展示关闭时才调用
-        # OpenClientPushMessageReceiver.onForegroundMessageArrived。
-        # App 在该回调中刷新服务端未读数并补发一条本地通知；后台/锁屏时仍由
-        # vivo 系统展示，因而任何场景都只会出现一条通知。
-        "foregroundShow": False,
-        # vivo 官方 VPush 角标字段：离线收到通知时先由桌面自动 +1；
-        # App 启动/同步后再由统一未读数通过系统 API 精确校准。
-        "addBadge": True,
         "clientCustomMap": extras,
     }
-    # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
+    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
-    # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
+    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
-    # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
+    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
+    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
     if extras.get("notificationId"):
         payload["skipType"] = 4
-        payload["skipContent"] = _vivo_click_intent_uri(extras)
+        payload["skipContent"] = _click_intent_uri(extras)
     else:
         payload["skipType"] = 1
     if settings.VIVO_PUSH_CATEGORY:
@@ -709,37 +401,14 @@ def _send_vivo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
     data = _request_json(
         "POST",
         settings.VIVO_PUSH_SEND_ENDPOINT,
-        vendor="vivo",
-        operation="send_notification",
         json=payload,
         headers={
             "Content-Type": "application/json",
             "authToken": auth_token,
         },
     )
-    # 10089 = 当前 vivo 应用尚未开通 VPush「离线自动角标」能力。
-    # 角标只是通知中心未读数的镜像，不能因此让整条通知发送失败：去掉 addBadge
-    # 原样重试，通知到达/应用同步后仍由客户端系统 API 按统一未读数精确设置。
-    if int(data.get("result", -1)) == 10089:
-        logger.warning("vivo VPush badge not enabled (10089); retrying without addBadge")
-        fallback_payload = dict(payload)
-        fallback_payload.pop("addBadge", None)
-        fallback_payload["requestId"] = uuid.uuid4().hex
-        data = _request_json(
-            "POST",
-            settings.VIVO_PUSH_SEND_ENDPOINT,
-            vendor="vivo",
-            operation="send_notification",
-            json=fallback_payload,
-            headers={
-                "Content-Type": "application/json",
-                "authToken": auth_token,
-            },
-        )
-        if int(data.get("result", -1)) == 0:
-            data = {**data, "badgeFallback": True}
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"vivo push failed: {data}")
     return data
 
 
@@ -777,40 +446,14 @@ def _send_xiaomi(token: str, title: str, body: str, extras: dict[str, str]) -> d
     data = _request_form(
         "POST",
         settings.XIAOMI_PUSH_SEND_ENDPOINT,
-        vendor="xiaomi",
-        operation="send_notification",
         data=form,
         headers={"Authorization": f"key={app_secret}"},
     )
     code = data.get("code")
     if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"xiaomi push failed: {data}")
     if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"xiaomi push failed: {data}")
-    return data
-
-
-def _send_xiaomi_data(token: str, payload: dict[str, str]) -> dict[str, Any]:
-    app_secret = _require(settings.XIAOMI_PUSH_APP_SECRET, "XIAOMI_PUSH_APP_SECRET")
-    data = _request_form(
-        "POST",
-        settings.XIAOMI_PUSH_SEND_ENDPOINT,
-        vendor="xiaomi",
-        operation="send_data_event",
-        data={
-            "registration_id": token,
-            "restricted_package_name": settings.ANDROID_PACKAGE_NAME,
-            "payload": json.dumps(payload, ensure_ascii=False),
-            "pass_through": "1",
-            "time_to_live": str(settings.PUSH_TIME_TO_LIVE_SEC * 1000),
-        },
-        headers={"Authorization": f"key={app_secret}"},
-    )
-    code = data.get("code")
-    if code not in (0, "0", None):
-        raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
-    if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi data push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -847,24 +490,6 @@ def _click_intent_uri(extras: dict[str, str]) -> str:
     return ";".join(parts)
 
 
-def _vivo_click_intent_uri(extras: dict[str, str]) -> str:
-    """vivo skipType=4 使用官方要求的完整 intent deeplink 形态。
-
-    `?#Intent` 中的问号不能省；OriginOS 对缺少它的 URI 会退化为“仅打开应用”，
-    不把 S. 参数交给目标 Activity。
-    """
-    pkg = settings.ANDROID_PACKAGE_NAME
-    parts = [
-        "intent://push/detail?#Intent",
-        "scheme=shaguabijia",
-        f"component={pkg}/{pkg}.MainActivity",
-        "launchFlags=0x14000000",
-    ]
-    parts += [f"S.{key}={quote(str(value), safe='')}" for key, value in _click_extras(extras).items()]
-    parts.append("end")
-    return ";".join(parts)
-
-
 def _xiaomi_template_param(title: str, alert: str) -> str:
     rendered = (
         settings.XIAOMI_PUSH_TEMPLATE_PARAM_JSON
@@ -897,8 +522,6 @@ def _oppo_auth_token() -> str:
     data = _request_form(
         "POST",
         settings.OPPO_PUSH_AUTH_ENDPOINT,
-        vendor="oppo",
-        operation="authenticate",
         data={
             "app_key": app_key,
             "timestamp": timestamp,
@@ -907,10 +530,10 @@ def _oppo_auth_token() -> str:
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
     if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo auth failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo auth failed: {data}")
     token = (data.get("data") or {}).get("auth_token") or data.get("auth_token")
     if not token:
-        raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo auth missing auth_token: {data}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -925,11 +548,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         "off_line_ttl": ttl_hours,
         "action_parameters": json.dumps(_click_extras(extras), ensure_ascii=False),
     }
-    if extras.get("notificationId"):
-        # 只有已落库的站内消息才计入角标；保活提醒等临时推送不应留下无法消除的未读数。
-        # 客户端打开/已读后会按服务端真实未读总数覆盖，避免长期漂移。
-        notification["badge_operation_type"] = 1
-        notification["badge_message_count"] = 1
     # 点击落地:OPPO SDK 没有点击回调,参数只能靠服务端点击动作配置送达——action_parameters 的
     # 键值对仅在 click_action_type=1/4 时才会注入目标 Activity 的 intent extras(type=0「启动应用」
     # 会忽略它,extras 全丢 → 点了没反应,与小米 notify_effect=1 同款坑)。
@@ -955,8 +573,6 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
     data = _request_form(
         "POST",
         settings.OPPO_PUSH_SEND_ENDPOINT,
-        vendor="oppo",
-        operation="send_notification",
         data={
             "auth_token": auth_token,
             "message": json.dumps(message, ensure_ascii=False),
@@ -964,5 +580,5 @@ def _send_oppo(token: str, title: str, body: str, extras: dict[str, str]) -> dic
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
     if int(data.get("code", -1)) != 0:
-        raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
+        raise VendorPushError(f"oppo push failed: {data}")
     return data

app/models/guide_video.py

@@ -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 再插),
+        Index(
-        # N 个并发 /start 会都读到同一个已用次数、算出同一个 seq,不拦就能各拿一个 token、
+            "uq_guide_video_play_user_scene_seq",
-        # 各发一次金币,3 次上限形同虚设(改包即可无限刷)。seq 唯一 → 并发同 seq 必撞,
+            "user_id",
-        # start_play 捕获 IntegrityError 降级成"这次不放视频"。
+            "scene",
-        # 用 unique Index 而非 UniqueConstraint:与迁移里的 create_index 对齐(SQLite 加约束
+            "seq",
-        # 要整表重建),autogenerate 才不会每次报一条假 diff。
+            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)
+    # prepare 不占次数，seq=NULL；start 成功才写入 1-based seq。
-    # 唯一键让并发 /start 只能成一个 —— 见 __table_args__。
+    seq: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    seq: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
-    # 当次下发的视频地址(运营换片后能回溯用户当时看的是哪支)。
     video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
-    # 实发金币;未发时 0。
     coin: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # playing(已开播未发币) / granted(已发币)。
+    duration_ms: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    status: Mapped[str] = mapped_column(String(16), nullable=False, default="playing")
+    config_version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
-    # 客户端上报时是否播完(true=自然播完 / false=中途关闭)。仅留痕:两者都发币。
+    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(
+    prepared_at: Mapped[datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+        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"<GuideVideoPlay user={self.user_id} scene={self.scene} seq={self.seq} "
-            f"{self.status} coin={self.coin}>"
+            f"{self.status} circles={self.settled_circles}>"
         )

app/models/wallet.py

@@ -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)

app/repositories/guide_video.py

@@ -1,33 +1,20 @@
-"""新手引导视频:运营配置读写 + 播放计次 + 发币。
+"""双场景引导视频配置与 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
+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
 
 SCENES = ("coupon", "comparison")
@@ -35,38 +22,47 @@ _KEY_BY_SCENE = {
     "coupon": "coupon_guide_video",
     "comparison": "comparison_guide_video",
 }
-
-#: 金币流水 biz_type。客户端收益明细按它显示「新手引导视频奖励」。
 BIZ_TYPE = "guide_video"
-
+CIRCLE_COUNT = 10
-# 默认值 = 「运营还没配」时的行为:video_url 为空 → 一律不下发引导视频,浮层维持现状(放广告)。
+PLAN_TTL = timedelta(minutes=10)
-# 所以本功能上线后**不配视频就等于没上线**,不会影响存量用户。
+MIN_PLAYS = 1
-_DEFAULTS: dict[str, Any] = {
-    "enabled": True,
-    "video_url": None,   # None/空 = 未配片 → 不下发,浮层照旧放广告
-    "max_plays": 3,      # 每个账号前 N 次浮层放引导视频
-    "reward_coin": 100,  # 每次固定金币
-}
-
-_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)
 
 
-def _merge(raw: Any) -> dict[str, Any]:
+class PlayStateError(Exception):
-    """DB 里(可能不全的)dict 叠加到默认上,得到完整配置(4 个字段,无 updated_at)。"""
+    def __init__(self, code: str, message: str, *, reprepare_required: bool = False):
-    out = dict(_DEFAULTS)
+        super().__init__(message)
-    if isinstance(raw, dict):
+        self.code = code
-        for k in _FIELDS:
+        self.message = message
-            v = raw.get(k)
+        self.reprepare_required = reprepare_required
-            if v is not None:
+
-                out[k] = v
+    def detail(self) -> dict[str, Any]:
-    return out
+        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:
@@ -75,34 +71,60 @@ def _config_key(scene: str) -> str:
     return _KEY_BY_SCENE[scene]
 
 
+def _merge(raw: Any) -> dict[str, Any]:
+    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]
+    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]:
-    """完整配置 + updated_at(admin 读 / 业务读共用)。"""
     row = db.get(AppConfig, _config_key(scene))
-    cfg = _merge(row.value if row is not None else None)
+    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 _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]:
-    """整体覆写该行(value 须为完整字段 dict),返回合并后的完整配置(含 updated_at)。"""
     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)
         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)
+    return get_config(db, scene)
-    out["updated_at"] = row.updated_at.isoformat() if row.updated_at else None
-    return out
 
 
 def update_config(
@@ -115,199 +137,360 @@ def update_config(
     admin_id: int,
     commit: bool = True,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
-    """改开关 / 次数 / 金币(只改传了的字段;视频走 set_video)。返回 (before, after) 供审计。"""
+    before = get_config(db, scene)
-    row = db.get(AppConfig, _config_key(scene))
+    raw = {key: before[key] for key in _FIELDS}
-    before = _merge(row.value if row is not None else None)
+    candidate_plays = int(max_plays if max_plays is not None else raw["max_plays"])
-    new_value = {k: before[k] for k in _FIELDS}
+    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
+        raw["enabled"] = enabled
-    if max_plays is not None:
+    raw["max_plays"] = candidate_plays
-        new_value["max_plays"] = max(0, min(int(max_plays), MAX_PLAYS_LIMIT))
+    raw["reward_coin"] = candidate_reward
-    if reward_coin is not None:
+    raw["config_version"] = int(raw.get("config_version") or 0) + 1
-        new_value["reward_coin"] = max(0, min(int(reward_coin), REWARD_COIN_LIMIT))
+    after = _write(db, raw, scene=scene, admin_id=admin_id, commit=commit)
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
     return before, after
 
 
 def set_video(
-    db: Session, video_url: str | None, *, scene: str = "coupon",
+    db: Session,
-    admin_id: int, commit: bool = True
+    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'] 供调用方删旧文件。"""
+    before = get_config(db, scene)
-    row = db.get(AppConfig, _config_key(scene))
+    raw = {key: before[key] for key in _FIELDS}
-    before = _merge(row.value if row is not None else None)
+    raw["video_url"] = video_url
-    new_value = {k: before[k] for k in _FIELDS}
+    if video_url:
-    new_value["video_url"] = video_url
+        if not analysis or analysis.get("analysis_status") != "valid":
-    after = _write(db, new_value, scene=scene, admin_id=admin_id, commit=commit)
+            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, scene: str = "coupon") -> int:
-    """该账号已用掉的引导视频次数(开播即算,含未发币的)。"""
     return int(
         db.execute(
             select(func.count()).select_from(GuideVideoPlay).where(
                 GuideVideoPlay.user_id == user_id,
                 GuideVideoPlay.scene == scene,
+                GuideVideoPlay.status != "prepared",
             )
         ).scalar_one()
     )
 
 
-def play_stats(db: Session, scene: str = "coupon") -> dict[str, int]:
+def _prepare_miss(scene: str, reason: str, cfg: dict[str, Any], used: int) -> dict[str, Any]:
-    """全站播放统计(admin 页展示):总播放次数 / 其中已发币次数。"""
+    maximum = int(cfg.get("max_plays") or 0)
-    total = int(
+    return {
-        db.execute(
+        "should_play": False,
-            select(func.count()).select_from(GuideVideoPlay).where(
+        "reason": reason,
-                GuideVideoPlay.scene == scene
+        "scene": scene,
-            )
+        "video_url": None,
-        ).scalar_one()
+        "play_token": "",
-    )
+        "config_version": int(cfg.get("config_version") or 0),
-    granted = int(
+        "duration_ms": 0,
-        db.execute(
+        "circle_count": CIRCLE_COUNT,
-            select(func.count()).select_from(GuideVideoPlay).where(
+        "circle_duration_ms": 0,
-                GuideVideoPlay.status == "granted",
+        "reward_coin": int(cfg.get("reward_coin") or 0),
-                GuideVideoPlay.scene == scene,
+        "reward_per_circle": int(cfg.get("reward_coin") or 0) // CIRCLE_COUNT,
-            )
+        "seq": used,
-        ).scalar_one()
+        "remaining": max(0, maximum - used),
-    )
+        "expires_at": None,
-    return {"total_plays": total, "granted_plays": granted}
+    }
 
 
-def start_play(
+def prepare_play(db: Session, user_id: int, *, scene: str = "coupon") -> dict[str, Any]:
-    db: Session, user_id: int, *, scene: str = "coupon", commit: bool = True
-) -> dict[str, Any]:
-    """决定这次浮层是否放引导视频;命中则**当场计次**并返回 play_token。
-
-    返回 dict:
-      should_play  是否放引导视频(False → 客户端照旧放广告)
-      video_url    相对地址(/media/...);客户端自行拼 BASE_URL
-      play_token   发币幂等键(should_play=False 时为空串)
-      reward_coin  播完/中途关闭都发的固定金币
-      seq / remaining  第几次 / 发完这次还剩几次(仅展示与排查用)
-    """
-    _config_key(scene)
     cfg = get_config(db, scene)
-    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, 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 _miss(used_now: int) -> dict[str, Any]:
+    now = _now()
-        return {
-            "should_play": False,
-            "video_url": None,
-            "play_token": "",
-            "reward_coin": reward_coin,
-            "seq": used_now,
-            "remaining": max(0, max_plays - used_now),
-        }
-
-    if not cfg.get("enabled") or not video_url or max_plays <= 0 or used >= max_plays:
-        return _miss(used)
-
-    seq = used + 1
     play = GuideVideoPlay(
         user_id=user_id,
         play_token=uuid.uuid4().hex,
         scene=scene,
-        seq=seq,
+        seq=None,
         video_url=video_url,
-        coin=reward_coin,
+        coin=int(cfg["reward_coin"]),
-        status="playing",
+        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。
+    db.commit()
-    # (user_id, seq) 唯一键让只有一个能落库,其余撞键 → 回滚后按"这次不放视频"降级,
-    # 客户端照旧走广告链路。没有它,并发就能绕过 max_plays 无限刷金币。
-    try:
-        if commit:
-            db.commit()
-        else:
-            db.flush()
-    except IntegrityError:
-        db.rollback()
-        return _miss(used_plays(db, user_id, scene))
     return {
         "should_play": True,
+        "reason": "ready",
+        "scene": scene,
         "video_url": video_url,
         "play_token": play.play_token,
-        "reward_coin": reward_coin,
+        "config_version": play.config_version,
-        "seq": seq,
+        "duration_ms": duration,
-        "remaining": max(0, max_plays - seq),
+        "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(
+def _start_out(play: GuideVideoPlay, maximum: int, status: str) -> dict[str, Any]:
-    db: Session, user_id: int, *, play_token: str, completed: bool
+    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 表示**本次调用真的入账了**;
+    final = circle == CIRCLE_COUNT
-    重复上报返回 granted=False + 已发金币(客户端据此不重复累加 toast 金额)。
-    """
-    token = (play_token or "").strip()
-    # 金币额度以**服务端配置**为准,不信客户端(客户端只上报"播完/关闭")。
-    play = _find_play(db, user_id, token)
-    coin = int(play.coin if play is not None else 0)
-
-    # 幂等核心:把 status 放进 WHERE 做条件更新(compare-and-set),而不是"先读再判再写"。
-    # 「播完」与「✕ 关闭」抢跑、或客户端超时重试时,两个请求会都读到 status='playing',
-    # 无锁的话就都往下发币、都 commit,金币入账两次(不用恶意,重试就会中招)。改成条件更新后
-    # 并发里只有一条 rowcount=1,另一条拿 0 → 按已发返回,不二次铸币。
-    # (PG READ COMMITTED 下后到的 UPDATE 阻塞到对手提交,再按新版本重判 status;SQLite 写串行。)
-    #
-    # 别指望 IntegrityError 兜底:这里只 UPDATE 不 INSERT,撞不到 uq_guide_video_play_token;
-    # 而 biz_type='guide_video' 的金币流水也不在 ux_coin_transaction_task_ref 的谓词
-    # (biz_type LIKE 'task%')覆盖范围内 —— 两个唯一键在这条路径上都是不生效的。
     won = db.execute(
         update(GuideVideoPlay)
         .where(
-            GuideVideoPlay.play_token == token,
+            GuideVideoPlay.id == play.id,
-            GuideVideoPlay.user_id == user_id,
+            GuideVideoPlay.status == "started",
-            GuideVideoPlay.status == "playing",
+            GuideVideoPlay.settled_circles == settled,
         )
         .values(
-            status="granted",
+            settled_circles=circle,
-            coin=coin,
+            status="completed" if final else "started",
-            completed=1 if completed else 0,
+            completed=1 if final else 0,
-            granted_at=datetime.now(rewards.CN_TZ).replace(tzinfo=None),
+            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)
+        fresh = _find_play(db, user_id, play_token)
-        if play is None:
+        fresh_settled = int(fresh.settled_circles if fresh else 0)
-            return {"granted": False, "coin": 0, "status": "not_found"}
+        return {
-        return {"granted": False, "coin": play.coin, "status": "already_granted"}
+            **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:
+    per_circle = play.coin // CIRCLE_COUNT
-        crud_wallet.grant_coins(
+    account, _ = crud_wallet.grant_coins(
-            db,
+        db,
-            user_id,
+        user_id,
-            coin,
+        per_circle,
-            biz_type=BIZ_TYPE,
+        biz_type=BIZ_TYPE,
-            ref_id=token,
+        ref_id=f"{play.play_token}:{circle}",
-            remark="新手引导视频奖励",
+        remark=f"新手引导视频第{circle}圈奖励",
-        )
+    )
-    db.commit()
+    try:
-    return {"granted": True, "coin": coin, "status": "granted"}
+        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

app/repositories/wallet.py

@@ -132,10 +132,20 @@ def get_or_create_account(
 ) -> CoinAccount:
     """取用户金币账户,不存在则建一个空账户。
 
-    lock=True 时对已存在的账户行加 SELECT FOR UPDATE(读-算-写余额的调用方串行化,防并发
+    lock=True 时对已存在的账户行加 SELECT FOR UPDATE，并强制刷新 identity map 中可能
-    双写余额错位,如 admin set 模式连点);默认 False 不改 C 端发奖行为。SQLite 下为 no-op。
+    缓存的旧余额；所有金币写入口都会使用它。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
 

app/schemas/guide_video.py

@@ -1,38 +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(领券浮层);预留给日后比价等场景。"""
+    play_token: str = Field(min_length=1, max_length=64)
-
-    scene: Literal["coupon", "comparison"] = "coupon"
 
 
 class GuideVideoStartOut(BaseModel):
-    """should_play=False 时客户端照旧走广告链路,其余字段无意义。"""
+    started: bool
-
+    status: Literal["started", "already_started"]
-    should_play: bool
+    play_token: str
-    video_url: str | None = None   # 相对地址 /media/...;客户端自行拼 BASE_URL
+    scene: GuideScene
-    play_token: str = ""           # 发奖幂等键
+    video_url: str
-    reward_coin: int = 0           # 播完/中途关闭都发的固定金币
+    config_version: int
-    seq: int = 0                   # 本账号第几次
+    duration_ms: int
-    remaining: int = 0             # 发完这次还剩几次
+    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: Literal[
-    status: str
+        "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

app/services/notification_events.py

@@ -136,17 +136,9 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
                 )
                 continue
             try:
-                response = vendor_push.send_notification(
+                vendor_push.send_notification(
                     vendor, dev.push_token, title=title, body=body, extras=extras
                 )
-                if vendor == "huawei" and row.type in {"feedback_reply", "feedback_reward"}:
-                    # 华为反馈推送联调日志：保留厂商返回码/requestId 等排障信息，
-                    # 请求本身的 token、Authorization 和应用密钥不会进入 response。
-                    logger.info(
-                        "huawei feedback push response user_id=%s type=%s "
-                        "notification_id=%s response=%s",
-                        row.user_id, row.type, row.id, response,
-                    )
                 logger.info(
                     "push sent user_id=%s type=%s vendor=%s notification_id=%s",
                     row.user_id, row.type, vendor, row.id,
@@ -155,19 +147,6 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
                 logger.warning(
                     "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
                 )
-            try:
-                vendor_push.send_data_event(
-                    vendor,
-                    dev.push_token,
-                    event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
-                    notification_id=str(row.id),
-                )
-            except vendor_push.VendorPushError as e:
-                # 透传只负责前台铃铛实时刷新，失败不能影响通知栏消息或站内消息。
-                logger.warning(
-                    "push data event failed user_id=%s type=%s vendor=%s: %s",
-                    row.user_id, row.type, vendor, e,
-                )
     except Exception:  # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
         logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
 

tests/test_push_center.py

@@ -7,9 +7,7 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
 from __future__ import annotations
 
 import json
-import logging
 
-import httpx
 import pytest
 from fastapi.testclient import TestClient
 
@@ -112,99 +110,6 @@ def test_huawei_non_success_code_raises(monkeypatch) -> None:
         vendor_push.send_notification("huawei", "bad-token", title="t", body="b")
 
 
-def test_vendor_http_logs_device_token_raw_response_and_elapsed(monkeypatch, caplog) -> None:
-    vendor_push._token_cache.clear()
-
-    def _fake_request(method, url, **kwargs):  # noqa: ANN001
-        if url == vendor_push.settings.HUAWEI_PUSH_TOKEN_ENDPOINT:
-            return _Resp({"access_token": "raw-access-token", "expires_in": 3600})
-        return _Resp({"code": "80000000", "msg": "Success", "requestId": "vendor-request-1"})
-
-    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_ID", "10086001")
-    monkeypatch.setattr(vendor_push.settings, "HUAWEI_PUSH_APP_SECRET", "raw-app-secret")
-    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
-
-    with caplog.at_level(logging.INFO, logger=vendor_push.logger.name):
-        vendor_push.send_notification(
-            "huawei",
-            "raw-device-token",
-            title="需要进入日志的标题",
-            body="需要进入日志的正文",
-            extras={"type": "withdraw_success", "notificationId": "90001"},
-        )
-
-    messages = "\n".join(record.getMessage() for record in caplog.records)
-    assert "operation=authenticate" in messages
-    assert "operation=send_notification" in messages
-    assert "request=" in messages
-    assert "response=" in messages
-    assert "elapsed_ms=" in messages
-    assert "vendor-request-1" in messages
-    assert "withdraw_success" in messages
-    assert "raw-access-token" in messages
-    assert "raw-device-token" in messages
-    assert "需要进入日志的标题" in messages
-    assert "需要进入日志的正文" in messages
-    assert "<redacted" in messages
-    assert "raw-app-secret" not in messages
-    send_http_log = next(
-        record.getMessage()
-        for record in caplog.records
-        if "operation=send_notification" in record.getMessage()
-        and "vendor push http completed" in record.getMessage()
-    )
-    assert "raw-device-token" in send_http_log
-    assert "raw-access-token" not in send_http_log  # 请求头中的厂商鉴权 token 仍脱敏
-
-
-def test_vendor_http_network_error_logs_request_context(monkeypatch, caplog) -> None:
-    def _fake_request(*args, **kwargs):  # noqa: ANN002, ANN003
-        raise httpx.ConnectError("connection refused")
-
-    monkeypatch.setattr(vendor_push.settings, "XIAOMI_PUSH_APP_SECRET", "raw-xiaomi-secret")
-    monkeypatch.setattr(vendor_push.httpx, "request", _fake_request)
-
-    with (
-        caplog.at_level(logging.ERROR, logger=vendor_push.logger.name),
-        pytest.raises(vendor_push.VendorPushError, match="push http error"),
-    ):
-        vendor_push.send_notification(
-            "xiaomi",
-            "raw-xiaomi-token",
-            title="title",
-            body="body",
-            extras={"type": "push_test"},
-        )
-
-    messages = "\n".join(record.getMessage() for record in caplog.records)
-    assert "outcome=network_error" in messages
-    assert "response=<no_response>" in messages
-    assert "operation=send_notification" in messages
-    assert "elapsed_ms=" in messages
-    assert "raw-xiaomi-secret" not in messages
-    assert "raw-xiaomi-token" in messages
-
-
-@pytest.mark.parametrize(
-    ("request_kwargs", "device_token"),
-    [
-        ({"json": {"token": ["honor-device-token"]}}, "honor-device-token"),
-        ({"json": {"message": {"token": ["huawei-device-token"]}}}, "huawei-device-token"),
-        ({"json": {"regId": "vivo-device-token"}}, "vivo-device-token"),
-        ({"data": {"registration_id": "xiaomi-device-token"}}, "xiaomi-device-token"),
-        (
-            {"data": {"message": '{"target_value":"oppo-device-token"}'}},
-            "oppo-device-token",
-        ),
-    ],
-)
-def test_all_vendor_device_token_fields_remain_visible_in_request_summary(
-    request_kwargs, device_token
-) -> None:
-    summary = vendor_push._request_summary(request_kwargs)
-    assert device_token in summary
-
-
 def test_vendor_aliases_normalize() -> None:
     assert vendor_push.normalize_vendor("华为") == "huawei"
     assert vendor_push.normalize_vendor("HMS") == "huawei"