feat(auth): M2 phone_rebind_log 换绑台账模型 + 配置 + 迁移

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

alembic/versions/phone_rebind_log.py

@@ -0,0 +1,32 @@
+"""phone_rebind_log 表(M2 换绑 30 天限制台账)
+
+Revision ID: phone_rebind_log
+Revises: comparison_llm_cost
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "phone_rebind_log"
+down_revision = "comparison_llm_cost"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "phone_rebind_log",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("phone", sa.String(length=20), nullable=False),
+        sa.Column("old_user_id", sa.Integer(), nullable=True),
+        sa.Column("new_user_id", sa.Integer(), nullable=False),
+        sa.Column("source", sa.String(length=32), nullable=False, server_default="wechat_conflict"),
+        sa.Column("rebound_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
+    )
+    op.create_index("ix_phone_rebind_log_phone", "phone_rebind_log", ["phone"])
+    op.create_index("ix_phone_rebind_log_rebound_at", "phone_rebind_log", ["rebound_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_phone_rebind_log_rebound_at", table_name="phone_rebind_log")
+    op.drop_index("ix_phone_rebind_log_phone", table_name="phone_rebind_log")
+    op.drop_table("phone_rebind_log")

app/core/config.py

@@ -44,9 +44,11 @@ class Settings(BaseSettings):
     JWT_ALGORITHM: str = "HS256"
     JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int = 120
     JWT_REFRESH_TOKEN_EXPIRE_DAYS: int = 30
-    # 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(分钟)。
-    # 见 auth.wechat_login / security.create_bind_ticket。
+    # 微信登录未命中 openid 时签发的"待绑手机"令牌有效期(JWT_SECRET_KEY 签名,typ=wechat_bind;
+    # 见 security.create_bind_ticket)。需覆盖"授权→输手机号→收短信→输验证码"整个绑定流程。
     WECHAT_BIND_TICKET_EXPIRE_MINUTES: int = 10
+    # 一个手机号 30 天内最多换绑一次(微信占用冲突页的"换绑"动作)。见 phone_rebind_log。
+    PHONE_REBIND_LIMIT_DAYS: int = 30
 
     # ===== Admin 后台 =====
     # admin 用独立 JWT secret(≠ JWT_SECRET_KEY),App 用户 token 无法越权访问后台。
@@ -84,6 +86,7 @@ class Settings(BaseSettings):
     SMS_SIGN_ID: int = 31729       # 极光短信签名 ID(非机密,可被 .env 覆盖)
     SMS_TEMPLATE_ID: int = 1       # 极光短信模板 ID(变量名 code,有效期 5 分钟)
     SMS_CODE_LENGTH: int = 6       # 验证码位数(本服务生成;前端 code 字段 4-8 位兼容)
+    SMS_DAILY_LIMIT_PER_PHONE: int = 10   # 单手机号每日发送上限(防刷 + 控费)
     SMS_MAX_VERIFY_ATTEMPTS: int = 5      # 单个验证码最多校验失败次数,超过即作废(防爆破)
 
     # ===== 测试账号(release 包全流程联调用)=====
@@ -109,6 +112,9 @@ class Settings(BaseSettings):
     # 美团调用走的代理。本机开发直连美团会 SSL EOF,需填 http://127.0.0.1:7897;
     # 线上国内服务器留空(=直连)。见 .env.example 与 integrations/meituan.py。
     MT_CPS_PROXY: str = ""
+    # 本地开发:开启后 /feed 接口直接返回 mock 数据,不调美团 API、不查离线库,
+    # 方便前端联调 feed 卡片样式、分页、距离排序等 UI。生产必须 false。
+    MT_CPS_MOCK_FEED: bool = True
 
     @property
     def mt_cps_configured(self) -> bool:

app/models/__init__.py

@@ -32,6 +32,7 @@ from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
 from app.models.launch_confirm_sample import LaunchConfirmSample  # noqa: F401
 from app.models.meituan_coupon import MeituanCoupon  # noqa: F401
 from app.models.onboarding import OnboardingCompletion  # noqa: F401
+from app.models.phone_rebind_log import PhoneRebindLog  # noqa: F401
 from app.models.ops_marquee_seed import OpsMarqueeSeed  # noqa: F401
 from app.models.ops_stat_config import OpsStatConfig  # noqa: F401
 from app.models.price_observation import PriceObservation  # noqa: F401

app/models/phone_rebind_log.py

@@ -0,0 +1,31 @@
+"""手机号换绑台账。
+
+记录"手机号从老账号被夺走、重建为新账号(X 注销 → Y)"这一破坏性事件,支撑"一个手机号
+30 天内最多换绑一次"的限制。手机号级、渠道无关(source 标来源);普通微信绑定不写此表。
+见 M2 spec §4.1。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class PhoneRebindLog(Base):
+    __tablename__ = "phone_rebind_log"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    # 被换绑的真实手机号(注意:存真实号,不是老账号被腾号后的 deleted_<id>)
+    phone: Mapped[str] = mapped_column(String(20), index=True, nullable=False)
+    # 被注销的老账号 X;P 换绑时已被腾空(极边界)则为空
+    old_user_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 换绑后新建的账号 Y
+    new_user_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    # 换绑来源。手机号级配额、渠道无关,留字段给未来其他换绑路径共用同一份 30 天限制。
+    source: Mapped[str] = mapped_column(String(32), nullable=False, default="wechat_conflict")
+    rebound_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )

tests/test_wechat_conflict.py

@@ -0,0 +1,49 @@
+"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。
+
+沿用 tests/test_wechat_login.py 风格:HTTP 走 client;微信 code→openid 用 monkeypatch;
+短信走 SMS_MOCK(任意 6 位过)。数据变更用"再走一遍 wechat-login 看 openid 落在哪个账号"做行为断言。
+"""
+from __future__ import annotations
+
+import pytest
+
+from app.api.v1 import auth  # noqa: F401  (后续测试打桩 verify_and_get_phone 用)
+from app.core import security
+from app.integrations import wxpay
+from app.models.phone_rebind_log import PhoneRebindLog
+
+
+def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
+    def _f(code: str) -> dict:
+        return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
+    return _f
+
+
+def _sms_occupy(client, phone: str) -> int:
+    """用普通短信登录占用一个手机号(register_channel=sms),返回该账号 id。"""
+    assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
+    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
+    assert r.status_code == 200, r.text
+    return r.json()["user"]["id"]
+
+
+def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id: str) -> dict:
+    """微信登录(新 openid)→ 绑同一手机号 → 返回 phone_occupied 的响应体(含 conflict_ticket)。"""
+    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo(openid))
+    ticket = client.post(
+        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": device_id}
+    ).json()["bind_ticket"]
+    r = client.post(
+        "/api/v1/auth/wechat/bind-phone/sms",
+        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": device_id},
+    )
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["status"] == "phone_occupied"
+    return body
+
+
+# ===== Task 1: 模型可导入(建表由 conftest 的 create_all 完成) =====
+
+def test_phone_rebind_log_model_importable() -> None:
+    assert PhoneRebindLog.__tablename__ == "phone_rebind_log"