fix(推送): 隔离推送设备账号绑定并支持定向联调,新建22222222222测试账号 (#208 )

Co-authored-by: exinglang <exinglang@qq.com> Reviewed-on: #208 Co-authored-by: zuochenyong <zuochenyong@wonderable.ai> Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>
fix(广告发奖): eCPM 过低单份金币四舍五入成 0 时兜底发 1,避免看了广告被记 too_short 零发 (#205 )
2026-07-31 14:34:07 +08:00 · 2026-07-31 11:18:00 +08:00 · 2026-07-31 11:17:30 +08:00 · 2026-07-30 19:15:45 +08:00 · 2026-07-30 16:32:32 +08:00
29 changed files with 1682 additions and 95 deletions
@@ -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=
@@ -69,8 +72,10 @@ CHUANGLAN_SMS_TIMEOUT_SEC=10
 # 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
 # (real 模式下也跳过校验)、每次登录【都重走新手引导】,并有【每日登录上限】防被人猜到号后脚本刷。
 # 逻辑见 app/core/test_account.py,与其他业务解耦。
-# ⚠️ 留空 = 关闭整功能(生产默认);要启用才填号(如 11111111111)。改完重启生效,随时可清空停用。
+# ⚠️ 两项都留空 = 关闭整功能(生产默认)。多账号用英文逗号分隔,改完重启生效。
 TEST_ACCOUNT_PHONE=
+# 推荐新配置；例如联调环境填 11111111111,22222222222。
+TEST_ACCOUNT_PHONES=
 # 该测试号每日最多登录次数,当日超过即拒绝(429),次日归零。
 TEST_ACCOUNT_DAILY_LIMIT=500

@@ -176,6 +181,13 @@ PANGLE_REPORT_SITE_ID_TEST=5832303
 # APPLOG_MAX_BODY_BYTES=2097152            # 请求体上限 2MB(超 → 413;运行期可调)
 # APPLOG_MAX_MSG_BYTES=8192                # 单条 msg 超此字节数截断

+# ===== 荣耀 Push =====
+HONOR_PUSH_APP_ID=
+HONOR_PUSH_CLIENT_ID=
+HONOR_PUSH_CLIENT_SECRET=
+# 0=正式消息(默认)；1=测试消息(仅开发联调，生产必须保持 0)
+HONOR_PUSH_TARGET_USER_TYPE=0
+
 # ===== 华为 Push =====
 HUAWEI_PUSH_APP_ID=
 HUAWEI_PUSH_APP_SECRET=
@@ -0,0 +1,89 @@
+"""isolate vendor push binding across user accounts
+
+Revision ID: push_binding_isolation
+Revises: guide_video_ten_circle_v2
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "push_binding_isolation"
+down_revision = "guide_video_ten_circle_v2"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.add_column(sa.Column("push_binding_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(
+            sa.Column(
+                "push_binding_revoked",
+                sa.Boolean(),
+                nullable=False,
+                server_default=sa.false(),
+            )
+        )
+        batch_op.create_index(
+            "ix_device_liveness_push_binding_id",
+            ["push_binding_id"],
+            unique=False,
+        )
+
+    connection = op.get_bind()
+    device = sa.table(
+        "device_liveness",
+        sa.column("id", sa.Integer),
+        sa.column("push_vendor", sa.String),
+        sa.column("push_token", sa.String),
+        sa.column("updated_at", sa.DateTime),
+    )
+    connection.execute(
+        device.update()
+        .where(
+            sa.or_(
+                device.c.push_vendor == "",
+                device.c.push_token == "",
+            )
+        )
+        .values(push_vendor=None, push_token=None)
+    )
+
+    rows = connection.execute(
+        sa.select(device.c.id, device.c.push_vendor, device.c.push_token)
+        .where(
+            device.c.push_vendor.is_not(None),
+            device.c.push_token.is_not(None),
+        )
+        .order_by(device.c.updated_at.desc(), device.c.id.desc())
+    )
+    seen: set[tuple[str, str]] = set()
+    duplicate_ids: list[int] = []
+    for row in rows:
+        key = (row.push_vendor, row.push_token)
+        if key in seen:
+            duplicate_ids.append(row.id)
+        else:
+            seen.add(key)
+    if duplicate_ids:
+        connection.execute(
+            device.update()
+            .where(device.c.id.in_(duplicate_ids))
+            .values(push_vendor=None, push_token=None)
+        )
+
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.create_unique_constraint(
+            "uq_device_liveness_vendor_token",
+            ["push_vendor", "push_token"],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("device_liveness") as batch_op:
+        batch_op.drop_constraint("uq_device_liveness_vendor_token", type_="unique")
+        batch_op.drop_index("ix_device_liveness_push_binding_id")
+        batch_op.drop_column("push_binding_revoked")
+        batch_op.drop_column("push_binding_id")
@@ -71,8 +71,8 @@ admin_app = FastAPI(

 # admin 前端独立部署。生产同域(nginx)无需 CORS;本地 next dev 跨域需放行开发源。
 _dev_origins = [
-    "http://localhost:3002",
-    "http://127.0.0.1:3002",
+    "http://localhost:3001",
+    "http://127.0.0.1:3001",
    "http://localhost:3000",
    "http://127.0.0.1:3000",
 ]
@@ -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,
    )

@@ -256,7 +259,7 @@ def sms_login(req: SmsLoginRequest, request: Request, db: DbSession) -> TokenWit
    # 放在最前面:命中即不校验验证码;先扣当日额度,超限直接拒,挡住有人猜到号后脚本刷。
    # 测试账号走自己的每日额度、不受下面 (设备+IP) 每小时限流约束(QA 需在一小时内反复登录联调)。
    if test_account.is_test_account(req.phone):
-        if not test_account.try_consume_quota():
+        if not test_account.try_consume_quota(req.phone):
            raise HTTPException(status_code=429, detail="测试账号今日使用次数已达上限,请明天再试")
        user = user_repo.upsert_user_for_login(db, phone=req.phone, register_channel="sms")
        risk_repo.record_behavior_event(
@@ -19,8 +19,10 @@ from app.api.deps import CurrentUser, DbSession
 from app.integrations import vendor_push
 from app.repositories import device as device_repo
 from app.schemas.device import (
+    DeferredDeviceUnregisterRequest,
    DeviceOut,
    DeviceRegisterRequest,
+    DeviceUnregisterRequest,
    HeartbeatRequest,
    LivenessAckRequest,
    LivenessOut,
@@ -78,6 +80,7 @@ def register_device(
        registration_id=req.registration_id,
        push_vendor=req.push_vendor,
        push_token=req.push_token,
+        push_binding_id=req.push_binding_id,
        platform=req.platform,
        app_version=req.app_version,
    )
@@ -106,6 +109,36 @@ def report_heartbeat(
        registration_id=req.registration_id,
        push_vendor=req.push_vendor,
        push_token=req.push_token,
+        push_binding_id=req.push_binding_id,
+    )
+    return OkResponse()
+
+
+@router.post("/unregister", response_model=OkResponse, summary="解绑当前用户的本机推送目标")
+def unregister_device(
+    req: DeviceUnregisterRequest,
+    user: CurrentUser,
+    db: DbSession,
+) -> OkResponse:
+    device_repo.unregister_push_binding(db, user_id=user.id, device_id=req.device_id)
+    logger.info("device unregister user_id=%d device_id=%s", user.id, req.device_id)
+    return OkResponse()
+
+
+@router.post(
+    "/unregister/deferred",
+    response_model=OkResponse,
+    summary="离线退出后按登录会话凭据幂等解绑",
+)
+def unregister_device_deferred(
+    req: DeferredDeviceUnregisterRequest,
+    db: DbSession,
+) -> OkResponse:
+    # 不返回是否命中，避免公开接口泄露设备绑定状态。
+    device_repo.unregister_push_binding_deferred(
+        db,
+        device_id=req.device_id,
+        push_binding_id=req.push_binding_id,
    )
    return OkResponse()

@@ -79,6 +79,8 @@ class Settings(BaseSettings):
    HONOR_PUSH_APP_ID: str = ""
    HONOR_PUSH_CLIENT_ID: str = ""
    HONOR_PUSH_CLIENT_SECRET: str = ""
+    # 0=正式消息(默认)；1=测试消息(仅开发联调，勿用于生产)。
+    HONOR_PUSH_TARGET_USER_TYPE: int = Field(default=0, ge=0, le=1)
    HONOR_PUSH_TOKEN_ENDPOINT: str = "https://iam.developer.honor.com/auth/token"
    HONOR_PUSH_SEND_ENDPOINT_TEMPLATE: str = (
        "https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage"
@@ -148,6 +150,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 = ""            # 系统赠送签名(自定义签名下发易失败)
@@ -191,14 +196,27 @@ class Settings(BaseSettings):
    # (real 模式下也跳过校验)、每次登录【强制重走新手引导】,并设【每日使用次数上限】防被人
    # 猜到号后脚本滥用。两个值都能随时改 .env。逻辑全在 app/core/test_account.py,与其他业务解耦。
    # ⚠️ TEST_ACCOUNT_PHONE 留空 = 整个功能关闭(生产默认安全;要启用才显式填号)。
-    TEST_ACCOUNT_PHONE: str = ""           # 测试手机号(11 位,如 11111111111);空=关闭整功能
+    TEST_ACCOUNT_PHONE: str = ""           # 兼容旧配置:单个测试手机号
+    TEST_ACCOUNT_PHONES: str = ""          # 多个测试手机号,英文逗号分隔
    TEST_ACCOUNT_DAILY_LIMIT: int = 500    # 该测试号每日最多登录次数,当日超过即拒绝登录

    @property
    def test_account_phone(self) -> str:
-        """规整后的测试手机号(去空白);空串=功能关闭。"""
+        """兼容旧调用:规整后的单个测试手机号。"""
        return self.TEST_ACCOUNT_PHONE.strip()

+    @property
+    def test_account_phones(self) -> frozenset[str]:
+        """所有测试手机号；新旧配置取并集，便于线上平滑迁移。"""
+        phones = {
+            phone.strip()
+            for phone in self.TEST_ACCOUNT_PHONES.split(",")
+            if phone.strip()
+        }
+        if self.test_account_phone:
+            phones.add(self.test_account_phone)
+        return frozenset(phones)
+
    # ===== 美团联盟 CPS =====
    # 未配置时所有 /api/v1/meituan/* 接口 200 返空(优雅降级),不影响登录/领券等其他业务。
    MT_CPS_APP_KEY: str = ""
@@ -491,7 +509,7 @@ class Settings(BaseSettings):
        return self.APP_ENV == "prod"

    @model_validator(mode="after")
-    def _enforce_prod_secrets(self) -> "Settings":
+    def _enforce_prod_secrets(self) -> Settings:
        """prod 下强校验 JWT secret,弱/默认/空即启动报错(fail-fast,挡住 token 被伪造)。

        只校验两个签发凭证:App 用户的 JWT_SECRET_KEY、后台的 ADMIN_JWT_SECRET——它们沿用默认值
@@ -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 下线:它与广告实际收益脱钩,产品确认
@@ -9,9 +9,8 @@
  3. **每日使用次数上限**:防被人猜到这个号后写脚本一直刷。当天登录数超过上限即拒绝(429),
     次日自动归零。

-手机号与上限都在 .env 配(`TEST_ACCOUNT_PHONE` / `TEST_ACCOUNT_DAILY_LIMIT`),随时可改。
-`TEST_ACCOUNT_PHONE` 留空 = 整个功能关闭(生产默认态),`is_test_account()` 对任何号都返回
-False,登录/短信回到原逻辑,零影响。
+手机号与上限都在 .env 配(`TEST_ACCOUNT_PHONES` / `TEST_ACCOUNT_DAILY_LIMIT`),随时可改。
+兼容旧的单号配置 `TEST_ACCOUNT_PHONE`;两项都留空 = 整个功能关闭(生产默认态)。

 计数存**进程内存**(单 worker 够用,与 sms.py 同款约定):重启清零、多 worker 不共享。作为
 一个测试号的粗粒度防滥用闸够用;且因所有登录都落同一个 phone → 同一个 user,滥用面天然只
@@ -27,9 +26,9 @@ from app.core.config import settings

 logger = logging.getLogger("shagua.test_account")

-# 进程内每日计数:(date_str, 当日已登录次数)。单 worker 有效,重启清零(见模块 docstring)。
+# 进程内每日计数:{手机号: (date_str, 当日已登录次数)}。单 worker 有效,重启清零。
 _lock = Lock()
-_usage: tuple[str, int] = ("", 0)
+_usage: dict[str, tuple[str, int]] = {}


 def _today() -> str:
@@ -37,16 +36,16 @@ def _today() -> str:


 def is_enabled() -> bool:
-    """功能总开关:配了 TEST_ACCOUNT_PHONE 才启用(空=关闭)。"""
-    return bool(settings.test_account_phone)
+    """功能总开关:至少配置了一个测试手机号才启用。"""
+    return bool(settings.test_account_phones)


 def is_test_account(phone: str) -> bool:
-    """该手机号是否为配置的测试账号。功能关闭时对任何号都返回 False。"""
-    return is_enabled() and phone == settings.test_account_phone
+    """该手机号是否在测试账号集合中。"""
+    return phone in settings.test_account_phones


-def try_consume_quota() -> bool:
+def try_consume_quota(phone: str) -> bool:
    """测试账号登录时调:当日计数 +1。

    Returns:
@@ -59,17 +58,17 @@ def try_consume_quota() -> bool:
    limit = settings.TEST_ACCOUNT_DAILY_LIMIT
    with _lock:
        today = _today()
-        day, cnt = _usage
+        day, cnt = _usage.get(phone, ("", 0))
        if day != today:  # 跨天归零
            cnt = 0
        if cnt >= limit:
-            _usage = (today, cnt)  # 已满,保持不变
+            _usage[phone] = (today, cnt)  # 已满,保持不变
            logger.warning(
-                "测试账号 %s 今日登录数已达上限 %d,拒绝", settings.test_account_phone, limit
+                "测试账号 %s 今日登录数已达上限 %d,拒绝", phone, limit
            )
            return False
-        _usage = (today, cnt + 1)
-        logger.info("测试账号 %s 第 %d/%d 次登录", settings.test_account_phone, cnt + 1, limit)
+        _usage[phone] = (today, cnt + 1)
+        logger.info("测试账号 %s 第 %d/%d 次登录", phone, cnt + 1, limit)
        return True


@@ -77,4 +76,4 @@ def _reset_for_test() -> None:
    """仅供单测:清空进程内计数,隔离用例间状态。"""
    global _usage
    with _lock:
-        _usage = ("", 0)
+        _usage = {}
@@ -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
@@ -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, "发送过于频繁,请稍后再试"),
 }
@@ -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()
@@ -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({
@@ -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)
@@ -224,6 +224,10 @@ def send_data_event(
    if vendor == "vivo":
        return {"skipped": True, "reason": "foreground notification callback"}

+    # 小米服务端不再发送透传消息；通知栏消息仍照常发送，前台角标由客户端主动刷新。
+    if vendor == "xiaomi":
+        return {"skipped": True, "reason": "xiaomi data messages disabled"}
+
    # OPush 当前只支持通知栏消息，没有服务端透传单推接口。
    # 客户端在 OPPO 首页可见时轮询未读数；这里必须安全跳过，不能请求不存在的
    # /message/transparent/unicast（该地址会稳定返回 HTTP 404）。
@@ -233,7 +237,6 @@ def send_data_event(
    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)

@@ -520,7 +523,7 @@ def _send_honor(token: str, title: str, body: str, extras: dict[str, str]) -> di
        "notification": {"title": title, "body": body},
        "android": {
            "ttl": f"{settings.PUSH_TIME_TO_LIVE_SEC}s",
-            "targetUserType": 1,
+            "targetUserType": settings.HONOR_PUSH_TARGET_USER_TYPE,
            "notification": {
                "title": title,
                "body": body,
@@ -34,6 +34,11 @@ class DeviceLiveness(Base):
    __tablename__ = "device_liveness"
    __table_args__ = (
        UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
+        UniqueConstraint(
+            "push_vendor",
+            "push_token",
+            name="uq_device_liveness_vendor_token",
+        ),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -48,6 +53,11 @@ class DeviceLiveness(Base):
    push_vendor: Mapped[str | None] = mapped_column(String(32), nullable=True)
    # 厂商 push token / regId / registration_id;不同厂商命名不同,后端统一存这里。
    push_token: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    # 客户端每次登录生成的高熵随机值。离线退出后只凭此值精确撤销旧会话绑定，
+    # 新账号/新会话注册时会替换它，故旧解绑任务绝不能误解绑新登录。
+    push_binding_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    # 被退出/转移的 binding 留作墓碑，阻止旧的在途 register/heartbeat 把 token 抢回。
+    push_binding_revoked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
    platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android")
    app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)

@@ -1,9 +1,9 @@
 """device 表读写(设备注册 / 心跳 / 超时扫描)。"""
 from __future__ import annotations

-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta

-from sqlalchemy import select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.orm import Session

 from app.models.device import DeviceLiveness
@@ -24,13 +24,29 @@ def register_or_update(
    registration_id: str | None = None,
    push_vendor: str | None = None,
    push_token: str | None = None,
+    push_binding_id: str | None = None,
    platform: str = "android",
    app_version: str | None = None,
 ) -> DeviceLiveness:
-    """注册设备或更新其厂商 push token / 元信息。upsert by (user_id, device_id)。"""
+    """注册设备或更新推送目标；同一设备/token 的旧用户绑定在同一事务内失效。"""
    normalized_vendor = _normalize_push_vendor(push_vendor)
    normalized_token = push_token.strip() if push_token else None
+    normalized_binding = push_binding_id.strip() if push_binding_id else None
    device = _get(db, user_id=user_id, device_id=device_id)
+    stale_binding = bool(
+        device
+        and normalized_binding
+        and device.push_binding_id == normalized_binding
+        and device.push_binding_revoked
+    )
+    if not stale_binding:
+        _release_conflicting_push_bindings(
+            db,
+            user_id=user_id,
+            device_id=device_id,
+            push_vendor=normalized_vendor,
+            push_token=normalized_token,
+        )
    if device is None:
        device = DeviceLiveness(
            user_id=user_id,
@@ -38,17 +54,22 @@ def register_or_update(
            registration_id=registration_id,
            push_vendor=normalized_vendor,
            push_token=normalized_token,
+            push_binding_id=normalized_binding,
+            push_binding_revoked=False,
            platform=platform or "android",
            app_version=app_version,
        )
        db.add(device)
    else:
-        if registration_id:
+        if registration_id and not stale_binding:
            device.registration_id = registration_id
-        if normalized_vendor:
+        if normalized_vendor and not stale_binding:
            device.push_vendor = normalized_vendor
-        if normalized_token:
+        if normalized_token and not stale_binding:
            device.push_token = normalized_token
+        if normalized_binding and not stale_binding:
+            device.push_binding_id = normalized_binding
+            device.push_binding_revoked = False
        if platform:
            device.platform = platform
        if app_version:
@@ -67,26 +88,45 @@ def touch_heartbeat(
    registration_id: str | None = None,
    push_vendor: str | None = None,
    push_token: str | None = None,
+    push_binding_id: str | None = None,
 ) -> DeviceLiveness:
    """处理一次心跳(心跳也能自注册)。

    service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、
    清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。
    """
-    now = datetime.now(timezone.utc)
+    now = datetime.now(UTC)
    device = _get(db, user_id=user_id, device_id=device_id)
+    normalized_vendor = _normalize_push_vendor(push_vendor)
+    normalized_token = push_token.strip() if push_token else None
+    normalized_binding = push_binding_id.strip() if push_binding_id else None
+    stale_binding = bool(
+        device
+        and normalized_binding
+        and device.push_binding_id == normalized_binding
+        and device.push_binding_revoked
+    )
+    if (normalized_token or normalized_binding) and not stale_binding:
+        _release_conflicting_push_bindings(
+            db,
+            user_id=user_id,
+            device_id=device_id,
+            push_vendor=normalized_vendor,
+            push_token=normalized_token,
+        )
    if device is None:
        device = DeviceLiveness(user_id=user_id, device_id=device_id)
        db.add(device)

-    if registration_id:
+    if registration_id and not stale_binding:
        device.registration_id = registration_id
-    normalized_vendor = _normalize_push_vendor(push_vendor)
-    normalized_token = push_token.strip() if push_token else None
-    if normalized_vendor:
+    if normalized_vendor and not stale_binding:
        device.push_vendor = normalized_vendor
-    if normalized_token:
+    if normalized_token and not stale_binding:
        device.push_token = normalized_token
+    if normalized_binding and not stale_binding:
+        device.push_binding_id = normalized_binding
+        device.push_binding_revoked = False
    device.last_report_protection_on = accessibility_enabled

    if accessibility_enabled:
@@ -107,7 +147,7 @@ def list_overdue(db: Session, *, timeout_minutes: int) -> list[DeviceLiveness]:

    即使没有厂商 token 也要检出,后续由 kill_alert_pending 走客户端进 App 后兜底提醒。
    """
-    cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+    cutoff = datetime.now(UTC) - timedelta(minutes=timeout_minutes)
    stmt = select(DeviceLiveness).where(
        DeviceLiveness.ever_protected.is_(True),
        DeviceLiveness.liveness_state == "alive",
@@ -126,7 +166,7 @@ def mark_notified(db: Session, *, device_id_pk: int) -> None:
    device = db.get(DeviceLiveness, device_id_pk)
    if device is not None:
        device.liveness_state = "notified"
-        device.notified_at = datetime.now(timezone.utc)
+        device.notified_at = datetime.now(UTC)
        device.kill_alert_pending = True
        db.commit()

@@ -177,6 +217,69 @@ def list_push_targets(db: Session, *, user_id: int) -> list[DeviceLiveness]:
    return targets


+def unregister_push_binding(db: Session, *, user_id: int, device_id: str) -> None:
+    """当前用户主动退出：清推送归属但保留设备存活历史。"""
+    device = _get(db, user_id=user_id, device_id=device_id)
+    if device is not None:
+        _clear_push_binding(device)
+        db.commit()
+
+
+def unregister_push_binding_deferred(
+    db: Session, *, device_id: str, push_binding_id: str
+) -> None:
+    """离线退出补偿；只撤销完全匹配的旧登录会话，且始终幂等。"""
+    stmt = select(DeviceLiveness).where(
+        DeviceLiveness.device_id == device_id,
+        DeviceLiveness.push_binding_id == push_binding_id,
+    )
+    device = db.execute(stmt).scalar_one_or_none()
+    if device is not None:
+        _clear_push_binding(device)
+        db.commit()
+
+
+def _release_conflicting_push_bindings(
+    db: Session,
+    *,
+    user_id: int,
+    device_id: str,
+    push_vendor: str | None,
+    push_token: str | None,
+) -> None:
+    """把设备或 token 从其他记录转移走，当前 (user,device) 行除外。"""
+    conflicts = [DeviceLiveness.device_id == device_id]
+    if push_vendor and push_token:
+        conflicts.append(
+            and_(
+                DeviceLiveness.push_vendor == push_vendor,
+                DeviceLiveness.push_token == push_token,
+            )
+        )
+    stmt = select(DeviceLiveness).where(
+        or_(*conflicts),
+        ~and_(
+            DeviceLiveness.user_id == user_id,
+            DeviceLiveness.device_id == device_id,
+        ),
+    )
+    released = False
+    for old in db.execute(stmt).scalars():
+        _clear_push_binding(old)
+        released = True
+    if released:
+        # 唯一约束下必须先落旧行清理，再把同一 token 写给新行。
+        db.flush()
+
+
+def _clear_push_binding(device: DeviceLiveness) -> None:
+    device.registration_id = None
+    device.push_vendor = None
+    device.push_token = None
+    # binding_id 留作撤销墓碑；同一旧会话的在途请求不能重新认领 token。
+    device.push_binding_revoked = bool(device.push_binding_id)
+
+
 def _normalize_push_vendor(push_vendor: str | None) -> str | None:
    if not push_vendor:
        return None
@@ -7,9 +7,19 @@
 """
 from __future__ import annotations

+import re
 from datetime import datetime

-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.core.config import settings
+
+
+def _validate_login_phone(phone: str) -> str:
+    """普通号保持大陆手机号格式；仅显式配置的测试号允许例外。"""
+    if re.fullmatch(r"^1\d{10}$", phone) or phone in settings.test_account_phones:
+        return phone
+    raise ValueError("invalid phone")

 # ===== 用户对外信息 =====

@@ -72,7 +82,7 @@ class JverifyLoginRequest(BaseModel):
 # ===== 短信验证码 =====

 class SmsSendRequest(BaseModel):
-    phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
+    phone: str = Field(..., min_length=11, max_length=11)
    device_id: str = Field(
        "", max_length=64,
        description="硬件级设备标识(Android ANDROID_ID),用于发码防刷按 设备+IP 限流;空=按 IP 聚一桶",
@@ -80,6 +90,7 @@ class SmsSendRequest(BaseModel):
    device_model: str = Field(
        "", max_length=128, description="客户端设备型号快照,用于短信安全审计"
    )
+    _valid_phone = field_validator("phone")(_validate_login_phone)


 class SmsSendResponse(BaseModel):
@@ -89,7 +100,7 @@ class SmsSendResponse(BaseModel):


 class SmsLoginRequest(BaseModel):
-    phone: str = Field(..., min_length=11, max_length=11, pattern=r"^1\d{10}$")
+    phone: str = Field(..., min_length=11, max_length=11)
    code: str = Field(..., min_length=4, max_length=8)
    device_id: str = Field(
        "", max_length=64,
@@ -98,6 +109,7 @@ class SmsLoginRequest(BaseModel):
    device_model: str = Field(
        "", max_length=128, description="客户端设备型号快照,用于短信验证安全审计"
    )
+    _valid_phone = field_validator("phone")(_validate_login_phone)


 # ===== Refresh =====
@@ -12,6 +12,7 @@ class DeviceRegisterRequest(BaseModel):
    registration_id: str | None = None
    push_vendor: str | None = None
    push_token: str | None = None
+    push_binding_id: str | None = Field(default=None, max_length=128)
    platform: str = "android"
    app_version: str | None = None

@@ -23,6 +24,7 @@ class HeartbeatRequest(BaseModel):
    registration_id: str | None = None
    push_vendor: str | None = None
    push_token: str | None = None
+    push_binding_id: str | None = Field(default=None, max_length=128)


 class DeviceOut(BaseModel):
@@ -33,6 +35,7 @@ class DeviceOut(BaseModel):
    registration_id: str | None
    push_vendor: str | None
    push_token: str | None
+    push_binding_id: str | None
    ever_protected: bool
    liveness_state: str
    last_heartbeat_at: datetime | None
@@ -43,6 +46,14 @@ class OkResponse(BaseModel):
    ok: bool = True


+class DeviceUnregisterRequest(BaseModel):
+    device_id: str = Field(min_length=1, max_length=128)
+
+
+class DeferredDeviceUnregisterRequest(DeviceUnregisterRequest):
+    push_binding_id: str = Field(min_length=32, max_length=128)
+
+
 class LivenessOut(BaseModel):
    """本机掉线告警状态(后置检测 pull 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导,
    故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state 等)。从未注册过 → 默认 False(无告警)。"""
@@ -83,6 +83,7 @@ def _dispatch(
    extra: dict[str, str] | None = None,
    dedup_key: str | None = None,
    push_vars: dict[str, str] | None = None,
+    push_token_contains: str | None = None,
 ) -> Notification | None:
    """落一条站内消息并向该用户设备直推。返回落库行;去重命中/失败返回 None。"""
    logger.info(
@@ -119,11 +120,16 @@ def _dispatch(
        "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)
+    _push_to_user_devices(db, row, push_vars, push_token_contains)
    return row


-def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, str] | None) -> None:
+def _push_to_user_devices(
+    db: Session,
+    row: Notification,
+    push_vars: dict[str, str] | None,
+    push_token_contains: str | None = None,
+) -> None:
    """向消息归属用户的全部厂商推送目标直推(best-effort,单设备失败不影响其余)。"""
    try:
        title, body = catalog.render_push(row.type, push_vars)
@@ -133,9 +139,17 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
        extras["notificationId"] = str(row.id)

        targets = device_repo.list_push_targets(db, user_id=row.user_id)
+        if push_token_contains:
+            normalized_filter = push_token_contains.casefold()
+            targets = [
+                target
+                for target in targets
+                if normalized_filter in (target.push_token or "").casefold()
+            ]
        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),
+            "push targets resolved user_id=%s type=%s notification_id=%s "
+            "token_filter=%s target_count=%s",
+            row.user_id, row.type, row.id, bool(push_token_contains), len(targets),
        )
        if not targets:
            logger.warning(
@@ -230,7 +244,12 @@ def _push_to_user_devices(db: Session, row: Notification, push_vars: dict[str, s
 # ---------------------------------------------------------------------------


-def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None:
+def notify_withdraw_success(
+    db: Session,
+    order: WithdrawOrder,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """#3 提现成功:款项已存入微信零钱。点击无跳转仅消红点(extra 空)。"""
    _dispatch(
        db,
@@ -244,10 +263,16 @@ def notify_withdraw_success(db: Session, order: WithdrawOrder) -> None:
        extra={},
        dedup_key=order.out_bill_no,
        push_vars={"amount": notif_repo.cash_yuan(order.amount_cents)},
+        push_token_contains=push_token_contains,
    )


-def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None:
+def notify_withdraw_failed(
+    db: Session,
+    order: WithdrawOrder,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """#4 提现失败/退回:含微信侧失败、审核拒绝、解绑退回。点击跳提现页重新提现。

    失败原因用 order.fail_reason(与 /withdraw/status 下发的用户可读原因同源)。
@@ -265,10 +290,16 @@ def notify_withdraw_failed(db: Session, order: WithdrawOrder) -> None:
        extra={"withdrawId": order.out_bill_no},
        dedup_key=order.out_bill_no,
        push_vars={"amount": notif_repo.cash_yuan(order.amount_cents), "reason": reason},
+        push_token_contains=push_token_contains,
    )


-def notify_feedback_reply(db: Session, feedback: Feedback) -> None:
+def notify_feedback_reply(
+    db: Session,
+    feedback: Feedback,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """#9 官方回复:运营审核了反馈且未采纳(用户可见原因/留言落在反馈记录上)。
    点击跳反馈历史页滚动高亮该条(extra.feedbackId)。"""
    _dispatch(
@@ -278,10 +309,16 @@ def notify_feedback_reply(db: Session, feedback: Feedback) -> None:
        info_rows=[{"label": "说明文案", "value": "快去看看官方给您的回复吧～"}],
        extra={"feedbackId": str(feedback.id)},
        dedup_key=str(feedback.id),
+        push_token_contains=push_token_contains,
    )


-def notify_feedback_reward(db: Session, feedback: Feedback) -> None:
+def notify_feedback_reward(
+    db: Session,
+    feedback: Feedback,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """#10 反馈奖励:反馈被采纳,金币已到账。PRD 约定发奖必带官方留言(admin_reply);
    运营漏填时省略该信息行,不硬造文案。"""
    coins = int(feedback.reward_coins or 0)
@@ -299,10 +336,16 @@ def notify_feedback_reward(db: Session, feedback: Feedback) -> None:
        extra={"feedbackId": str(feedback.id)},
        dedup_key=str(feedback.id),
        push_vars={"coins": str(coins)},
+        push_token_contains=push_token_contains,
    )


-def notify_report_approved(db: Session, report: PriceReport) -> None:
+def notify_report_approved(
+    db: Session,
+    report: PriceReport,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """#11 爆料审核通过:上报的更低价过审,金币已到账。点击跳爆料记录页高亮该条。"""
    coins = int(report.reward_coins or 0)
    store = (report.store_name or "").strip() or "该店铺"
@@ -318,11 +361,17 @@ def notify_report_approved(db: Session, report: PriceReport) -> None:
        extra={"reportId": str(report.id)},
        dedup_key=str(report.id),
        push_vars={"store": store, "coins": str(coins)},
+        push_token_contains=push_token_contains,
    )


 def notify_invite_order_reward(
-    db: Session, *, inviter_user_id: int, invitee_user_id: int, cash_cents: int
+    db: Session,
+    *,
+    inviter_user_id: int,
+    invitee_user_id: int,
+    cash_cents: int,
+    push_token_contains: str | None = None,
 ) -> None:
    """#12 好友下单到账:被邀请好友完成首次下单(比价),现金奖励已入邀请人账户。
    通知发给【邀请人】;每个好友只发一次奖 → dedup 按被邀请人。"""
@@ -340,4 +389,5 @@ def notify_invite_order_reward(
        extra={"inviteeNickname": nickname},
        dedup_key=str(invitee_user_id),
        push_vars={"nickname": nickname, "amount": _yuan_trim(cash_cents)},
+        push_token_contains=push_token_contains,
    )
@@ -87,6 +87,7 @@ PRD §5 的 13 条 push 文案(标题固定 ≤11 字不带变量;正文 `{var}`

 **真发注意**:
 - 目标手机必须先装 App 且客户端已集成对应厂商 SDK、`/device/register` 上报过 token;
+- 荣耀正式环境必须使用正式消息(`HONOR_PUSH_TARGET_USER_TYPE=0`);开发联调测试设备时才设为 `1`;
 - vivo 未上架前走测试推送(`VIVO_PUSH_MODE=1`),目标手机需在 vivo 开放平台加入测试设备;
 - 小米新设备需在开放平台把签名/包名配好,token 才有效。

@@ -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 一提交。
@@ -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=""` 即运行时秒回退到"纯极光"，无需改代码/迁移。
@@ -16,6 +16,10 @@ xiaomi/oppo/vivo)到该用户 device_liveness 里已注册的 push token。
  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --count 1          # 各 1 条(先小量验证通道)
  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --types withdraw_failed --count 3
  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --types all --count 2
+  .venv\\Scripts\\python.exe scripts\\fire_push_events.py --phone 22222222222 --token ABC123 --count 1
+
+`--token` 是可选、不区分大小写的 token 子串过滤器;会推送给该账号下所有 token 包含该
+子串的设备。没有匹配设备时脚本直接停止,不会生成站内消息。

 推送成败看输出里的 `shagua.vendor_push` 日志(push sent / push failed);2 台设备则每条各推 2 次。
 凭据缺失或 token 失效时 notification_events 只记日志、不抛错(站内消息仍会落库)。
@@ -58,33 +62,52 @@ _FAIL_REASONS = [
 ]


-def _fire_one(db, uid: int, type_key: str, i: int) -> None:
+def _fire_one(
+    db,
+    uid: int,
+    type_key: str,
+    i: int,
+    *,
+    push_token_contains: str | None = None,
+) -> None:
    """构造一条该类型的瞬态业务对象(不落业务表,只为给 notify 函数读字段),触发真实推送。"""
    if type_key == "withdraw_success":
        order = WithdrawOrder(user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=50, source="coin_cash")
-        notification_events.notify_withdraw_success(db, order)
+        notification_events.notify_withdraw_success(
+            db, order, push_token_contains=push_token_contains
+        )
    elif type_key == "withdraw_failed":
        order = WithdrawOrder(
            user_id=uid, out_bill_no=uuid.uuid4().hex, amount_cents=350, source="coin_cash",
            fail_reason=random.choice(_FAIL_REASONS),
        )
-        notification_events.notify_withdraw_failed(db, order)
+        notification_events.notify_withdraw_failed(
+            db, order, push_token_contains=push_token_contains
+        )
    elif type_key == "invite_order_reward":
        # 假被邀请人 id(> 真实用户范围,避重):昵称回退「好友」。真实昵称请走 API 流程(见文末说明)。
        fake_invitee = random.randint(900000, 999999)
        notification_events.notify_invite_order_reward(
-            db, inviter_user_id=uid, invitee_user_id=fake_invitee, cash_cents=INVITE_COMPARE_REWARD_CENTS
+            db,
+            inviter_user_id=uid,
+            invitee_user_id=fake_invitee,
+            cash_cents=INVITE_COMPARE_REWARD_CENTS,
+            push_token_contains=push_token_contains,
        )
    elif type_key == "feedback_reward":
        fb = Feedback(user_id=uid, content="(直发)", contact="", status="adopted",
                      reward_coins=300, admin_reply="感谢反馈,您说的问题已修复上线,金币请查收～")
        fb.id = random.randint(900000, 999999)
-        notification_events.notify_feedback_reward(db, fb)
+        notification_events.notify_feedback_reward(
+            db, fb, push_token_contains=push_token_contains
+        )
    elif type_key == "feedback_reply":
        fb = Feedback(user_id=uid, content="(直发)", contact="", status="rejected",
                      admin_reply="您的建议我们记录啦,会在后续版本评估~")
        fb.id = random.randint(900000, 999999)
-        notification_events.notify_feedback_reply(db, fb)
+        notification_events.notify_feedback_reply(
+            db, fb, push_token_contains=push_token_contains
+        )
    elif type_key == "report_approved":
        rep = PriceReport(
            user_id=uid, reported_platform_id="jd", reported_platform_name="京东外卖",
@@ -92,7 +115,9 @@ def _fire_one(db, uid: int, type_key: str, i: int) -> None:
            reward_coins=PRICE_REPORT_REWARD_COINS, store_name=f"测试火锅店{i:02d}",
        )
        rep.id = random.randint(900000, 999999)
-        notification_events.notify_report_approved(db, rep)
+        notification_events.notify_report_approved(
+            db, rep, push_token_contains=push_token_contains
+        )
    else:
        raise SystemExit(f"未知类型: {type_key}(可选: {', '.join(ALL_TYPES)})")

@@ -100,6 +125,11 @@ def _fire_one(db, uid: int, type_key: str, i: int) -> None:
 def main() -> None:
    parser = argparse.ArgumentParser(description="直接触发消息通知中心真实推送(后台驱动不了的事件用)")
    parser.add_argument("--phone", default=DEFAULT_PHONE, help=f"目标用户手机号(默认 {DEFAULT_PHONE})")
+    parser.add_argument(
+        "--token",
+        default="",
+        help="可选的厂商 token 子串(不区分大小写);仅推送到该账号下所有匹配设备",
+    )
    parser.add_argument("--count", type=int, default=10, help="每类发多少条(默认 10)")
    parser.add_argument(
        "--types", default=",".join(DEFAULT_TYPES),
@@ -122,14 +152,40 @@ def main() -> None:
        uid = user.id

        targets = device_repo.list_push_targets(db, user_id=uid)
-        print(f"目标用户 {args.phone}(id={uid});已注册推送设备 {len(targets)} 台:"
-              f"{[t.push_vendor for t in targets] or '无(手机收不到!先在 App 上报 push token)'}")
+        token_filter = args.token.strip()
+        normalized_filter = token_filter.casefold()
+        matched_targets = [
+            target
+            for target in targets
+            if not token_filter
+            or normalized_filter in (target.push_token or "").casefold()
+        ]
+        if token_filter and not matched_targets:
+            print(
+                f"❌ 用户 {args.phone} 的 {len(targets)} 台有效推送设备中，"
+                f"没有 token 包含 {token_filter!r}；未生成站内消息，也未发送推送。"
+            )
+            return
+
+        print(
+            f"目标用户 {args.phone}(id={uid});有效推送设备 {len(targets)} 台;"
+            f"本次匹配 {len(matched_targets)} 台:"
+            f"{[target.push_vendor for target in matched_targets] or '无(手机收不到!先在 App 上报 push token)'}"
+        )
+        if token_filter:
+            print(f"token 子串过滤(不区分大小写): {token_filter!r}")
        print(f"即将触发:{types},每类 {args.count} 条 → 共 {len(types) * args.count} 条\n")

        for t in types:
            print(f"── {t} ×{args.count} " + "─" * 30)
            for i in range(1, args.count + 1):
-                _fire_one(db, uid, t, i)
+                _fire_one(
+                    db,
+                    uid,
+                    t,
+                    i,
+                    push_token_contains=token_filter or None,
+                )

        print(f"\n✅ 已触发完。站内消息已落 notification 表(用 {args.phone} 登录 App 可在消息中心看到);"
              "\n   手机推送成败见上方 `shagua.vendor_push` 日志(push sent=成功 / push failed=失败)。")
@@ -15,6 +15,10 @@ services/notification_events.notify_invite_order_reward —— 与生产同一
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py                        # 随机金额发 1 条
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --cents 200            # 固定 2.00 元
  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --invitee-phone 12000000001  # 真实昵称
+  .venv\\Scripts\\python.exe scripts\\test_push_invite_order_reward.py --phone 22222222222 --token gkha
+
+`--token` 是可选、不区分大小写的 token 子串过滤器；会推送给该账号下所有 token 包含该
+子串的设备。没有匹配设备时脚本直接停止，不会生成站内消息。

 结果判读(看输出日志):
  push sent   = 厂商接口受理成功,手机应弹「好友下单奖励到账」通知
@@ -62,6 +66,11 @@ def _notif_count(db, uid: int) -> int:
 def main() -> None:
    parser = argparse.ArgumentParser(description="#12 好友下单到账 推送联调(每次 1 条,金额默认随机)")
    parser.add_argument("--phone", default="11111111111", help="邀请人(收通知方)手机号,默认 11111111111")
+    parser.add_argument(
+        "--token",
+        default="",
+        help="可选的厂商 token 子串(不区分大小写);仅推送到该账号下所有匹配设备",
+    )
    parser.add_argument("--cents", type=int, default=None,
                        help="奖励金额,单位分(默认随机 1~9999;线上真实值 200)")
    parser.add_argument("--invitee-phone", default="",
@@ -87,13 +96,37 @@ def main() -> None:
            invitee_id = random.randint(900000, 999999)  # 假 id,昵称兜底「好友」,永不去重

        targets = device_repo.list_push_targets(db, user_id=user.id)
-        vendors = [t.push_vendor for t in targets]
-        print(f"邀请人 {args.phone}(id={user.id});推送设备 {len(targets)} 台:{vendors or '无 ← 手机收不到!先在 App 上报 push token'}")
+        token_filter = args.token.strip()
+        normalized_filter = token_filter.casefold()
+        matched_targets = [
+            target
+            for target in targets
+            if not token_filter
+            or normalized_filter in (target.push_token or "").casefold()
+        ]
+        if token_filter and not matched_targets:
+            print(
+                f"❌ 用户 {args.phone} 的 {len(targets)} 台有效推送设备中，"
+                f"没有 token 包含 {token_filter!r}；未生成站内消息，也未发送推送。"
+            )
+            return
+
+        print(
+            f"邀请人 {args.phone}(id={user.id});有效推送设备 {len(targets)} 台;"
+            f"本次匹配 {len(matched_targets)} 台:"
+            f"{[target.push_vendor for target in matched_targets] or '无 ← 手机收不到!先在 App 上报 push token'}"
+        )
+        if token_filter:
+            print(f"token 子串过滤(不区分大小写): {token_filter!r}")

        before = _notif_count(db, user.id)
        print(f"→ 本次奖励 【{cents / 100:.2f} 元】(invitee_user_id={invitee_id}),手机上按金额认领这条通知")
        notification_events.notify_invite_order_reward(
-            db, inviter_user_id=user.id, invitee_user_id=invitee_id, cash_cents=cents
+            db,
+            inviter_user_id=user.id,
+            invitee_user_id=invitee_id,
+            cash_cents=cents,
+            push_token_contains=token_filter or None,
        )

        created = _notif_count(db, user.id) - before
@@ -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()
@@ -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
@@ -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"
@@ -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 路由,运行时可切

@@ -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"]
Author	SHA1	Message	Date
zuochenyong	d9177cbb71	fix(推送): 隔离推送设备账号绑定并支持定向联调,新建22222222222测试账号 (#208 ) Co-authored-by: exinglang <exinglang@qq.com> Reviewed-on: #208 Co-authored-by: zuochenyong <zuochenyong@wonderable.ai> Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>	2026-07-31 14:34:07 +08:00
guke	eeecb5faf0	fix(广告发奖): eCPM 过低单份金币四舍五入成 0 时兜底发 1,避免看了广告被记 too_short 零发 (#205 ) ## 背景 / 问题线上 record 4667:用户在比价等候期看满一条低 eCPM(¥0.43 CPM)的信息流广告(≥10s), 但单份金币按公式 `0.43/1000 × 0.1(档位因子) × 1.0(LT因子) × 10000 = 0.43`,四舍五入成 0。在 `grant_feed_reward` 里 `unit_cap=0` → `coin<=0` → 记 `too_short` 零发。结果:用户看满了广告却什么都没拿到,还被标成"时长不足"。 ## 改动 `app/core/rewards.py` 的 `calculate_ad_reward_coin`(发奖与后台审计对账的唯一口径): - 有真实正 eCPM 时,单份金币 floor 到 1(`max(0, …)` → `max(1, …)`)。 - eCPM 缺失 / 为 0 / 非法(没有真实广告价值)时提前 `return 0` —— 不凭空铸币、不破坏 `ecpm_missing` 语义。关键设计 if ecpm_yuan <= 0: return 0 是防铸币防线:没它的话 reward_video 路径 ecpm="0" 会被 floor 成 1。不影响防刷:上限仍由 AD_ECPM_MAX_FEN(¥500 CPM)钳顶;LT 因子最低 1.0、从不归零,限流靠每日 500 次上限。正常量级 eCPM 本就 ≥1,下限不改变其取值(纯回归保护)。 --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #205	2026-07-31 11:18:00 +08:00
guke	31bff63ed4	feat(sms): 短信验证码极光→创蓝容灾 fallback (#204 ) 背景短信验证码已是可切换 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	2026-07-31 11:17:30 +08:00
guke	ef0ab9d95a	fix(短信): 号码无效提示由「手机号无效」改为「请输入有效的手机号」 (#206 ) 极光/阿里云/创蓝三处 provider 的号码无效错误统一改成更友好的用户提示。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #206	2026-07-30 19:15:45 +08:00
zuochenyong	06cd718610	修复之前错误配置的端口 (#203 ) Co-authored-by: exinglang <exinglang@qq.com> Reviewed-on: #203 Co-authored-by: zuochenyong <zuochenyong@wonderable.ai> Co-committed-by: zuochenyong <zuochenyong@wonderable.ai>	2026-07-30 16:32:32 +08:00