Merge branch 'main' into fix/ad-reward-min-one-coin

背景
短信验证码已是可切换 provider 架构(极光 / 创蓝)。极光(默认)一旦供应商侧故障(欠费 / 网络 / 服务异常),/sms/send 直接 503 → 用户收不到码、登录中断。本 PR 把极光设为主、创蓝设为备,在极光供应商不可用时自动转创蓝补发,并让后台可区分每次实际走的渠道。

方案(4 个关键决策)
# | 决策 | 结论
-- | -- | --
A | fallback 触发范围 | 仅主返回「供应商不可用」(SmsError.status_code == 503:网络 / 余额 / 服务故障)才转备。本地冷却 & 超频(429)、手机号无效(400)不转——不绕过防刷、不为无效号白烧
B | 校验路由 | try-both:极光转创蓝后码在创蓝内存,校验遍历「启用的 fallback 链」(主→备),任一命中即通过;关闭 fallback 时链中只有极光、创蓝零参与
C | 后台可见性 | 成功侧 EVENT_SMS_SEND.details 记 provider / fallback + 分派层日志,风控后台可按号/设备查本次走哪家、是否 fallback
D | 默认开关 | SMS_FALLBACK_PROVIDER 默认空=关(保持现状零风险),生产设 chuanglan 开启,置空即秒回退。仅 Mode B(jiguang/chuanglan)互为主备

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #204

.env.example

@@ -45,6 +45,9 @@ SMS_SEND_INTERVAL_SEC=60
 # chuanglan:创蓝云智(253)模板短信,本服务生成码、创蓝只下发、本地校验(Mode B,与极光同);缺凭证 503。
 #            用 YZM 前缀验证码账号;服务器出网 IP 需在创蓝控制台加白名单(否则 117)。见 docs/integrations/chuanglan/tpl-send.md。
 SMS_PROVIDER=jiguang
+# 极光→创蓝 容灾:主 provider(SMS_PROVIDER)供应商不可用(503)时转投的备。
+# 空=关闭 fallback(默认);生产设 chuanglan 开启极光主、创蓝备;置空即秒回退。
+SMS_FALLBACK_PROVIDER=
 ALIYUN_SMS_ACCESS_KEY_ID=
 ALIYUN_SMS_ACCESS_KEY_SECRET=
 ALIYUN_SMS_SIGN_NAME=
@@ -175,3 +178,9 @@ 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/6d2309208549_merge_comparison_platforms_fail_reason_.py

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

alembic/versions/comparison_platforms_col.py

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

alembic/versions/comparison_record_fail_reason.py

@@ -0,0 +1,49 @@
+"""comparison_record.fail_reason (失败卡展示原因)
+
+Revision ID: comparison_record_fail_reason
+Revises: user_manual_risk_fields
+Create Date: 2026-07-28 12:00:00.000000
+
+失败记录的展示原因:information 具体则=它;笼统则由写路径从 platform_results 捞出的
+业务原因;纯系统失败为 None(端侧品牌兜底)。见 repositories.comparison._derive_fail_display。
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'comparison_record_fail_reason'
+down_revision: Union[str, Sequence[str], None] = 'user_manual_risk_fields'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('fail_reason', sa.String(length=256), nullable=True))
+    # 回填老失败记录:information 具体的直接搬过来(笼统/系统失败留 None → 端侧品牌兜底)。
+    # 新记录由写路径 _derive_fail_display 落库(含 platform_results 救援/补判),不走这条。
+    # platform_results 只在 raw_payload 里,SQL 里不易解析,故老记录不做救援/补判(可接受:
+    # 老 mixed/打烊记录回退品牌兜底);具体 information 的老记录本次即可显示真实原因。
+    op.execute(
+        """
+        UPDATE comparison_record
+        SET fail_reason = information
+        WHERE status = 'failed'
+          AND information IS NOT NULL
+          AND information <> ''
+          AND information NOT IN (
+              '比价过程出错，请稍后重试',
+              '比价出错',
+              '比价未完成',
+              'done 参数缺少可验证的目标平台结果'
+          )
+        """
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
+        batch_op.drop_column('fail_reason')

alembic/versions/guide_video_scene_unique.py

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

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 本模块,两进程互不影响。
 """

app/admin/repositories/ad_revenue.py

@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
 
 
-def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def business_code_ids(db: Session, app_env: str | None) -> set[str]:
     """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
     prod_config = app_config.get_ad_config(db)
     prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
 
     # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
     # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_code_ids: set[str] | None = None
+    business_ids: set[str] | None = None
     if revenue_scope == "business":
-        business_code_ids = _business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_code_ids]
+        business_ids = business_code_ids(db, app_env)
+        events = [e for e in events if e.get("our_code_id") in business_ids]
 
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
             date_from=date_from,
             date_to=date_to,
             app_env=app_env,
-            our_code_ids=business_code_ids,
+            our_code_ids=business_ids,
         )
         if pangle_aggs:
             by_date = {a["date"]: a for a in pangle_aggs}

app/admin/repositories/queries.py

@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
 
 from app.core import rewards
 from app.core.config import settings
+from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
@@ -1257,11 +1258,15 @@ def user_reward_stats(
     date_from: datetime | None = None,
     date_to: datetime | None = None,
     withdraw_source: str | None = None,
+    app_env: str | None = None,
+    revenue_scope: str = "all",
+    feed_scene: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
-    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
-    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
+    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
+    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
+    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
     withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ def user_reward_stats(
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
+    business_ids: set[str] | None = None
+    if revenue_scope == "business":
+        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
+        from app.admin.repositories.ad_revenue import business_code_ids
+
+        business_ids = business_code_ids(db, app_env)
+
+    rv_conds = [
+        AdRewardRecord.user_id == user_id,
+        AdRewardRecord.reward_scene == "reward_video",
+        AdRewardRecord.status == "granted",
+        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        rv_conds.append(AdRewardRecord.app_env == app_env)
+    if business_ids is not None:
+        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
     rv = db.execute(
         select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            AdRewardRecord.user_id == user_id,
-            AdRewardRecord.reward_scene == "reward_video",
-            AdRewardRecord.status == "granted",
-            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
+            *rv_conds,
         )
     ).all()
     rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
     rv_coins = sum(r.coin for r in rv)
 
-    feed = db.execute(
+    feed_reward_conds = [
+        AdFeedRewardRecord.user_id == user_id,
+        AdFeedRewardRecord.status == "granted",
+        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
+    feed_rewards = db.execute(
         select(
             AdFeedRewardRecord.unit_count,
-            AdFeedRewardRecord.ecpm_raw,
             AdFeedRewardRecord.coin,
         ).where(
-            AdFeedRewardRecord.user_id == user_id,
-            AdFeedRewardRecord.status == "granted",
-            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
+            *feed_reward_conds,
         )
     ).all()
-    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
-    feed_coins = sum(f.coin for f in feed)
+    feed_coins = sum(f.coin for f in feed_rewards)
+
+    feed_impression_conds = [
+        AdEcpmRecord.user_id == user_id,
+        AdEcpmRecord.ad_type.in_(("draw", "feed")),
+        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
+    ]
+    if app_env is not None:
+        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
+    if feed_scene is not None:
+        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
+    if business_ids is not None:
+        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
+    feed_impressions = db.execute(
+        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
+    ).all()
+    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
+    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
 
     trad_coins = db.execute(
         select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,7 @@ def user_reward_stats(
         "reward_video_count": len(rv),
         "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
         "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed)),
+        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
         "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
         "feed_cash_cents": _coins_to_cents(feed_coins),
     }

app/admin/routers/feedback.py

@@ -1,6 +1,7 @@
 """admin 反馈工单:列表筛选 + 审核采纳/拒绝(带金币发放与审计)。"""
 from __future__ import annotations
 
+import logging
 from datetime import datetime
 from typing import Annotated
 
@@ -25,6 +26,8 @@ from app.models.feedback import Feedback
 from app.repositories import wallet as wallet_repo
 from app.services import notification_events
 
+logger = logging.getLogger("shagua.admin.feedback")
+
 router = APIRouter(
     prefix="/admin/api/feedbacks",
     tags=["admin-feedback"],
@@ -46,10 +49,24 @@ def _approve_feedback(
     *,
     bulk: bool = False,
 ) -> FeedbackOut:
+    logger.info(
+        "feedback approve started feedback_id=%s admin_id=%s bulk=%s",
+        feedback_id, admin.id, bulk,
+    )
     fb = db.get(Feedback, feedback_id, with_for_update=True)
     if fb is None:
+        logger.warning(
+            "feedback approve rejected not found feedback_id=%s admin_id=%s bulk=%s",
+            feedback_id, admin.id, bulk,
+        )
         raise HTTPException(status_code=404, detail="反馈不存在")
-    _ensure_pending(fb)
+    if fb.status not in {"pending", "new"}:
+        logger.warning(
+            "feedback approve rejected invalid status feedback_id=%s user_id=%s "
+            "admin_id=%s status=%s bulk=%s",
+            feedback_id, fb.user_id, admin.id, fb.status, bulk,
+        )
+        _ensure_pending(fb)
 
     before = fb.status
     mutations.review_feedback(
@@ -91,8 +108,18 @@ def _approve_feedback(
     )
     db.commit()
     db.refresh(fb)
+    logger.info(
+        "feedback approve committed feedback_id=%s user_id=%s admin_id=%s "
+        "before=%s after=%s reward_coins=%s bulk=%s",
+        feedback_id, fb.user_id, admin.id, before, fb.status, payload.reward_coins, bulk,
+    )
     out = FeedbackOut.model_validate(fb)
     notification_events.notify_feedback_reward(db, fb)
+    logger.info(
+        "feedback approve notification dispatch returned feedback_id=%s user_id=%s "
+        "admin_id=%s notification_type=feedback_reward bulk=%s",
+        feedback_id, fb.user_id, admin.id, bulk,
+    )
     return out
 
 
@@ -105,10 +132,24 @@ def _reject_feedback(
     *,
     bulk: bool = False,
 ) -> FeedbackOut:
+    logger.info(
+        "feedback reject started feedback_id=%s admin_id=%s bulk=%s",
+        feedback_id, admin.id, bulk,
+    )
     fb = db.get(Feedback, feedback_id, with_for_update=True)
     if fb is None:
+        logger.warning(
+            "feedback reject rejected not found feedback_id=%s admin_id=%s bulk=%s",
+            feedback_id, admin.id, bulk,
+        )
         raise HTTPException(status_code=404, detail="反馈不存在")
-    _ensure_pending(fb)
+    if fb.status not in {"pending", "new"}:
+        logger.warning(
+            "feedback reject rejected invalid status feedback_id=%s user_id=%s "
+            "admin_id=%s status=%s bulk=%s",
+            feedback_id, fb.user_id, admin.id, fb.status, bulk,
+        )
+        _ensure_pending(fb)
 
     before = fb.status
     mutations.review_feedback(
@@ -142,8 +183,18 @@ def _reject_feedback(
     )
     db.commit()
     db.refresh(fb)
+    logger.info(
+        "feedback reject committed feedback_id=%s user_id=%s admin_id=%s "
+        "before=%s after=%s bulk=%s",
+        feedback_id, fb.user_id, admin.id, before, fb.status, bulk,
+    )
     out = FeedbackOut.model_validate(fb)
     notification_events.notify_feedback_reply(db, fb)
+    logger.info(
+        "feedback reject notification dispatch returned feedback_id=%s user_id=%s "
+        "admin_id=%s notification_type=feedback_reply bulk=%s",
+        feedback_id, fb.user_id, admin.id, bulk,
+    )
     return out
 
 
@@ -201,6 +252,10 @@ def bulk_approve_feedbacks(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackBulkResult:
+    logger.info(
+        "feedback bulk approve started admin_id=%s item_count=%s",
+        admin.id, len(body.ids),
+    )
     results: list[FeedbackBulkItemResult] = []
     ip = get_client_ip(request)
     for feedback_id in body.ids:
@@ -209,11 +264,24 @@ def bulk_approve_feedbacks(
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
         except HTTPException as exc:
             db.rollback()
+            logger.warning(
+                "feedback bulk approve item failed feedback_id=%s admin_id=%s error=%s",
+                feedback_id, admin.id, exc.detail,
+            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
         except Exception:  # noqa: BLE001 - 单笔失败不打断整批
             db.rollback()
+            logger.exception(
+                "feedback bulk approve item failed feedback_id=%s admin_id=%s",
+                feedback_id, admin.id,
+            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
+    result = _bulk_result(results)
+    logger.info(
+        "feedback bulk approve completed admin_id=%s total=%s success=%s failed=%s",
+        admin.id, result.total, result.success, result.failed,
+    )
+    return result
 
 
 @router.post("/bulk/reject", response_model=FeedbackBulkResult, summary="批量拒绝采纳反馈")
@@ -223,6 +291,10 @@ def bulk_reject_feedbacks(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackBulkResult:
+    logger.info(
+        "feedback bulk reject started admin_id=%s item_count=%s",
+        admin.id, len(body.ids),
+    )
     results: list[FeedbackBulkItemResult] = []
     ip = get_client_ip(request)
     for feedback_id in body.ids:
@@ -231,11 +303,24 @@ def bulk_reject_feedbacks(
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=True, status=out.status))
         except HTTPException as exc:
             db.rollback()
+            logger.warning(
+                "feedback bulk reject item failed feedback_id=%s admin_id=%s error=%s",
+                feedback_id, admin.id, exc.detail,
+            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error=str(exc.detail)))
         except Exception:  # noqa: BLE001 - 单笔失败不打断整批
             db.rollback()
+            logger.exception(
+                "feedback bulk reject item failed feedback_id=%s admin_id=%s",
+                feedback_id, admin.id,
+            )
             results.append(FeedbackBulkItemResult(id=feedback_id, ok=False, error="系统异常"))
-    return _bulk_result(results)
+    result = _bulk_result(results)
+    logger.info(
+        "feedback bulk reject completed admin_id=%s total=%s success=%s failed=%s",
+        admin.id, result.total, result.success, result.failed,
+    )
+    return result
 
 
 @router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
@@ -258,7 +343,16 @@ def approve_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    try:
+        return _approve_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    except HTTPException:
+        raise
+    except Exception:
+        logger.exception(
+            "feedback approve failed feedback_id=%s admin_id=%s",
+            feedback_id, admin.id,
+        )
+        raise
 
 
 @router.post("/{feedback_id}/reject", response_model=FeedbackOut, summary="拒绝采纳反馈")
@@ -269,4 +363,13 @@ def reject_feedback(
     admin: Annotated[AdminUser, Depends(require_role("operator"))],
     db: AdminDb,
 ) -> FeedbackOut:
-    return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    try:
+        return _reject_feedback(db, admin, feedback_id, payload, get_client_ip(request))
+    except HTTPException:
+        raise
+    except Exception:
+        logger.exception(
+            "feedback reject failed feedback_id=%s admin_id=%s",
+            feedback_id, admin.id,
+        )
+        raise

app/admin/routers/guide_video.py

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

app/admin/routers/users.py

@@ -88,6 +88,11 @@ def get_user_reward_stats(
     withdraw_source: Annotated[
         str | None, Query(pattern="^(coin_cash|invite_cash)$")
     ] = None,
+    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
+    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
+    feed_scene: Annotated[
+        str | None, Query(pattern="^(comparison|coupon|welfare)$")
+    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if not user_repo.user_exists(db, user_id):
@@ -99,6 +104,9 @@ def get_user_reward_stats(
             date_from=date_from,
             date_to=date_to,
             withdraw_source=withdraw_source,
+            app_env=app_env,
+            revenue_scope=revenue_scope,
+            feed_scene=feed_scene,
         )
     )
 

app/admin/schemas/guide_video.py

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

app/admin/schemas/user.py

@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
     reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
     reward_video_cash_cents: int  # 激励视频提现(金币折现)
     feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
+    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
     feed_cash_cents: int  # 信息流广告提现(金币折现)
 
 

app/api/v1/auth.py

@@ -207,7 +207,8 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
     check_rate_limits(request, subject=req.device_id, rules=send_rules)
 
     try:
-        cooldown = send_code(req.phone)
+        send_result = send_code(req.phone)
+        cooldown = send_result.cooldown_sec
     except SmsError as e:
         risk_repo.record_behavior_event(
             db,
@@ -238,7 +239,9 @@ def sms_send(req: SmsSendRequest, request: Request, db: DbSession) -> SmsSendRes
         phone=req.phone,
         client_ip=_client_ip(request),
         outcome="success",
-        details={"mock": settings.SMS_MOCK},
+        details={"mock": settings.SMS_MOCK,
+                 "provider": send_result.provider,
+                 "fallback": send_result.fallback},
         evaluate_rule=risk_repo.RULE_SMS_HOURLY,
     )
 

app/api/v1/compare.py

@@ -108,7 +108,7 @@ def _harvest_done_blocking(
 
 def _harvest_abort_blocking(
     trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
-) -> None:
+) -> int | None:
     with SessionLocal() as db:
         rec = crud_compare.harvest_abort(
             db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,6 +118,7 @@ def _harvest_abort_blocking(
         extra={"phase": "harvest_abort",
                "status": (rec.status if rec else None), "reason": reason},
     )
+    return rec.id if rec is not None else None
 
 
 async def _forward(
@@ -291,7 +292,10 @@ async def trace_epilogue(
 
 @router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request, user: OptionalUser, db: DbSession
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
 ) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -302,12 +306,16 @@ async def trace_finalize(
         request, "/api/trace/finalize", user, harvest_first_frame=False,
     )
     try:
-        await run_in_threadpool(
+        record_id = await run_in_threadpool(
             _harvest_abort_blocking, trace_id,
             (meta.get("status") or "cancelled"),
             (meta.get("reason") or meta.get("information")),
             (resp.get("trace_url") if isinstance(resp, dict) else None),
         )
+        if record_id is not None:
+            background_tasks.add_task(
+                backfill_comparison_llm_cost, record_id, trace_id
+            )
     except Exception as e:  # noqa: BLE001
         logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/guide_video.py

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

app/core/config.py

@@ -88,6 +88,8 @@ 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"
@@ -146,6 +148,9 @@ class Settings(BaseSettings):
     # aliyun:阿里云 dypns 号码认证,阿里云生成+下发+校验(Mode A);缺凭证时 /sms/* 返 503(优雅降级)。
     # chuanglan:创蓝云智(253)模板短信,本服务生成码、创蓝只下发、本地校验(Mode B,与极光同);缺凭证 503。
     SMS_PROVIDER: Literal["jiguang", "aliyun", "chuanglan"] = "jiguang"
+    # 主 provider 供应商不可用(SmsError 503)时转投的备 provider；空=关闭 fallback(默认，
+    # 保持现状零风险)。仅 Mode B(jiguang/chuanglan)互为主备有意义；生产建议设 "chuanglan"。
+    SMS_FALLBACK_PROVIDER: Literal["jiguang", "chuanglan", ""] = ""
     ALIYUN_SMS_ACCESS_KEY_ID: str = ""
     ALIYUN_SMS_ACCESS_KEY_SECRET: str = ""
     ALIYUN_SMS_SIGN_NAME: str = ""            # 系统赠送签名(自定义签名下发易失败)
@@ -436,6 +441,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:
-    """保存新手引导视频(运营后台上传的运营素材),返回相对 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("视频时长必须在 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"
-    (_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:

app/core/rewards.py

@@ -233,10 +233,16 @@ def calculate_ad_reward_coin(ecpm: str | int | float | None, count_after_this: i
     eCPM 在此先钳到 AD_ECPM_MAX_FEN(¥500 CPM):信息流广告一期 eCPM 由客户端上报,伪造天价值
     会铸天量金币;钳在这唯一入口,feed 与 reward_video 回退客户端 eCPM 的路径都护住,且阈值高于
     所有真实值,不影响正规发奖。
+
+    下限(2026-07):有真实正 eCPM 时单份至少 1 金币——低 eCPM 单份收益四舍五入成 0 时兜底为 1,
+    避免用户看了广告却因数值太小被记 too_short 零发。eCPM 缺失/为 0/非法(没有真实广告价值)仍返 0,
+    不凭空铸币、不破坏 ecpm_missing 语义;防刷上限仍由 AD_ECPM_MAX_FEN 钳顶把守。
     """
     ecpm_yuan = min(parse_ecpm_yuan(ecpm), AD_ECPM_MAX_FEN / 100.0)
+    if ecpm_yuan <= 0:
+        return 0
     yuan = (ecpm_yuan / 1000.0) * ad_ecpm_factor(ecpm_yuan) * ad_lt_factor(count_after_this)
-    return max(0, round(yuan * COIN_PER_YUAN))
+    return max(1, round(yuan * COIN_PER_YUAN))
 
 
 # 签到膨胀(SIGNIN_BOOST_COIN,固定 3000)已于 2026-07 下线:它与广告实际收益脱钩,产品确认

app/integrations/sms/__init__.py

@@ -1,37 +1,72 @@
-"""短信验证码服务 —— provider 分派入口。
+"""短信验证码服务 —— provider 分派入口（极光主 + 可选创蓝备）。
 
-对外只暴露 `send_code` / `verify_code` / `SmsError`,api 层无需关心用哪个 provider。
-provider 由 `settings.SMS_PROVIDER` 选择(**每次调用读取**,支持运行时切换 + 灰度回退):
-  - `jiguang`(默认):自管码(本服务生成、内存存/校验,极光只发)。见 [jiguang.py](jiguang.py)。
-  - `aliyun`:阿里云号码认证(阿里云生成+下发+校验,Mode A)。见 [aliyun.py](aliyun.py)。
-  - `chuanglan`:创蓝云智(253)模板短信,自管码 Mode B(本服务生成、内存存/校验,创蓝只发)。见 [chuanglan.py](chuanglan.py)。
-
-mock(`SMS_MOCK=true`)与各 provider 的行为差异都封在 provider 内部;本层只做路由。
-拆包前本模块是单文件 `sms.py`;拆包后极光逻辑迁入 `jiguang` 子模块,行为零改动。
+对外暴露 `send_code`（返回 SendResult） / `verify_code` / `SmsError` / `SendResult`。
+主 provider = settings.SMS_PROVIDER（默认 jiguang）；备 = settings.SMS_FALLBACK_PROVIDER
+（默认空=无备）。**每次调用读 settings**，支持运行时切换 / 灰度回退。
 """
 from __future__ import annotations
 
+import logging
+
 from app.core.config import settings
 
 from . import aliyun, chuanglan, jiguang
-from .base import SmsError
+from .base import SendResult, SmsError
 
-__all__ = ["SmsError", "send_code", "verify_code"]
+__all__ = ["SmsError", "SendResult", "send_code", "verify_code"]
 
-# provider 名 -> 模块;未知/缺省值回退 jiguang(默认兜底,防误配把登录打挂)。
-_PROVIDERS = {"aliyun": aliyun, "chuanglan": chuanglan}
+logger = logging.getLogger("shagua.sms")
+
+# provider 名 <-> 模块。未知/缺省主 provider 回退 jiguang（防误配把登录打挂）。
+_ALL = {"jiguang": jiguang, "aliyun": aliyun, "chuanglan": chuanglan}
+_NAME = {jiguang: "jiguang", aliyun: "aliyun", chuanglan: "chuanglan"}
 
 
-def _provider():
-    """按配置选 provider 模块(每次调用读 settings,支持运行时切换 / 测试注入)。"""
-    return _PROVIDERS.get(settings.SMS_PROVIDER, jiguang)
+def _primary():
+    """当前主 provider 模块（未知/缺省回退 jiguang）。"""
+    return _ALL.get(settings.SMS_PROVIDER, jiguang)
 
 
-def send_code(phone: str) -> int:
-    """发送验证码,返回距下次可发的冷却秒数;失败抛 SmsError。委托给当前 provider。"""
-    return _provider().send_code(phone)
+def _fallback():
+    """启用的备 provider 模块；未配置 / 同主 / 未知名 → None（= 不启用 fallback）。"""
+    name = (settings.SMS_FALLBACK_PROVIDER or "").strip()
+    if not name or name == settings.SMS_PROVIDER:
+        return None
+    return _ALL.get(name)
+
+
+def send_code(phone: str) -> SendResult:
+    """发码：主成功即返回；仅主「供应商不可用(503)」且配置了备时转备补发。
+
+    429（本地冷却/超频）、400（手机号无效）不转——不绕过防刷、不为无效号白烧。
+    备也失败则抛备的 SmsError。返回 SendResult（cooldown + 实际渠道 + 是否 fallback）。
+    """
+    primary = _primary()
+    fb = _fallback()
+    try:
+        cooldown = primary.send_code(phone)
+        return SendResult(cooldown_sec=cooldown, provider=_NAME[primary], fallback=False)
+    except SmsError as e:
+        if fb is not None and e.status_code == 503:
+            logger.warning("[SMS] primary=%s 不可用(%s)，fallback→%s",
+                           _NAME[primary], e, _NAME[fb])
+            cooldown = fb.send_code(phone)   # 备的冷却/错误码原样透出
+            return SendResult(cooldown_sec=cooldown, provider=_NAME[fb], fallback=True)
+        raise
 
 
 def verify_code(phone: str, code: str) -> bool:
-    """校验验证码,返回是否通过;provider 异常降级抛 SmsError。委托给当前 provider。"""
-    return _provider().verify_code(phone, code)
+    """校验：try-both，遍历「启用的 fallback 链」（主→备），任一命中即 True。
+
+    码只存在实际发码那家（fallback 前主已 pop 掉自己的码），另一家 rec is None 即 False、
+    不误判、不累加其防爆破计数。关闭 fallback 时链中只有主，备完全不参与。
+    """
+    chain = [_primary()]
+    fb = _fallback()
+    if fb is not None:
+        chain.append(fb)
+    for prov in chain:
+        if prov.verify_code(phone, code):   # Mode B：纯本地内存比对，不联网
+            logger.info("[SMS] verify hit provider=%s", _NAME[prov])
+            return True
+    return False

app/integrations/sms/aliyun.py

@@ -35,7 +35,7 @@ _GC_THRESHOLD = 10000                    # 超此阈值,send 时顺手清老于
 
 # 发码错误码 → (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
 _SEND_ERRORS: dict[str, tuple[int, str]] = {
-    "MOBILE_NUMBER_ILLEGAL": (400, "手机号无效"),
+    "MOBILE_NUMBER_ILLEGAL": (400, "请输入有效的手机号"),
     "BUSINESS_LIMIT_CONTROL": (429, "今日发送次数过多,请明天再试"),
     "FREQUENCY_FAIL": (429, "发送过于频繁,请稍后再试"),
 }

app/integrations/sms/base.py

@@ -2,9 +2,12 @@
 
 各 provider(jiguang / aliyun)都 `from .base import SmsError`,api 层也从包入口拿到同一个
 `SmsError` —— 保证无论用哪个 provider,异常类型与 HTTP 码映射语义一致。
+`SendResult` 是分派层 `send_code` 的返回类型,携带 cooldown、实际 provider 及是否走了 fallback。
 """
 from __future__ import annotations
 
+from dataclasses import dataclass
+
 from app.core.config import settings
 
 
@@ -18,6 +21,19 @@ class SmsError(Exception):
         self.status_code = status_code
 
 
+@dataclass(frozen=True)
+class SendResult:
+    """发码结果：距下次可发秒数 + 实际发码的 provider 名 + 是否走了备(fallback)。
+
+    分派层 send_code 的返回类型（provider 各自的 send_code 仍返回 int cooldown，
+    由分派层包装）。auth 层据此把渠道记入风控流水 details。
+    """
+
+    cooldown_sec: int
+    provider: str
+    fallback: bool = False
+
+
 def mock_verify(code: str) -> bool:
     """mock 模式校验:放行任意 SMS_CODE_LENGTH 位数字(provider 无关,测试/开发便利,不真校验)。"""
     return len(code) == settings.SMS_CODE_LENGTH and code.isdigit()

app/integrations/sms/chuanglan.py

@@ -53,7 +53,7 @@ _GC_THRESHOLD = 10000  # 任一内存 dict 超此阈值,send 时顺手清过期
 # 发码错误码(创蓝 `code`)→ (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
 _SEND_ERRORS: dict[str, tuple[int, str]] = {
     "103": (429, "发送过于频繁，请稍后再试"),   # 提交速度过快
-    "107": (400, "手机号无效"),                 # 手机号码错误
+    "107": (400, "请输入有效的手机号"),                 # 手机号码错误
 }
 # 需运维介入的配置/开通/余额类错误:打 critical 日志(仍归 503)。
 _SEND_CRITICAL_CODES = frozenset({

app/integrations/sms/jiguang.py

@@ -185,5 +185,5 @@ def _send_via_jiguang(phone: str, code: str) -> None:
     if ecode == 50009:  # 极光侧超频
         raise SmsError("发送过于频繁,请稍后再试", status_code=429)
     if ecode == 50006:  # 手机号无效(schema 已挡格式,这里多是空号/停机)
-        raise SmsError("手机号无效", status_code=400)
+        raise SmsError("请输入有效的手机号", status_code=400)
     raise SmsError(f"短信发送失败(code={ecode})", status_code=503)

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
+from urllib.parse import quote, urlsplit
 
 import httpx
 
@@ -31,6 +31,7 @@ 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 → 中文名(测试/配置状态接口展示用)
@@ -69,6 +70,18 @@ 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:
@@ -115,8 +128,16 @@ def send_notification(
 
     if mock:
         logger.info(
-            "[mock push] vendor=%s token=%s... title=%s body=%s extras=%s",
-            vendor, token[:12], title, body, extras,
+            "vendor push mock vendor=%s request=%s",
+            vendor,
+            _log_summary(
+                {
+                    "push_token": token,
+                    "title": title,
+                    "body": body,
+                    "extras": extras,
+                }
+            ),
         )
         return {
             "mock": True,
@@ -133,7 +154,28 @@ def send_notification(
         "xiaomi": _send_xiaomi,
         "oppo": _send_oppo,
     }
-    return dispatch[vendor](token, title, body, extras)
+    started = time.perf_counter()
+    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(
@@ -153,19 +195,188 @@ 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 _vendor_response_failed(vendor: str, data: dict[str, Any]) -> bool:
+    """识别 HTTP 200 中明确的厂商业务失败，避免将失败请求记录成 success。"""
+    if data.get("error") or data.get("success") is False:
+        return True
+    if vendor == "xiaomi":
+        code = data.get("code")
+        result = str(data.get("result", "ok")).lower()
+        return code not in (0, "0", None) or result not in ("ok", "success")
+    if vendor == "oppo" and data.get("code") is not None:
+        return int(data["code"]) != 0
+    if vendor == "vivo" and data.get("result") is not None:
+        return int(data["result"]) != 0
+    if vendor == "honor" and data.get("code") is not None:
+        return int(data["code"]) != 200
+    if vendor == "huawei" and data.get("code") is not None:
+        return str(data["code"]) != "80000000"
+    return False
+
+
 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,
@@ -174,41 +385,85 @@ 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("vendor push http failed url=%s http=%s body=%s", url, resp.status_code, resp.text[:500])
+        logger.error(
+            "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:
-        return resp.json()
+        data = 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
+    vendor_failed = _vendor_response_failed(vendor, data)
+    log = logger.warning if vendor_failed else logger.info
+    log(
+        "vendor push http completed vendor=%s operation=%s method=%s endpoint=%s "
+        "outcome=%s http_status=%s request=%s response=%s elapsed_ms=%.1f",
+        vendor,
+        operation,
+        method,
+        _endpoint_for_log(url),
+        "vendor_error" if vendor_failed else "success",
+        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]:
-    try:
-        resp = httpx.request(
-            method,
-            url,
-            timeout=settings.PUSH_REQUEST_TIMEOUT_SEC,
-            **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
+    return _request_json(
+        method,
+        url,
+        vendor=vendor,
+        operation=operation,
+        expected_status=expected_status,
+        **kwargs,
+    )
 
 
 def _cache_get(key: str) -> str | None:
@@ -234,6 +489,8 @@ 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,
@@ -243,16 +500,22 @@ def _honor_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"honor auth failed: {data}")
+        raise VendorPushError(f"honor auth failed: {_raw_log_summary(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": {
@@ -261,7 +524,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
             "notification": {
                 "title": title,
                 "body": body,
-                "clickAction": {"type": 3},
+                "clickAction": click_action,
                 "importance": "NORMAL",
             },
         },
@@ -270,6 +533,8 @@ 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",
@@ -279,7 +544,28 @@ 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: {data}")
+        raise VendorPushError(f"honor push failed: {_raw_log_summary(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
 
 
@@ -294,6 +580,8 @@ 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,
@@ -303,7 +591,7 @@ def _huawei_access_token() -> str:
     )
     token = data.get("access_token")
     if not token:
-        raise VendorPushError(f"huawei auth failed: {data}")
+        raise VendorPushError(f"huawei auth failed: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), data.get("expires_in"))
 
 
@@ -312,18 +600,24 @@ 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": {"type": 3},
+                    "click_action": click_action,
                     "importance": "NORMAL",
                 },
             },
@@ -333,6 +627,8 @@ 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",
@@ -340,7 +636,32 @@ 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: {data}")
+        raise VendorPushError(f"huawei push failed: {_raw_log_summary(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
 
 
@@ -357,6 +678,8 @@ 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,
@@ -366,10 +689,10 @@ def _vivo_auth_token() -> str:
         headers={"Content-Type": "application/json"},
     )
     if int(data.get("result", -1)) != 0:
-        raise VendorPushError(f"vivo auth failed: {data}")
+        raise VendorPushError(f"vivo auth failed: {_raw_log_summary(data)}")
     token = data.get("authToken")
     if not token:
-        raise VendorPushError(f"vivo auth missing authToken: {data}")
+        raise VendorPushError(f"vivo auth missing authToken: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -385,15 +708,22 @@ 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,
     }
-    # 点击落地:消息中心推送(带 notificationId)→ skipType=4 + skipContent=intent uri,由 vivo
-    # 系统直启 MainActivity 并携带 S. extras(与小米 notify_effect=2 同机制)。不依赖客户端
-    # VivoPushReceiver.onNotificationMessageClicked 里的后台 startActivity——Android 10+ BAL
-    # 会静默拦掉,receiver 路径仅作兜底。无 notificationId 的召回类保持 skipType=1 仅打开首页。
+    # vivo Push SDK 480+ 已不再回调 skipType=3；必须使用 skipType=4 的完整 Intent URI。
+    # URI 同时包含 data/scheme、显式 component 和 S. extras，OriginOS 才会把业务参数
+    # 原样交给 MainActivity（仅靠隐式 deeplink 会打开 App，但可能剥掉 extras）。
     if extras.get("notificationId"):
         payload["skipType"] = 4
-        payload["skipContent"] = _click_intent_uri(extras)
+        payload["skipContent"] = _vivo_click_intent_uri(extras)
     else:
         payload["skipType"] = 1
     if settings.VIVO_PUSH_CATEGORY:
@@ -401,14 +731,37 @@ 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: {data}")
+        raise VendorPushError(f"vivo push failed: {_raw_log_summary(data)}")
     return data
 
 
@@ -446,14 +799,43 @@ 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: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(data)}")
     if str(data.get("result", "ok")).lower() not in ("ok", "success"):
-        raise VendorPushError(f"xiaomi push failed: {data}")
+        raise VendorPushError(f"xiaomi push failed: {_raw_log_summary(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")
+    form = {
+        "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),
+    }
+    if settings.XIAOMI_PUSH_CHANNEL_ID.strip():
+        form["extra.channel_id"] = settings.XIAOMI_PUSH_CHANNEL_ID.strip()
+    data = _request_form(
+        "POST",
+        settings.XIAOMI_PUSH_SEND_ENDPOINT,
+        vendor="xiaomi",
+        operation="send_data_event",
+        data=form,
+        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
 
 
@@ -490,6 +872,24 @@ 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
@@ -522,6 +922,8 @@ 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,
@@ -530,10 +932,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: {data}")
+        raise VendorPushError(f"oppo auth failed: {_raw_log_summary(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: {data}")
+        raise VendorPushError(f"oppo auth missing auth_token: {_raw_log_summary(data)}")
     return _cache_put(cache_key, str(token), 24 * 3600)
 
 
@@ -548,6 +950,11 @@ 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 同款坑)。
@@ -573,6 +980,8 @@ 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),
@@ -580,5 +989,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: {data}")
+        raise VendorPushError(f"oppo push failed: {_raw_log_summary(data)}")
     return data

app/models/comparison.py

@@ -102,12 +102,21 @@ class ComparisonRecord(Base):
     # done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
     # 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
     information: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # 失败卡「原因」行的展示文案(仅 status=failed 时非空):information 具体则=它;笼统则从
+    # platform_results 捞出的业务原因(打烊/未起送/找不到店或菜/单点不配送);纯系统失败为 None
+    # → 端侧显示品牌兜底「网络开小差…」。写路径(harvest_done / upsert_record)落库时派生。
+    # 见 repositories.comparison._derive_fail_display。
+    fail_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
 
     # ===== 明细(JSON,越详细越好)=====
     # 下单菜品 [{name, qty, specs?}]
     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 全量),未来取数兜底

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 再插),
-        # 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=NULL；start 成功才写入 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}>"
         )

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/comparison.py

@@ -52,6 +52,81 @@ def _product_names_from_items(items: list | None) -> str | None:
     return joined[:500] or None
 
 
+# ---- 失败记录的展示文案(记录页失败卡「原因」行)------------------------------
+# information 具体就直出;笼统(_GENERIC_INFO)则从 platform_results 捞一条用户可读的业务
+# 原因;捞不到 → None(端侧显示品牌兜底「网络开小差…」)。pricebot 把 store_closed /
+# no_delivery 漏成了 status=failed,这里按 reason 关键字补判;打烊类 reason 常带一坨脏店名
+# (店名+月售+起送+配送…),统一成简短模板。自动化黑话(搜索失败/读价失败/购物车残留/裸
+# FAILED…)不给用户看 → 归入品牌兜底。
+
+# pricebot 组不出具体原因时的笼统 information(线上统计的大头),一律走品牌兜底。
+_GENERIC_INFO = {
+    "比价过程出错，请稍后重试",
+    "比价出错",
+    "比价未完成",
+    "done 参数缺少可验证的目标平台结果",
+}
+# 干净业务结局 status(直接可信),按展示优先级(越靠前越先选)。
+_BIZ_STATUS_PRIORITY = (
+    "below_minimum",
+    "no_delivery",
+    "store_closed",
+    "items_not_found",
+    "store_not_found",
+)
+
+
+def _store_closed_text(reason: str | None) -> str:
+    """打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
+    r = reason or ""
+    if "暂停营业" in r:
+        state = "暂停营业"
+    elif "休息" in r:
+        state = "休息中"
+    else:
+        state = "已打烊"
+    return f"门店{state}，无法比价"
+
+
+def _target_display_reason(platform_results: dict | None) -> str | None:
+    """从逐平台结果里挑一条"可展示给用户"的失败原因;挑不到返回 None。
+    ① status 命中干净业务结局集 → 直接采信(打烊套模板,其余用 reason);
+    ② 补判 pricebot 漏成 status=failed 的两类:打烊(套模板)、单点不配送(reason 本身干净);
+    自动化黑话(搜索失败/读价失败/购物车残留/裸 FAILED…)一律不展示 → None。"""
+    pr = platform_results or {}
+    targets = [
+        v for v in pr.values() if isinstance(v, dict) and not v.get("is_source")
+    ]
+    for want in _BIZ_STATUS_PRIORITY:  # ① 干净 status 优先
+        for v in targets:
+            if v.get("status") == want:
+                if want == "store_closed":
+                    return _store_closed_text(v.get("reason"))
+                if v.get("reason"):
+                    return v["reason"]
+    for v in targets:  # ② 漏成 failed 的业务结局补判
+        if v.get("status") != "failed":
+            continue
+        reason = (v.get("reason") or "").strip()
+        if any(k in reason for k in ("打烊", "暂停营业", "休息")):
+            return _store_closed_text(reason)
+        if "单点不配送" in reason:
+            return reason
+    return None
+
+
+def _derive_fail_display(
+    information: str | None, platform_results: dict | None
+) -> str | None:
+    """失败记录展示文案:information 具体则直出;笼统则从 platform_results 捞/补判;
+    都拿不到 → None(端侧品牌兜底)。仅在 status=failed 时调用。"""
+    info = (information or "").strip()
+    text = info if (info and info not in _GENERIC_INFO) else _target_display_reason(
+        platform_results
+    )
+    return text[:256] if text else None
+
+
 def _derive(payload: ComparisonRecordIn) -> dict:
     """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
     results = payload.comparison_results
@@ -89,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
@@ -105,6 +181,11 @@ def _derive(payload: ComparisonRecordIn) -> dict:
         "saved_amount_cents": saved_amount_cents,
         "is_source_best": is_source_best,
         "status": status,
+        "fail_reason": (
+            _derive_fail_display(payload.information, _pr)
+            if status == "failed"
+            else None
+        ),
     }
 
 
@@ -117,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,
@@ -134,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,
@@ -206,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 对象。
@@ -254,7 +356,63 @@ 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)
+    if best is None:
+        # pricebot 没标 is_best(如全平台 has_dish_diff「相似替换/仅供参考」→ 不认定权威最低价)
+        # 但仍有有价行 → 兜底取有价行里最低价当参考 best,避免记录级 best_*/saved 整条落 NULL
+        # (否则首页价 0.00 / 记录页无最低红框 / 省额丢失 /「累计发现可省」漏计)。**候选含源**:
+        # 源常年全菜、价可信,源本身最便宜时 best 回落源(saved=0、is_source_best=True),与 _derive
+        # 「全目标缺菜回落源、不虚报省」同一语义;若排除源强选更贵目标,saved 会变负、倒扣「累计
+        # 发现可省」(get_stats 对 saved_amount_cents 求和不带 >0 过滤)。
+        priced = [p for p in rows if p.get("price") is not None]
+        if priced:
+            best = min(priced, key=lambda p: (p["price"], p.get("display_order") or 0))
+    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"),
     }
 
 
@@ -413,12 +571,40 @@ def harvest_done(
     返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
     行不存在(理论上帧0已建;防御)则新建。"""
     results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
-    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
-    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
+    # 展示模型统一数组(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")
+        )
+        if derived["status"] == "failed"
+        else None
+    )
     fields = dict(
         business_type=business_type or "food",
         information=done_params.get("information") or None,
+        fail_reason=fail_reason,
         # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
         # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
         trace_url=trace_url or done_params.get("trace_url"),
@@ -426,6 +612,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,

app/repositories/guide_video.py

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

app/repositories/wallet.py

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

app/schemas/compare_record.py

@@ -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 定——是
@@ -173,8 +180,13 @@ class ComparisonRecordOut(BaseModel):
     skipped_dish_count: int | None = None
     status: str
     information: str | None = None
+    # 失败卡「原因」文案:具体失败给具体原因,纯系统失败为 None(端侧品牌兜底)。见模型 fail_reason。
+    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。

app/schemas/guide_video.py

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

app/services/comparison_llm_backfill.py

@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
                 select(ComparisonRecord.id, ComparisonRecord.trace_id)
                 .where(
                     *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
                     ComparisonRecord.llm_cost_yuan.is_(None),
                 )
                 .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())

app/services/notification_events.py

@@ -85,6 +85,10 @@ def _dispatch(
     push_vars: dict[str, str] | None = None,
 ) -> Notification | None:
     """落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
+    logger.info(
+        "notification dispatch started user_id=%s type=%s dedup_key=%s",
+        user_id, type_key, dedup_key,
+    )
     try:
         row = notif_repo.create_notification(
             db,
@@ -111,6 +115,10 @@ def _dispatch(
             logger.exception("rollback after notification failure also failed")
         return None
 
+    logger.info(
+        "notification created user_id=%s type=%s notification_id=%s dedup_key=%s",
+        user_id, type_key, row.id, dedup_key,
+    )
     _push_to_user_devices(db, row, push_vars)
     return row
 
@@ -124,29 +132,95 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
         extras.update({str(k): str(v) for k, v in (row.extra or {}).items()})
         extras["notificationId"] = str(row.id)
 
-        for dev in device_repo.list_push_targets(db, user_id=row.user_id):
+        targets = device_repo.list_push_targets(db, user_id=row.user_id)
+        logger.info(
+            "push targets resolved user_id=%s type=%s notification_id=%s target_count=%s",
+            row.user_id, row.type, row.id, len(targets),
+        )
+        if not targets:
+            logger.warning(
+                "push skipped no targets user_id=%s type=%s notification_id=%s",
+                row.user_id, row.type, row.id,
+            )
+            return
+
+        sent = failed = skipped = data_sent = data_failed = 0
+        for dev in targets:
             vendor = vendor_push.normalize_vendor(dev.push_vendor)
             if not vendor or vendor not in vendor_push.SUPPORTED_VENDORS:
-                continue
-            if vendor_push.missing_settings(vendor):
-                # 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
-                logger.info(
-                    "skip push (vendor %s not configured) user_id=%s type=%s",
-                    vendor, row.user_id, row.type,
+                skipped += 1
+                logger.warning(
+                    "push target skipped unsupported vendor user_id=%s type=%s "
+                    "notification_id=%s device_id=%s raw_vendor=%s normalized_vendor=%s",
+                    row.user_id, row.type, row.id, dev.device_id, dev.push_vendor, vendor,
                 )
                 continue
+            missing = vendor_push.missing_settings(vendor)
+            if missing:
+                skipped += 1
+                # 本地/测试环境凭据不齐 → 只落站内消息,不发真推送(与 push/vendors 的报缺口径一致)
+                logger.warning(
+                    "push target skipped vendor not configured user_id=%s type=%s "
+                    "notification_id=%s device_id=%s vendor=%s missing_settings=%s",
+                    row.user_id, row.type, row.id, dev.device_id, vendor, missing,
+                )
+                continue
+            logger.info(
+                "push send started user_id=%s type=%s notification_id=%s "
+                "device_id=%s vendor=%s",
+                row.user_id, row.type, row.id, dev.device_id, vendor,
+            )
             try:
-                vendor_push.send_notification(
+                response = 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,
+                    "push sent user_id=%s type=%s vendor=%s notification_id=%s device_id=%s",
+                    row.user_id, row.type, vendor, row.id, dev.device_id,
+                )
+                sent += 1
+            except vendor_push.VendorPushError as e:
+                failed += 1
+                logger.warning(
+                    "push failed user_id=%s type=%s vendor=%s notification_id=%s "
+                    "device_id=%s error=%s",
+                    row.user_id, row.type, vendor, row.id, dev.device_id, e,
+                )
+            try:
+                data_response = vendor_push.send_data_event(
+                    vendor,
+                    dev.push_token,
+                    event=vendor_push.DATA_EVENT_NOTIFICATION_CREATED,
+                    notification_id=str(row.id),
+                )
+                data_sent += 1
+                logger.info(
+                    "push data event completed user_id=%s type=%s vendor=%s "
+                    "notification_id=%s device_id=%s response=%s",
+                    row.user_id, row.type, vendor, row.id, dev.device_id, data_response,
                 )
             except vendor_push.VendorPushError as e:
+                data_failed += 1
+                # 透传只负责前台铃铛实时刷新，失败不能影响通知栏消息或站内消息。
                 logger.warning(
-                    "push failed user_id=%s type=%s vendor=%s: %s", row.user_id, row.type, vendor, e
+                    "push data event failed user_id=%s type=%s vendor=%s notification_id=%s "
+                    "device_id=%s error=%s",
+                    row.user_id, row.type, vendor, row.id, dev.device_id, e,
                 )
+        logger.info(
+            "push dispatch completed user_id=%s type=%s notification_id=%s targets=%s "
+            "sent=%s failed=%s skipped=%s data_sent=%s data_failed=%s",
+            row.user_id, row.type, row.id, len(targets),
+            sent, failed, skipped, data_sent, data_failed,
+        )
     except Exception:  # noqa: BLE001 — 渲染/查设备等意外失败同样不外抛
         logger.exception("push notification failed user_id=%s type=%s", row.user_id, row.type)
 

docs/superpowers/plans/2026-07-30-sms-jiguang-chuanglan-fallback.md

@@ -0,0 +1,639 @@
+# 短信 极光→创蓝 容灾 fallback 实现计划
+
+> **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:** 极光为主、创蓝为备的短信验证码容灾——极光供应商不可用(503)时自动转创蓝补发，发/校验全程后台可区分实际渠道，由配置开关控制、可秒回退。
+
+**Architecture:** 全部逻辑收在分派层 `app/integrations/sms/__init__.py`（provider 文件 jiguang/chuanglan/aliyun 一行不动）。`send_code` 返回携带渠道的 `SendResult`；`verify_code` 用 try-both 遍历「启用的 fallback 链」。开关 `SMS_FALLBACK_PROVIDER` 默认空=关。
+
+**Tech Stack:** FastAPI + pydantic-settings + pytest(monkeypatch)。SMS provider 均 Mode B(本地内存自管码)。
+
+设计依据：`docs/superpowers/specs/2026-07-30-sms-jiguang-chuanglan-fallback-design.md`
+
+---
+
+## 文件结构
+
+| 文件 | 职责 | 动作 |
+|---|---|---|
+| `app/integrations/sms/base.py` | 加 `SendResult` 结果类型（provider 无关，与 `SmsError` 同层） | 修改 |
+| `app/core/config.py` | 加 `SMS_FALLBACK_PROVIDER` 开关 | 修改 |
+| `app/integrations/sms/__init__.py` | 分派层：主/备解析 + `send_code` fallback + `verify_code` try-both | 修改（核心） |
+| `app/api/v1/auth.py` | `sms_send` 读 `SendResult.cooldown_sec` + `details` 记 `provider`/`fallback` | 修改（一处） |
+| `.env.example` | 加 `SMS_FALLBACK_PROVIDER=` 注释 | 修改 |
+| `tests/test_sms_dispatch.py` | 现有 3 处 `== 60` 断言改读 `.cooldown_sec` | 修改 |
+| `tests/test_sms_fallback.py` | fallback + try-both 全量新测试 | 新建 |
+| `tests/test_auth.py` | `sms_send` 成功 `details` 含 provider/fallback | 修改（加 1 用例） |
+
+`jiguang.py` / `chuanglan.py` / `aliyun.py` **不动**。
+
+---
+
+## Task 1: `SendResult` 结果类型（base.py）
+
+**Files:**
+- Modify: `app/integrations/sms/base.py`
+- Test: `tests/test_sms_base.py`（新建）
+
+- [ ] **Step 1: 写失败测试**
+
+Create `tests/test_sms_base.py`:
+
+```python
+"""SMS 共享基座类型 SendResult。"""
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from app.integrations.sms.base import SendResult
+
+
+def test_send_result_fields():
+    r = SendResult(cooldown_sec=60, provider="jiguang", fallback=True)
+    assert r.cooldown_sec == 60
+    assert r.provider == "jiguang"
+    assert r.fallback is True
+
+
+def test_send_result_fallback_defaults_false():
+    r = SendResult(cooldown_sec=60, provider="jiguang")
+    assert r.fallback is False
+
+
+def test_send_result_is_frozen():
+    r = SendResult(cooldown_sec=60, provider="jiguang")
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        r.provider = "chuanglan"
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_sms_base.py -q`
+Expected: FAIL — `ImportError: cannot import name 'SendResult'`
+
+- [ ] **Step 3: 最小实现**
+
+在 `app/integrations/sms/base.py` 顶部 import 区加 `from dataclasses import dataclass`，并在 `SmsError` 类后新增：
+
+```python
+@dataclass(frozen=True)
+class SendResult:
+    """发码结果：距下次可发秒数 + 实际发码的 provider 名 + 是否走了备(fallback)。
+
+    分派层 send_code 的返回类型（provider 各自的 send_code 仍返回 int cooldown，
+    由分派层包装）。auth 层据此把渠道记入风控流水 details。
+    """
+
+    cooldown_sec: int
+    provider: str
+    fallback: bool = False
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_sms_base.py -q`
+Expected: PASS (3 passed)
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add app/integrations/sms/base.py tests/test_sms_base.py
+git commit -m "feat(sms): 加 SendResult 结果类型(cooldown+provider+fallback)"
+```
+
+---
+
+## Task 2: `SMS_FALLBACK_PROVIDER` 配置开关（config.py）
+
+**Files:**
+- Modify: `app/core/config.py`（`SMS_PROVIDER` 定义附近，约 line 150）
+- Test: `tests/test_sms_fallback.py`（新建，本 Task 起逐步补全）
+
+- [ ] **Step 1: 写失败测试**
+
+Create `tests/test_sms_fallback.py`:
+
+```python
+"""极光→创蓝 容灾 fallback（分派层 send_code / verify_code）。
+
+monkeypatch 各 provider 的 send_code/verify_code 为标记函数，不发真网络；
+断言：仅主 503 才转备；429/400 不转；verify try-both 只遍历启用的链。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.core.config import settings
+from app.integrations import sms
+from app.integrations.sms import aliyun, chuanglan, jiguang
+from app.integrations.sms.base import SendResult, SmsError
+
+PHONE = "13800138000"
+
+
+def _raiser(exc: Exception):
+    """返回一个调用即 raise exc 的 send_code 替身（lambda 不能 raise）。"""
+    def _f(phone):
+        raise exc
+    return _f
+
+
+def test_fallback_provider_defaults_empty():
+    assert settings.SMS_FALLBACK_PROVIDER == ""
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_sms_fallback.py::test_fallback_provider_defaults_empty -q`
+Expected: FAIL — `AttributeError: 'Settings' object has no attribute 'SMS_FALLBACK_PROVIDER'`
+
+- [ ] **Step 3: 最小实现**
+
+在 `app/core/config.py` 的 `SMS_PROVIDER` 定义行下方新增：
+
+```python
+    # 主 provider 供应商不可用(SmsError 503)时转投的备 provider；空=关闭 fallback(默认，
+    # 保持现状零风险)。仅 Mode B(jiguang/chuanglan)互为主备有意义；生产建议设 "chuanglan"。
+    SMS_FALLBACK_PROVIDER: Literal["jiguang", "chuanglan", ""] = ""
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_sms_fallback.py::test_fallback_provider_defaults_empty -q`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add app/core/config.py tests/test_sms_fallback.py
+git commit -m "feat(sms): 加 SMS_FALLBACK_PROVIDER 开关(默认空=关)"
+```
+
+---
+
+## Task 3: 分派层 `send_code` 返回 SendResult（主路径）
+
+先让主路径返回 `SendResult`（不含 fallback），并把现有 `test_sms_dispatch.py` 的 `== 60` 断言迁到 `.cooldown_sec`。fallback 分支在 Task 4 加。
+
+**Files:**
+- Modify: `app/integrations/sms/__init__.py`
+- Modify: `tests/test_sms_dispatch.py`（line 20/22/24、line 45-53）
+- Test: `tests/test_sms_fallback.py`
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_sms_fallback.py` 追加：
+
+```python
+def test_send_code_primary_success_returns_sendresult(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
+    monkeypatch.setattr(jiguang, "send_code", lambda phone: 60)
+
+    r = sms.send_code(PHONE)
+    assert isinstance(r, SendResult)
+    assert r.cooldown_sec == 60
+    assert r.provider == "jiguang"
+    assert r.fallback is False
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_sms_fallback.py::test_send_code_primary_success_returns_sendresult -q`
+Expected: FAIL — `assert isinstance(r, SendResult)`（现 send_code 返回 int 60）
+
+- [ ] **Step 3: 最小实现**
+
+把 `app/integrations/sms/__init__.py` 整体替换为（本 Task 只到主路径；Task 4/5 再扩 fallback 与 try-both）：
+
+```python
+"""短信验证码服务 —— provider 分派入口（极光主 + 可选创蓝备）。
+
+对外暴露 `send_code`（返回 SendResult） / `verify_code` / `SmsError` / `SendResult`。
+主 provider = settings.SMS_PROVIDER（默认 jiguang）；备 = settings.SMS_FALLBACK_PROVIDER
+（默认空=无备）。**每次调用读 settings**，支持运行时切换 / 灰度回退。
+"""
+from __future__ import annotations
+
+import logging
+
+from app.core.config import settings
+
+from . import aliyun, chuanglan, jiguang
+from .base import SendResult, SmsError
+
+__all__ = ["SmsError", "SendResult", "send_code", "verify_code"]
+
+logger = logging.getLogger("shagua.sms")
+
+# provider 名 <-> 模块。未知/缺省主 provider 回退 jiguang（防误配把登录打挂）。
+_ALL = {"jiguang": jiguang, "aliyun": aliyun, "chuanglan": chuanglan}
+_NAME = {jiguang: "jiguang", aliyun: "aliyun", chuanglan: "chuanglan"}
+
+
+def _primary():
+    """当前主 provider 模块（未知/缺省回退 jiguang）。"""
+    return _ALL.get(settings.SMS_PROVIDER, jiguang)
+
+
+def _fallback():
+    """启用的备 provider 模块；未配置 / 同主 / 未知名 → None（= 不启用 fallback）。"""
+    name = (settings.SMS_FALLBACK_PROVIDER or "").strip()
+    if not name or name == settings.SMS_PROVIDER:
+        return None
+    return _ALL.get(name)
+
+
+def send_code(phone: str) -> SendResult:
+    """发码：主成功即返回；失败抛 SmsError。返回 SendResult（cooldown + 实际渠道）。"""
+    primary = _primary()
+    cooldown = primary.send_code(phone)
+    return SendResult(cooldown_sec=cooldown, provider=_NAME[primary], fallback=False)
+
+
+def verify_code(phone: str, code: str) -> bool:
+    """校验：委托当前主 provider（try-both 在 Task 5 加）。"""
+    return _primary().verify_code(phone, code)
+```
+
+- [ ] **Step 4: 迁移现有 dispatch 测试断言**
+
+在 `tests/test_sms_dispatch.py` 中，把 3 处 `assert sms.send_code("13800138000") == 60`（line 20/22/24）改为：
+
+```python
+    assert sms.send_code("13800138000").cooldown_sec == 60
+```
+
+（line 45-53 的 `test_unknown_provider_falls_back_to_jiguang` 用 `sms.send_code(...)` 不比较返回值，无需改；`test_verify_code_routes_by_provider` 不涉及返回值，无需改。）
+
+- [ ] **Step 5: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_sms_fallback.py tests/test_sms_dispatch.py -q`
+Expected: PASS（新用例 + 原 dispatch 3 用例全绿）
+
+- [ ] **Step 6: 提交**
+
+```bash
+git add app/integrations/sms/__init__.py tests/test_sms_dispatch.py tests/test_sms_fallback.py
+git commit -m "feat(sms): send_code 返回 SendResult(主路径),迁移 dispatch 断言"
+```
+
+---
+
+## Task 4: 分派层 `send_code` fallback 分支（仅 503 转备）
+
+**Files:**
+- Modify: `app/integrations/sms/__init__.py`（`send_code`）
+- Test: `tests/test_sms_fallback.py`
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_sms_fallback.py` 追加：
+
+```python
+def test_send_code_503_falls_back_to_configured_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        lambda phone: (calls.append("chuanglan"), 60)[1])
+
+    r = sms.send_code(PHONE)
+    assert r.provider == "chuanglan"
+    assert r.fallback is True
+    assert r.cooldown_sec == 60
+    assert calls == ["chuanglan"]
+
+
+@pytest.mark.parametrize("status", [429, 400])
+def test_send_code_non_503_does_not_fall_back(monkeypatch, status):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("x", status_code=status)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        lambda phone: (calls.append("chuanglan"), 60)[1])
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert ei.value.status_code == status
+    assert calls == []                      # 备未被调用
+
+
+def test_send_code_503_without_backup_reraises(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")   # 无备
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert ei.value.status_code == 503
+
+
+def test_send_code_backup_also_fails_raises_backup_error(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        _raiser(SmsError("创蓝也不可用", status_code=503)))
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert "创蓝" in str(ei.value)
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_sms_fallback.py -k "fall_back or 503 or non_503 or backup" -q`
+Expected: FAIL — 主抛 503 时当前 `send_code` 未捕获，直接透出（`test_...falls_back...` 期望走到创蓝但没走）
+
+- [ ] **Step 3: 实现 fallback 分支**
+
+把 `app/integrations/sms/__init__.py` 的 `send_code` 替换为：
+
+```python
+def send_code(phone: str) -> SendResult:
+    """发码：主成功即返回；仅主「供应商不可用(503)」且配置了备时转备补发。
+
+    429（本地冷却/超频）、400（手机号无效）不转——不绕过防刷、不为无效号白烧。
+    备也失败则抛备的 SmsError。返回 SendResult（cooldown + 实际渠道 + 是否 fallback）。
+    """
+    primary = _primary()
+    fb = _fallback()
+    try:
+        cooldown = primary.send_code(phone)
+        return SendResult(cooldown_sec=cooldown, provider=_NAME[primary], fallback=False)
+    except SmsError as e:
+        if fb is not None and e.status_code == 503:
+            logger.warning("[SMS] primary=%s 不可用(%s)，fallback→%s",
+                           _NAME[primary], e, _NAME[fb])
+            cooldown = fb.send_code(phone)   # 备的冷却/错误码原样透出
+            return SendResult(cooldown_sec=cooldown, provider=_NAME[fb], fallback=True)
+        raise
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_sms_fallback.py -q`
+Expected: PASS（含 parametrize 的 429/400 两条）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add app/integrations/sms/__init__.py tests/test_sms_fallback.py
+git commit -m "feat(sms): send_code 仅主 503 时转创蓝补发(429/400 不转)"
+```
+
+---
+
+## Task 5: 分派层 `verify_code` try-both（只遍历启用的链）
+
+**Files:**
+- Modify: `app/integrations/sms/__init__.py`（`verify_code`）
+- Test: `tests/test_sms_fallback.py`
+
+- [ ] **Step 1: 写失败测试**
+
+在 `tests/test_sms_fallback.py` 追加：
+
+```python
+def test_verify_hits_primary_without_touching_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), True)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is True
+    assert calls == ["jiguang"]             # 主命中即停，不查备
+
+
+def test_verify_falls_through_to_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), False)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is True
+    assert calls == ["jiguang", "chuanglan"]  # 主未命中 → 查备（码在备）
+
+
+def test_verify_all_miss_returns_false(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code", lambda p, c: False)
+    monkeypatch.setattr(chuanglan, "verify_code", lambda p, c: False)
+
+    assert sms.verify_code(PHONE, "123456") is False
+
+
+def test_verify_backup_not_touched_when_fallback_off(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")   # 关闭 fallback
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), False)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is False
+    assert calls == ["jiguang"]             # 关闭时创蓝零参与
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_sms_fallback.py -k verify -q`
+Expected: FAIL — `test_verify_falls_through_to_backup`（当前 verify 只问主，主 False 即返 False，不查备）
+
+- [ ] **Step 3: 实现 try-both**
+
+把 `app/integrations/sms/__init__.py` 的 `verify_code` 替换为：
+
+```python
+def verify_code(phone: str, code: str) -> bool:
+    """校验：try-both，遍历「启用的 fallback 链」（主→备），任一命中即 True。
+
+    码只存在实际发码那家（fallback 前主已 pop 掉自己的码），另一家 rec is None 即 False、
+    不误判、不累加其防爆破计数。关闭 fallback 时链中只有主，备完全不参与。
+    """
+    chain = [_primary()]
+    fb = _fallback()
+    if fb is not None:
+        chain.append(fb)
+    for prov in chain:
+        if prov.verify_code(phone, code):   # Mode B：纯本地内存比对，不联网
+            logger.info("[SMS] verify hit provider=%s", _NAME[prov])
+            return True
+    return False
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_sms_fallback.py tests/test_sms_dispatch.py -q`
+Expected: PASS（全部 fallback + 原 dispatch 用例）
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add app/integrations/sms/__init__.py tests/test_sms_fallback.py
+git commit -m "feat(sms): verify_code try-both 遍历启用的 fallback 链"
+```
+
+---
+
+## Task 6: auth `sms_send` 接入 SendResult + 记 provider/fallback
+
+**Files:**
+- Modify: `app/api/v1/auth.py`（line 210 调用、line 231-243 成功侧 details）
+- Test: `tests/test_auth.py`（加 1 用例）
+
+- [ ] **Step 1: 写失败测试**
+
+先确认现有 `test_auth.py` 里 `sms_send` 的调用/断言与 fixture 写法（找一个已有的 sms_send 成功用例作参照），再在 `tests/test_auth.py` 末尾追加：
+
+```python
+def test_sms_send_records_provider_in_details(client, db_session, monkeypatch):
+    """sms_send 成功后风控流水 details 带 provider/fallback（mock 下 jiguang/False）。"""
+    from app.repositories import risk as risk_repo
+
+    captured = {}
+    orig = risk_repo.record_behavior_event
+
+    def _spy(db, **kw):
+        if kw.get("event_type") == risk_repo.EVENT_SMS_SEND and kw.get("outcome") == "success":
+            captured.update(kw.get("details") or {})
+        return orig(db, **kw)
+
+    monkeypatch.setattr(risk_repo, "record_behavior_event", _spy)
+
+    resp = client.post("/api/v1/auth/sms/send",
+                       json={"phone": "13800009999", "device_id": "dev-provider-test"})
+    assert resp.status_code == 200, resp.text
+    assert captured.get("provider") == "jiguang"
+    assert captured.get("fallback") is False
+```
+
+> 注：`client` / `db_session` fixture 名、请求路径前缀（`/api/v1/auth/...`）、`record_behavior_event` 的 import 路径以 `tests/test_auth.py` 现有用例为准，若不同则对齐后再写。
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `python -m pytest tests/test_auth.py::test_sms_send_records_provider_in_details -q`
+Expected: FAIL — `captured.get("provider")` 为 None（当前 details 只有 `{"mock": ...}`）
+
+- [ ] **Step 3: 实现**
+
+在 `app/api/v1/auth.py` 的 `sms_send` 中：
+
+(a) line 209-210 的
+```python
+    try:
+        cooldown = send_code(req.phone)
+```
+改为
+```python
+    try:
+        send_result = send_code(req.phone)
+        cooldown = send_result.cooldown_sec
+```
+
+(b) line 231-243 成功侧 `record_behavior_event` 的 `details` 由
+```python
+        details={"mock": settings.SMS_MOCK},
+```
+改为
+```python
+        details={"mock": settings.SMS_MOCK,
+                 "provider": send_result.provider,
+                 "fallback": send_result.fallback},
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `python -m pytest tests/test_auth.py::test_sms_send_records_provider_in_details -q`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add app/api/v1/auth.py tests/test_auth.py
+git commit -m "feat(sms): sms_send 记 provider/fallback 到风控流水 details"
+```
+
+---
+
+## Task 7: `.env.example` + 全量回归 + ruff
+
+**Files:**
+- Modify: `.env.example`
+
+- [ ] **Step 1: 加配置注释**
+
+在 `.env.example` 的 `SMS_PROVIDER` 附近加：
+
+```bash
+# 极光→创蓝 容灾：主 provider(SMS_PROVIDER)供应商不可用(503)时转投的备。
+# 空=关闭 fallback(默认)；生产设 chuanglan 开启极光主、创蓝备；置空即秒回退。
+SMS_FALLBACK_PROVIDER=
+```
+
+- [ ] **Step 2: 全量 SMS 相关回归**
+
+Run: `python -m pytest tests/test_sms_fallback.py tests/test_sms_dispatch.py tests/test_sms_chuanglan.py tests/test_sms_aliyun.py tests/test_auth.py -q`
+Expected: PASS（新增全绿；原 SMS/auth 用例不回归）
+
+- [ ] **Step 3: lint**
+
+Run: `python -m ruff check app/integrations/sms/ app/core/config.py app/api/v1/auth.py tests/test_sms_fallback.py tests/test_sms_base.py`
+Expected: `All checks passed!`
+
+- [ ] **Step 4: 基线对比（确认无新增失败）**
+
+Run: `python -m pytest -q`
+Expected: 失败数 ≤ 基线（已知先前债 ~8 个）；本改动不新增失败。若有新增，定位并修。
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add .env.example
+git commit -m "docs(sms): .env.example 加 SMS_FALLBACK_PROVIDER 说明"
+```
+
+---
+
+## Self-Review
+
+**Spec 覆盖核对**（对 `2026-07-30-sms-jiguang-chuanglan-fallback-design.md`）：
+
+- §2-A 仅 503 转 / 429·400 不转 → Task 4（`test_send_code_503_falls_back` + `test_send_code_non_503_does_not_fall_back`）✅
+- §2-B try-both 只遍历启用链 → Task 5（4 个 verify 用例，含关闭时备零参与）✅
+- §2-C details 记 provider/fallback + 日志 → Task 6（details）+ Task 4/5（logger.warning/info）✅
+- §2-D 配置开关默认空、生产设开 → Task 2 + Task 7 ✅
+- §4.1 主/备解析（未知/同主→None）→ Task 3（`_primary`/`_fallback`）✅
+- §4.2 SendResult + 只认 503 + 备失败抛备错 → Task 1 + Task 4（`test_send_code_backup_also_fails`）✅
+- §4.4 码只在一家/防爆破不被绕过 → provider 内部行为（不改），try-both 逻辑 Task 5 覆盖；防爆破由 provider 各自既有测试保证 ✅
+- §4.5 auth 一处改、verify 调用点不变 → Task 6 ✅
+- §5 配置项 → Task 2 ✅
+- §6 mock 永不 fallback → mock 下主 send_code 成功不抛异常，天然不进 fallback 分支（现有 mock 测试在 Task 7 回归保证）✅
+- §10 send_code int→SendResult 仅 auth 一处 + dispatch 测试 → Task 3/6 ✅
+
+**占位扫描**：无 TBD/TODO；每个 code step 均含完整代码与命令。Task 6 Step 1 标注了「以现有 fixture 为准」——这是**对齐现有约定**的指示，非占位（fixture 名在同文件已存在，执行时照抄）。
+
+**类型/命名一致性**：`SendResult(cooldown_sec, provider, fallback)` 三字段在 Task 1/3/4/6 全程一致；`_primary()`/`_fallback()`/`_ALL`/`_NAME` 命名跨 Task 3/4/5 一致；`SMS_FALLBACK_PROVIDER` 跨 Task 2/3/4/5/7 一致。
+
+---
+
+## 执行顺序说明
+
+Task 1→2→3 有依赖（后者用前者的类型/配置/分派骨架），必须顺序执行。Task 4/5 都改 `send_code`/`verify_code`，也应在 Task 3 之后顺序做。Task 6 依赖 Task 3 的 `SendResult` 返回。Task 7 收尾。全程一条功能分支 `feat/sms-jiguang-chuanglan-fallback`，每 Task 一提交。

docs/superpowers/specs/2026-07-30-sms-jiguang-chuanglan-fallback-design.md

@@ -0,0 +1,208 @@
+# 短信验证码 极光→创蓝 容灾 fallback — 设计方案
+
+- 日期：2026-07-30
+- 状态：待评审
+- 范围：短信发送验证码服务，默认极光；**极光供应商不可用时自动转创蓝补发**；发/校验全程后台可区分本次走的是极光还是创蓝。
+- 关联：`docs/superpowers/specs/2026-07-26-chuanglan-sms-verify-design.md`（创蓝 provider 接入）
+
+## 1. 背景与目标
+
+短信验证码已是**可切换 provider** 架构（`app/integrations/sms/`：`__init__` 分派 + `jiguang` / `aliyun` / `chuanglan` + `base`）。极光（默认）与创蓝都是 **Mode B（自管码）**：本服务 `secrets` 生成码 → 存**各自模块的进程内存** → provider 只负责发送；`verify_code` 是**纯本地内存比对**，不联网。
+
+现状是"单选一个 provider"：`send_code` / `verify_code` 都只走 `SMS_PROVIDER` 指定的那一家。极光一旦供应商侧故障（欠费 / 网络 / 服务异常），`/sms/send` 直接 503，用户收不到码、登录中断。
+
+**目标**：把极光设为主、创蓝设为备，在**极光供应商不可用**时自动转创蓝补发，最大化验证码送达率；同时让后台能查到每次发码/校验实际走的是哪家、是否发生了 fallback。
+
+**非目标**：不改极光/创蓝各自的 Mode B 机器（发码/存码/校验/防刷），不引入 DB/Redis 共享存码，不做多 worker 共享（沿用现有内存态技术债）。
+
+## 2. 关键决策（已与需求方确认）
+
+| # | 决策 | 结论 |
+|---|---|---|
+| A | **fallback 触发范围** | **仅主 provider 返回"供应商不可用"（`SmsError.status_code == 503`：网络错误 / 余额不足 / 服务故障）时转备。** 本地单号 60s 冷却与极光侧超频（均 429）、手机号无效（400）**不转**——不绕过防刷、不为无效号白烧一条。 |
+| B | **校验路由** | **try-both，但只遍历"当前启用的 fallback 链"。** verify 依次问链上 provider（主→备），任一命中即通过；关闭 fallback 时链中只有主，创蓝完全不参与。 |
+| C | **后台可见性** | **风控行为流水 + 结构化日志。** 复用 `risk_repo.record_behavior_event(EVENT_SMS_SEND)` 的 `details` 加 `provider` / `fallback` 字段；dispatch 层对 fallback 触发、校验命中打点。不新增表、不改客户端响应。 |
+| D | **默认开关** | **配置开关 opt-in，生产设开。** 新增 `SMS_FALLBACK_PROVIDER`，默认空=关（保持现状零风险）；生产 `.env` 设 `chuanglan` 开启。出问题置空即秒回退、不改代码。 |
+
+补充约束：**fallback 仅在 Mode B ↔ Mode B（jiguang / chuanglan）之间有意义**。aliyun 是 Mode A（阿里云托管码、校验走接口），不参与 fallback 链。
+
+## 3. 现状回顾（设计依赖的事实）
+
+- **分派层** [`app/integrations/sms/__init__.py`](../../../app/integrations/sms/__init__.py)：`_provider()` 每次按 `settings.SMS_PROVIDER` 选一个模块（默认 jiguang），`send_code` / `verify_code` 委托给它。
+- **极光 / 创蓝均 Mode B**，各持一份**独立**的 `_codes` / `_last_sent` / `_lock`（两个互不相通的进程内存 dict）。`verify_code` 纯本地比对（`_codes.get(phone)` + `secrets.compare_digest`），**不发网络**；provider 的网络调用只在**发送**一步。
+- **`SmsError(message, status_code)`**（[`base.py`](../../../app/integrations/sms/base.py)）：`429`=过频/超频（客户端稍后再来）、`503`=供应商不可用、`400`=手机号无效。这是 A 决策用来判定"是否 fallback"的唯一信号。
+- **极光/创蓝 `send_code` 失败时"保留冷却、清掉未发出的码"**：`_last_sent[phone]=now` 保留、`_codes.pop(phone)`。此行为不改（见 §4.3 冷却交互）。
+- **auth 层落点**：`sms_send` 发码成功 [auth.py:231](../../../app/api/v1/auth.py) 记 `EVENT_SMS_SEND` `outcome="success" details={"mock":...}`；失败 [auth.py:212](../../../app/api/v1/auth.py) 记 `outcome="failed" reason=...`。`send_code` 是内部函数，全仓仅此一处调用。`verify_code` 在 `sms_login` 与换绑两处调用，签名 `(phone, code)->bool`。
+
+## 4. 设计
+
+### 4.1 fallback 链模型（分派层，`__init__.py`）
+
+引入"主 + 备"两级链，全部在分派层组装，**provider 文件一行不动**：
+
+```python
+_ALL = {"jiguang": jiguang, "aliyun": aliyun, "chuanglan": chuanglan}
+_NAME = {jiguang: "jiguang", aliyun: "aliyun", chuanglan: "chuanglan"}
+
+def _primary():
+    # 未知/缺省回退 jiguang（保持 test_unknown_provider_falls_back_to_jiguang 语义）
+    return _ALL.get(settings.SMS_PROVIDER, jiguang)
+
+def _fallback():
+    """启用的备 provider；未配置/同主/未知名 → None（= 不启用 fallback）。"""
+    name = (settings.SMS_FALLBACK_PROVIDER or "").strip()
+    if not name or name == settings.SMS_PROVIDER:
+        return None
+    return _ALL.get(name)  # 未知名 → None
+```
+
+### 4.2 发码 `send_code(phone) -> SendResult`（透出实际渠道）
+
+现返回 `int`（cooldown），改为返回携带渠道信息的小结果对象（放 `base.py`，provider 无关）：
+
+```python
+@dataclass(frozen=True)
+class SendResult:
+    cooldown_sec: int   # 距下次可发秒数（= 实际发码 provider 的 SMS_SEND_INTERVAL_SEC）
+    provider: str       # 实际发码的 provider 名："jiguang" / "chuanglan"
+    fallback: bool      # 是否走了备（True=主不可用转备）
+```
+
+分派逻辑：
+
+```python
+def send_code(phone) -> SendResult:
+    primary, fb = _primary(), _fallback()
+    try:
+        cooldown = primary.send_code(phone)
+        return SendResult(cooldown, _NAME[primary], fallback=False)
+    except SmsError as e:
+        # A 决策：仅"供应商不可用(503)"且配置了备时转备；429/400 原样抛
+        if fb is not None and e.status_code == 503:
+            logger.warning("[SMS] primary=%s 不可用(%s)，fallback→%s",
+                           _NAME[primary], e, _NAME[fb])
+            cooldown = fb.send_code(phone)   # 备也可能抛 SmsError（其冷却/错误码）→ 原样透出
+            return SendResult(cooldown, _NAME[fb], fallback=True)
+        raise
+```
+
+要点：
+- **只认 `status_code == 503`**：极光的 429（本地冷却 / 超频 50009）、400（手机号无效）直接 raise，不 fallback。
+- **备自带完整 Mode B 机器**：创蓝重新走自己的单号冷却 / 存码 / 错误码映射。备也失败则抛备的 `SmsError`，`auth` 按其 `status_code` 返回。
+- **主/备文件不动**：`jiguang.send_code` / `chuanglan.send_code` 仍返回 `int`；由分派层包装成 `SendResult`。
+
+### 4.3 冷却交互（已知行为，保守安全，不改 provider）
+
+极光 `send_code` 失败（503）时**保留自己的冷却**（现有行为，防"欠费/签名失效时前端狂打极光"）。在 fallback 下的连锁：
+
+1. 首发：极光 503（记极光冷却）→ 转创蓝发成功（记创蓝冷却）。用户收到创蓝的码。
+2. 60s 内重发：主=极光**冷却中 → 429**，`429 ≠ 503` → **不 fallback** → 用户收到 429"过于频繁"。**这是期望行为**：挡住 60s 内重复补发烧钱；且此时创蓝也在自己冷却期。
+3. 极光**持续故障**：每过 60s，极光冷却过期→再试极光（仍 503）→转创蓝补发。稳态下每 60s 白试一次极光、成功走创蓝，功能可用。
+
+结论：现有"失败保留冷却"语义在 fallback 下自洽且保守。**不改极光/创蓝任何冷却逻辑。**
+
+### 4.4 校验 `verify_code(phone, code) -> bool`（try-both over 启用链）
+
+```python
+def verify_code(phone, code) -> bool:
+    chain = [_primary()]
+    fb = _fallback()
+    if fb is not None:
+        chain.append(fb)
+    for prov in chain:
+        if prov.verify_code(phone, code):   # Mode B：纯本地内存比对，不联网、不抛异常
+            logger.info("[SMS] verify hit provider=%s", _NAME[prov])
+            return True
+    return False
+```
+
+要点：
+- **关闭 fallback 时 `chain=[极光]`**，与现状字节级等价，**创蓝完全不参与**。
+- **码只存在实际发码那家**（fallback 发码前极光已 `_codes.pop` 清掉自己的码）→ 同一手机号同一时刻只有一家有码；另一家 `rec is None` 立即 False，不误判、不累加其 `attempts`。**防爆破仍在真实那家正常生效，不被绕过**。
+- **命中即知渠道**（日志打点），天然满足 C 的校验侧可见性，无需显式记忆 phone→provider。
+- 绝大多数登录：极光发成功→码在极光→第一家命中，创蓝那次内存查根本不发生。
+- Mode B 的 `verify_code` 不抛 `SmsError`；若链中混入 Mode A（aliyun，本设计约束外）其 verify 降级抛 503，应原样透出中断——但按 §2 约束 aliyun 不入链，此路不触发。
+
+### 4.5 auth.py 改动（最小）
+
+- **`sms_send`**（唯一改动点）：
+  ```python
+  result = send_code(req.phone)              # 现 SendResult
+  cooldown = result.cooldown_sec
+  # 成功侧 details 加 provider/fallback（C）
+  details={"mock": settings.SMS_MOCK, "provider": result.provider, "fallback": result.fallback}
+  ```
+  `SmsSendResponse(cooldown_sec=cooldown, ...)` 不变——**不把 provider 返回客户端**（C 选风控流水，非接口返回）。
+- **`verify_code` 两处调用点不变**：签名 `(phone,code)->bool` 未改；校验命中渠道由分派层 §4.4 日志打点，auth 无需感知。
+- **失败可观测**：极光 503 且创蓝也失败时，抛创蓝的 `SmsError` → 现有失败侧 `record_behavior_event(reason=str(e))` 记录；fallback 触发/失败另有 §4.2 分派层 `logger.warning`。
+
+### 4.6 可观测性汇总（满足 C）
+
+| 事件 | 落点 | 内容 |
+|---|---|---|
+| 发码成功 | `EVENT_SMS_SEND` `details` | `provider`（jiguang/chuanglan）、`fallback`（bool） |
+| fallback 触发 | 分派层 `logger.warning` | 主不可用原因 + 转向的备 |
+| 校验命中 | 分派层 `logger.info` | 命中的 `provider` |
+| 发码彻底失败 | `EVENT_SMS_SEND` `outcome=failed` | `reason`（末端 SmsError 文案） |
+
+## 5. 配置项（`app/core/config.py` 新增）
+
+```python
+# 主 provider 供应商不可用(503)时转投的备 provider；空=关闭 fallback(默认，保持现状)。
+# 仅 Mode B(jiguang/chuanglan)互为主备有意义；生产建议设 "chuanglan"。
+SMS_FALLBACK_PROVIDER: Literal["jiguang", "chuanglan", ""] = ""  # Mode B only(aliyun 不入链)
+```
+
+- 复用现有 `SMS_PROVIDER`（主，默认 jiguang）与两家的凭证/旋钮（`CHUANGLAN_SMS_*`、`SMS_CODE_LENGTH` 等），**无其它新增**。
+- `.env.example` 加 `SMS_FALLBACK_PROVIDER=`（注释：生产设 `chuanglan` 开启极光→创蓝容灾）。
+- 生产前置：创蓝凭证齐全（`chuanglan_sms_configured`）+ 创蓝控制台 IP 白名单，否则 fallback 到创蓝仍 503。
+
+## 6. 错误处理与边界
+
+- **主 503 + 未配备**：`_fallback()` 返 None → 原样抛主的 503（= 现状）。
+- **主 503 + 备也失败**：抛备的 `SmsError`（按备错误码 429/400/503），auth 照常返回。
+- **主 429/400**：不 fallback，原样抛。
+- **mock 模式（`SMS_MOCK=true`）**：主 `send_code` mock 分支直接成功、不抛异常 → **永不触发 fallback**；`verify` mock 分支放行任意 N 位 → try-both 第一家（主）即 True。**现有 mock 测试全绿不变。**
+- **未知 `SMS_FALLBACK_PROVIDER` 名**：`_ALL.get` 返 None → 视为未配置、不 fallback（不崩）。
+- **`SMS_FALLBACK_PROVIDER == SMS_PROVIDER`**：`_fallback()` 返 None（不自我 fallback）。
+
+## 7. 测试计划
+
+新增 `tests/test_sms_fallback.py`（monkeypatch 各 provider 的 `send_code`/`verify_code`，不真发）：
+
+1. 主发成功 → `SendResult(provider=主, fallback=False)`，不调备。
+2. 主 503 + 配备 → 转备，备成功 → `SendResult(provider=备, fallback=True)`。
+3. 主 429 / 400 → 不 fallback，原样抛（断言备未被调用）。
+4. 主 503 + 未配备 → 抛 503。
+5. 主 503 + 备也 503 → 抛备的 SmsError。
+6. verify：码在主 → 命中主；码在备 → 命中备；两家都无 → False。
+7. **关闭 fallback**（`SMS_FALLBACK_PROVIDER=""`）→ verify 只问主，**断言备 `verify_code` 未被调用**；send 主失败原样抛、不转备。
+8. 防爆破：码在主，连续错到 `SMS_MAX_VERIFY_ATTEMPTS` 作废；全程备 `rec is None` 不受影响。
+
+auth 层扩 `tests/test_auth.py`：`sms_send` 成功后 `EVENT_SMS_SEND` 的 `details` 含 `provider`/`fallback`（mock 下 provider=jiguang、fallback=False）。
+
+回归：`test_sms_dispatch.py`、现有 `test_auth.py` 中依赖 `send_code` 返回 `int` 的断言改为读 `SendResult.cooldown_sec`。
+
+## 8. YAGNI（明确不做）
+
+- ❌ 不改极光/创蓝的 Mode B 机器（发/存/校验/冷却/防爆破）。
+- ❌ 不做 DB/Redis 共享存码或 phone→provider 记忆表（多 worker 债维持现状、两家一起将来迁）。
+- ❌ 不做 aliyun（Mode A）参与 fallback。
+- ❌ 不做多级（>2）fallback 链、不做主备自动健康探测/熔断。
+- ❌ 不把 provider 渠道透传给客户端（仅后台可见）。
+- ❌ 不做发送量/成功率统计报表（C 只取"风控流水+日志"，报表另议）。
+
+## 9. 验收标准
+
+- `SMS_FALLBACK_PROVIDER=chuanglan` 且创蓝配置齐全时：极光注入 503（欠费/网络）→ `/sms/send` 实际由创蓝下发、真机可收码并登录；`EVENT_SMS_SEND.details.provider=chuanglan`、`fallback=true`。
+- 极光正常时：走极光，`provider=jiguang`、`fallback=false`；创蓝零调用。
+- `SMS_FALLBACK_PROVIDER=""`（默认）：行为与当前完全一致，创蓝不参与发/校验。
+- 极光 429/400：不转创蓝，原样返回。
+- `SMS_MOCK=true`：任意 N 位通过、不真发、不触发 fallback。
+- `ruff check .` 通过；新增/现有 `pytest` 全绿。
+
+## 10. 影响面 / 回滚
+
+- **改动文件**：`sms/__init__.py`（分派层，主要）、`sms/base.py`（加 `SendResult`）、`core/config.py`（加开关）、`api/v1/auth.py`（`sms_send` 一处记 details）、`.env.example`、测试。**`jiguang.py` / `chuanglan.py` / `aliyun.py` 不动。**
+- **破坏性接口变更**：`send_code` 返回 `int → SendResult`，仅 `auth.py:210` 一处调用 + 相关测试需同步。
+- **回滚**：置 `SMS_FALLBACK_PROVIDER=""` 即运行时秒回退到"纯极光"，无需改代码/迁移。

tests/test_ad_reward_coin_floor.py

@@ -0,0 +1,59 @@
+"""发奖公式下限:有真实正 eCPM 时最低发 1 金币,eCPM 缺失/为 0 仍发 0。
+
+针对「eCPM 过低时公式四舍五入成 0 金币 → 被记 too_short 不发」的问题:只要广告有真实
+正 eCPM,单份金币至少 1(不再 0);但 eCPM 缺失/为 0/非法(没有真实广告价值)仍发 0,
+不凭空铸币、不破坏 ecpm_missing 语义。公式是发奖与后台审计对账的唯一口径,改这一处两边同源。
+"""
+from __future__ import annotations
+
+from app.core.rewards import calculate_ad_reward_coin
+
+
+def test_low_positive_ecpm_floors_to_one_coin() -> None:
+    """真实但极低的 eCPM 原本四舍五入成 0 金币,现在兜底为 1(线上 record 4667 的 43 分场景)。"""
+    # 43 分 = ¥0.43 CPM,因子1=0.1,重度用户 LT 第 69 条=1.0 → 0.43/1000×0.1×1.0×10000=0.43 → 旧口径 round=0
+    assert calculate_ad_reward_coin("43", 69) == 1
+    # 更低的 5 分同理:算出来 <0.5,旧口径也是 0
+    assert calculate_ad_reward_coin("5", 11) == 1
+
+
+def test_zero_or_missing_ecpm_stays_zero() -> None:
+    """eCPM 缺失 / 为 0 / 非法(没有真实广告价值)不兜底,仍发 0。"""
+    assert calculate_ad_reward_coin("0", 1) == 0
+    assert calculate_ad_reward_coin(None, 1) == 0
+    assert calculate_ad_reward_coin("", 1) == 0
+    assert calculate_ad_reward_coin("abc", 1) == 0
+
+
+def test_normal_ecpm_value_unchanged() -> None:
+    """正常量级 eCPM 本就 ≥1,下限不改变其取值(回归保护)。"""
+    # 200 分 = ¥2 CPM,因子1=0.1,LT 第 1 条=2.0 → 2/1000×0.1×2.0×10000 = 4
+    assert calculate_ad_reward_coin("200", 1) == 4
+
+
+def test_feed_reward_low_ecpm_grants_one_coin_instead_of_too_short() -> None:
+    """端到端:低 eCPM 的信息流广告看满一份(≥10s),原本单份金币算成 0 被记 too_short,
+    现在兜底发 1 金币、状态 granted(线上 record 4667 那类「看了却零发」的修复)。"""
+    from app.db.session import SessionLocal
+    from app.models.user import User
+    from app.repositories.ad_feed_reward import grant_feed_reward
+
+    db = SessionLocal()
+    try:
+        user = User(phone="19900000043", username="feedfloor43", register_channel="sms")
+        db.add(user)
+        db.commit()
+        db.refresh(user)
+
+        rec = grant_feed_reward(
+            db, user.id,
+            client_event_id="feed-floor-low-ecpm-0001",
+            ecpm="2",                 # ¥0.02 CPM,单份收益四舍五入 <0.5 → 旧口径算成 0
+            duration_seconds=15,      # ≥10s → 看满一份,不走「时长不足」的 else 分支
+            ad_type="draw",
+            feed_scene="comparison",
+        )
+        assert rec.status == "granted"
+        assert rec.coin == 1
+    finally:
+        db.close()

tests/test_admin_read.py

@@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
     assert invite.json()["cash_balance_cents"] == 456
 
 
+def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
+    from app.models.ad_ecpm import AdEcpmRecord
+    from app.models.ad_feed_reward import AdFeedRewardRecord
+
+    uid = _seed_user_with_data("13800000024")
+    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
+    db = SessionLocal()
+    try:
+        db.add_all(
+            [
+                AdFeedRewardRecord(
+                    client_event_id="reward-stats-granted-high",
+                    ad_session_id="reward-stats-granted-high",
+                    user_id=uid,
+                    reward_date="2038-01-15",
+                    duration_seconds=10,
+                    unit_count=1,
+                    ecpm_raw="9000",
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    coin=9,
+                    status="granted",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-low",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="1000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="feed",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-impression-mid",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="3000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="comparison",
+                    ad_session_id="reward-stats-other-scene",
+                    app_env="prod",
+                    our_code_id="104098712",
+                    ecpm_raw="7000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-test-env",
+                    app_env="test",
+                    our_code_id="104127529",
+                    ecpm_raw="8000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+                AdEcpmRecord(
+                    user_id=uid,
+                    ad_type="draw",
+                    feed_scene="coupon",
+                    ad_session_id="reward-stats-non-business",
+                    app_env="prod",
+                    our_code_id="demo-slot",
+                    ecpm_raw="9000",
+                    report_date="2038-01-15",
+                    created_at=created_at,
+                ),
+            ]
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        f"/admin/api/users/{uid}/reward-stats",
+        params={
+            "date_from": "2038-01-15T00:00:00Z",
+            "date_to": "2038-01-15T23:59:59Z",
+            "app_env": "prod",
+            "revenue_scope": "business",
+            "feed_scene": "coupon",
+        },
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    data = response.json()
+    assert data["feed_count"] == 1
+    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
+    assert data["feed_avg_ecpm"] == 2000.0
+
+
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
     """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
     naive = datetime(2038, 1, 1, 8, 0)

tests/test_auth.py

@@ -68,6 +68,29 @@ def test_sms_login_and_me_flow(client) -> None:
     assert r.json()["ok"] is True
 
 
+def test_sms_send_records_provider_and_fallback_in_details(client, monkeypatch) -> None:
+    """批次3(Task 6):sms_send 成功后风控流水 details 带 provider/fallback。
+    mock 默认走极光主(SMS_FALLBACK_PROVIDER 空)→ provider=jiguang、fallback=False。
+    (回归:修复前 send_code 返回 SendResult 被塞进 cooldown_sec:int → 500,此处 200 断言即挡住。)"""
+    from app.repositories import risk as risk_repo
+
+    captured: dict = {}
+    orig = risk_repo.record_behavior_event
+
+    def _spy(db, **kw):
+        if kw.get("event_type") == risk_repo.EVENT_SMS_SEND and kw.get("outcome") == "success":
+            captured.update(kw.get("details") or {})
+        return orig(db, **kw)
+
+    monkeypatch.setattr(risk_repo, "record_behavior_event", _spy)
+
+    r = client.post("/api/v1/auth/sms/send",
+                    json={"phone": "13800009999", "device_id": "dev-provider-test"})
+    assert r.status_code == 200, r.text
+    assert captured.get("provider") == "jiguang"
+    assert captured.get("fallback") is False
+
+
 def test_sms_send_too_frequent(client) -> None:
     phone = "13900139000"
     assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200

tests/test_compare_harvest.py

@@ -102,6 +102,7 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert rec.is_source_best is False
         assert rec.store_name == "测试店"
         assert rec.information == "美团更便宜"
+        assert rec.fail_reason is None               # 成功记录不派生失败原因
         assert rec.items == [{"name": "肥牛饭", "qty": 1}]
         assert rec.trace_url.endswith("/done/")
         # 再来一次(重试 done)→ 已 success,newly_success=False(发奖不重复触发)
@@ -110,6 +111,104 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
         assert newly2 is False
 
 
+def test_harvest_done_failed_derives_fail_reason(client) -> None:
+    """failed 记录:记录级 information 笼统,但 fail_reason 从 platform_results 救出具体原因
+    (id 3030 型:美团系统失败 + 京东 items_not_found → 展示京东那条)。"""
+    tid = _tid()
+    done_failed = {
+        "comparison_results": [
+            {"platform_id": "taobao_flash", "platform_name": "淘宝闪购",
+             "package": "com.taobao.taobao", "price": 23.04, "is_source": True, "rank": 1,
+             "items": [{"name": "肥牛饭", "qty": 1}]},
+        ],
+        "platform_results": {
+            "taobao_flash": {"is_source": True, "status": "source", "price": 23.04},
+            "meituan_waimai": {"is_source": False, "status": "failed",
+                               "reason": "搜索店铺失败, 无法跳转到搜索页"},
+            "jd_waimai_standalone": {"is_source": False, "status": "items_not_found",
+                                     "reason": "京东外卖此店内未找到这些菜品"},
+        },
+        "information": "比价过程出错，请稍后重试",
+    }
+    with SessionLocal() as db:
+        crud.harvest_running(db, trace_id=tid, user_id=None)
+        rec, newly = crud.harvest_done(db, trace_id=tid, user_id=None,
+                                       done_params=done_failed)
+        assert newly is False                        # 没落成 success
+        assert rec.status == "failed"
+        assert rec.fail_reason == "京东外卖此店内未找到这些菜品"
+        assert rec.information == "比价过程出错，请稍后重试"  # 原文案仍留存
+
+
+def _done_params_platforms_no_isbest() -> dict:
+    """id 3304 型:done 帧 platforms 全平台「相似替换/仅供参考」(has_dish_diff),pricebot
+    一个 is_best 都没标,但有有价目标(美团 57.8 < 源淘宝闪购 60.8)。"""
+    return {
+        "platforms": [
+            {"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
+             "package": "me.ele", "price": 60.8, "is_best": False, "has_dish_diff": False,
+             "store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
+            {"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
+             "package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
+             "has_dish_diff": True},
+            {"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
+             "package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
+        ],
+        "information": "美团更便宜（含相似商品替换）",
+    }
+
+
+def test_harvest_done_platforms_no_isbest_falls_back_to_cheapest_target(client) -> None:
+    """回归(id 3304):platforms 全无 is_best(全平台相似替换)但有有价目标 → best 应兜底取
+    有价目标里最低那家,不能让 best_*/saved 整条落 NULL(否则首页价 0.00 / 记录页无最低红框 /
+    省额丢失)。"""
+    tid = _tid()
+    with SessionLocal() as db:
+        crud.harvest_running(db, trace_id=tid, user_id=None)
+        rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
+                                   done_params=_done_params_platforms_no_isbest())
+        assert rec.status == "success"
+        assert rec.best_platform_id == "meituan_waimai"   # 有价目标里最低
+        assert rec.best_price_cents == 5780               # 57.8 元
+        assert rec.source_price_cents == 6080             # 源淘宝闪购 60.8
+        assert rec.saved_amount_cents == 300              # 60.8 - 57.8
+        assert rec.is_source_best is False                # 兜底选的是目标,非源
+
+
+def _done_params_no_isbest_source_cheapest() -> dict:
+    """无 is_best 且源本身最便宜:源淘宝闪购 50.0 < 全部 dish-diff 目标(美团 57.8 / 京东 74.9)。
+    此时不能强选更贵的目标当 best(否则 saved 变负、污染「累计发现可省」),应回落源。"""
+    return {
+        "platforms": [
+            {"role": "source", "platform_id": "eleme", "platform_name": "淘宝闪购",
+             "package": "me.ele", "price": 50.0, "is_best": False, "has_dish_diff": False,
+             "store_name": "窑鸡王", "items": [{"name": "招牌窑鸡 整只-香辣", "qty": 1}]},
+            {"role": "target", "platform_id": "meituan_waimai", "platform_name": "美团外卖",
+             "package": "com.sankuai.meituan.takeoutnew", "price": 57.8, "is_best": False,
+             "has_dish_diff": True},
+            {"role": "target", "platform_id": "jd_waimai_standalone", "platform_name": "京东外卖",
+             "package": "com.jd.waimai", "price": 74.9, "is_best": False, "has_dish_diff": True},
+        ],
+        "information": "源平台已是最低（其余为相似替换）",
+    }
+
+
+def test_harvest_done_no_isbest_source_cheapest_falls_back_to_source(client) -> None:
+    """回归:无 is_best 且源最便宜 → best 回落源(与 _derive「全目标缺菜回落源、不虚报省」同语义),
+    saved=0、is_source_best=True,绝不因强选更贵目标而让 saved 变负、倒扣「累计发现可省」。"""
+    tid = _tid()
+    with SessionLocal() as db:
+        crud.harvest_running(db, trace_id=tid, user_id=None)
+        rec, _ = crud.harvest_done(db, trace_id=tid, user_id=None,
+                                   done_params=_done_params_no_isbest_source_cheapest())
+        assert rec.status == "success"
+        assert rec.best_platform_id == "eleme"            # 回落到源(源最便宜)
+        assert rec.best_price_cents == 5000               # 源 50.0
+        assert rec.source_price_cents == 5000
+        assert rec.saved_amount_cents == 0                # 没省到,绝不为负
+        assert rec.is_source_best is True                 # 源就是最便宜
+
+
 def test_harvest_abort_cancels_running(client) -> None:
     tid = _tid()
     with SessionLocal() as db:
@@ -233,7 +332,9 @@ def test_trace_finalize_harvests_abort(client) -> None:
     with SessionLocal() as db:                        # 先有 running 行(帧0建的)
         crud.harvest_running(db, trace_id=tid, user_id=None)
     p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"})
-    with p:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/trace/finalize",
                         json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
     assert r.status_code == 200
@@ -241,6 +342,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
         rec = _get(db, tid)
         assert rec is not None and rec.status == "cancelled"
         assert rec.trace_url.endswith("/fin/")
+        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_price_step_binds_user_when_authed(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -72,6 +72,7 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
 
 def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
     missing_id = _record("llm-repair-missing")
+    cancelled_id = _record("llm-repair-cancelled", status="cancelled")
     running_id = _record("llm-repair-running", status="running")
     calls = [
         {
@@ -96,12 +97,15 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
         )
         assert result["repaired"] >= 1
         assert "llm-repair-missing" in seen
+        assert "llm-repair-cancelled" in seen
         assert "llm-repair-running" not in seen
         with SessionLocal() as db:
             assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
+            assert db.get(ComparisonRecord, cancelled_id).llm_cost_yuan is not None
             assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
     finally:
         _delete(missing_id)
+        _delete(cancelled_id)
         _delete(running_id)
 
 

tests/test_fail_reason.py

@@ -0,0 +1,122 @@
+"""失败卡展示原因派生(repositories.comparison._derive_fail_display)单元测试。
+
+用例取自线上真实 failed 记录(platform_results 形态),覆盖:
+  - information 具体 → 直出
+  - information 笼统 + platform_results 有干净业务结局 → 救援出该原因(id 3030/2964 型)
+  - information 笼统 + 仅系统失败(搜索失败等黑话)→ None(端侧品牌兜底,id 3027 型)
+  - store_closed / no_delivery 被 pricebot 漏成 status=failed → 按 reason 关键字补判
+  - 打烊类脏店名 blob → 统一简短模板
+  - platform_results 为空 / 非对象 → None
+"""
+from __future__ import annotations
+
+from app.repositories import comparison as crud
+
+
+def test_specific_information_passthrough() -> None:
+    # information 本身具体(未达起送/找不到菜等)→ 直出,不看 platform_results
+    assert (
+        crud._derive_fail_display("淘宝闪购未达起送门槛，可加菜凑单后下单", {})
+        == "淘宝闪购未达起送门槛，可加菜凑单后下单"
+    )
+    assert crud._derive_fail_display("未识别到商品", {}) == "未识别到商品"
+
+
+def test_generic_info_rescued_from_items_not_found() -> None:
+    # id 3030 型:美团系统失败 + 京东 items_not_found,记录级 information 笼统 → 救出京东那条
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 23.04},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "items_not_found",
+            "reason": "京东外卖此店内未找到这些菜品",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "京东外卖此店内未找到这些菜品"
+    )
+
+
+def test_generic_info_rescued_from_store_not_found() -> None:
+    # id 2964 型:美团系统失败 + 京东 store_not_found → 救出京东相似店铺文案
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 127.98},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "比价过程出错，请稍后重试",
+        },
+        "jd_waimai": {
+            "is_source": False, "status": "store_not_found",
+            "reason": "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺",
+        },
+    }
+    assert (
+        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
+        == "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺"
+    )
+
+
+def test_generic_info_pure_system_failure_returns_none() -> None:
+    # id 3027 型:唯一目标平台是自动化黑话失败 → 不给用户看 → None(端侧品牌兜底)
+    pr = {
+        "eleme": {"is_source": True, "status": "source", "price": 18.83},
+        "meituan_waimai": {
+            "is_source": False, "status": "failed",
+            "reason": "搜索店铺失败, 无法跳转到搜索页",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) is None
+
+
+def test_store_closed_leaked_to_failed_is_rescued_and_cleaned() -> None:
+    # 打烊被漏成 status=failed;reason 常带脏店名 blob → 统一简短模板
+    pr = {
+        "jd_waimai": {"is_source": True, "status": "source", "price": 25},
+        "taobao_flash": {
+            "is_source": False, "status": "failed",
+            "reason": "淘宝闪购「沙胆彪炭炉牛杂煲(...)，蜂鸟准时达，月售300+，起送¥20」本店已休息,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "门店休息中，无法比价"
+
+    pr2 = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 31.83},
+        "meituan": {
+            "is_source": False, "status": "failed",
+            "reason": "美团「奈雪的茶（北京王府井奥莱·香江」门店已打烊,无法比价",
+        },
+    }
+    assert crud._derive_fail_display("比价出错", pr2) == "门店已打烊，无法比价"
+
+
+def test_no_delivery_leaked_to_failed_is_rescued() -> None:
+    # 单点不配送被漏成 status=failed;reason 本身干净 → 直接用
+    reason = "京东外卖该商家所选商品单点不配送，无法进入结算比价"
+    pr = {
+        "taobao_flash": {"is_source": True, "status": "source", "price": 20.1},
+        "jd_waimai_standalone": {
+            "is_source": False, "status": "failed", "reason": reason,
+        },
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == reason
+
+
+def test_empty_or_missing_platform_results_returns_none() -> None:
+    # 「比价出错」+ 空 {} / None / 非对象:引擎早夭,无可展示原因 → None
+    assert crud._derive_fail_display("比价出错", {}) is None
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", None) is None
+    assert crud._derive_fail_display("比价出错", []) is None  # 老 array 形态,防御
+
+
+def test_clean_status_wins_over_priority_order() -> None:
+    # 多个业务结局同现时按 _BIZ_STATUS_PRIORITY 选(below_minimum 优先于 store_not_found)
+    pr = {
+        "src": {"is_source": True, "status": "source", "price": 30},
+        "a": {"is_source": False, "status": "store_not_found", "reason": "未找到店铺A"},
+        "b": {"is_source": False, "status": "below_minimum", "reason": "B未达起送门槛"},
+    }
+    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "B未达起送门槛"

tests/test_guide_video.py

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

tests/test_push_center.py

@@ -7,7 +7,9 @@ mock/真发/变量覆盖/站内联动/设备反查/参数校验。厂商 HTTP
 from __future__ import annotations
 
 import json
+import logging
 
+import httpx
 import pytest
 from fastapi.testclient import TestClient
 
@@ -110,6 +112,99 @@ 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"

tests/test_sms_base.py

@@ -0,0 +1,26 @@
+"""SMS 共享基座类型 SendResult。"""
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from app.integrations.sms.base import SendResult
+
+
+def test_send_result_fields():
+    r = SendResult(cooldown_sec=60, provider="jiguang", fallback=True)
+    assert r.cooldown_sec == 60
+    assert r.provider == "jiguang"
+    assert r.fallback is True
+
+
+def test_send_result_fallback_defaults_false():
+    r = SendResult(cooldown_sec=60, provider="jiguang")
+    assert r.fallback is False
+
+
+def test_send_result_is_frozen():
+    r = SendResult(cooldown_sec=60, provider="jiguang")
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        r.provider = "chuanglan"

tests/test_sms_dispatch.py

@@ -17,11 +17,11 @@ def test_send_code_routes_by_provider_and_switches_per_call(monkeypatch) -> None
     monkeypatch.setattr(chuanglan, "send_code", lambda phone: (calls.append("chuanglan"), 60)[1])
 
     monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
-    assert sms.send_code("13800138000") == 60
+    assert sms.send_code("13800138000").cooldown_sec == 60
     monkeypatch.setattr(settings, "SMS_PROVIDER", "aliyun")
-    assert sms.send_code("13800138000") == 60
+    assert sms.send_code("13800138000").cooldown_sec == 60
     monkeypatch.setattr(settings, "SMS_PROVIDER", "chuanglan")
-    assert sms.send_code("13800138000") == 60
+    assert sms.send_code("13800138000").cooldown_sec == 60
 
     assert calls == ["jiguang", "aliyun", "chuanglan"]  # 每次按当前 provider 路由,运行时可切
 

tests/test_sms_fallback.py

@@ -0,0 +1,142 @@
+"""极光→创蓝 容灾 fallback（分派层 send_code / verify_code）。
+
+monkeypatch 各 provider 的 send_code/verify_code 为标记函数，不发真网络；
+断言：仅主 503 才转备；429/400 不转；verify try-both 只遍历启用的链。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.core.config import settings
+from app.integrations import sms
+from app.integrations.sms import chuanglan, jiguang
+from app.integrations.sms.base import SendResult, SmsError
+
+PHONE = "13800138000"
+
+
+def _raiser(exc: Exception):
+    """返回一个调用即 raise exc 的 send_code 替身（lambda 不能 raise）。"""
+    def _f(phone):
+        raise exc
+    return _f
+
+
+def test_fallback_provider_defaults_empty():
+    assert settings.SMS_FALLBACK_PROVIDER == ""
+
+
+def test_send_code_primary_success_returns_sendresult(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
+    monkeypatch.setattr(jiguang, "send_code", lambda phone: 60)
+
+    r = sms.send_code(PHONE)
+    assert isinstance(r, SendResult)
+    assert r.cooldown_sec == 60
+    assert r.provider == "jiguang"
+    assert r.fallback is False
+
+
+def test_send_code_503_falls_back_to_configured_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        lambda phone: (calls.append("chuanglan"), 60)[1])
+
+    r = sms.send_code(PHONE)
+    assert r.provider == "chuanglan"
+    assert r.fallback is True
+    assert r.cooldown_sec == 60
+    assert calls == ["chuanglan"]
+
+
+@pytest.mark.parametrize("status", [429, 400])
+def test_send_code_non_503_does_not_fall_back(monkeypatch, status):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("x", status_code=status)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        lambda phone: (calls.append("chuanglan"), 60)[1])
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert ei.value.status_code == status
+    assert calls == []
+
+
+def test_send_code_503_without_backup_reraises(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert ei.value.status_code == 503
+
+
+def test_send_code_backup_also_fails_raises_backup_error(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "send_code",
+                        _raiser(SmsError("极光不可用", status_code=503)))
+    monkeypatch.setattr(chuanglan, "send_code",
+                        _raiser(SmsError("创蓝也不可用", status_code=503)))
+
+    with pytest.raises(SmsError) as ei:
+        sms.send_code(PHONE)
+    assert "创蓝" in str(ei.value)
+
+
+def test_verify_hits_primary_without_touching_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), True)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is True
+    assert calls == ["jiguang"]
+
+
+def test_verify_falls_through_to_backup(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), False)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is True
+    assert calls == ["jiguang", "chuanglan"]
+
+
+def test_verify_all_miss_returns_false(monkeypatch):
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "chuanglan")
+    monkeypatch.setattr(jiguang, "verify_code", lambda p, c: False)
+    monkeypatch.setattr(chuanglan, "verify_code", lambda p, c: False)
+
+    assert sms.verify_code(PHONE, "123456") is False
+
+
+def test_verify_backup_not_touched_when_fallback_off(monkeypatch):
+    calls = []
+    monkeypatch.setattr(settings, "SMS_PROVIDER", "jiguang")
+    monkeypatch.setattr(settings, "SMS_FALLBACK_PROVIDER", "")
+    monkeypatch.setattr(jiguang, "verify_code",
+                        lambda p, c: (calls.append("jiguang"), False)[1])
+    monkeypatch.setattr(chuanglan, "verify_code",
+                        lambda p, c: (calls.append("chuanglan"), True)[1])
+
+    assert sms.verify_code(PHONE, "123456") is False
+    assert calls == ["jiguang"]