From 0124131b9120b6b1341e65d21971f6f8c0522bc5 Mon Sep 17 00:00:00 2001 From: exinglang Date: Fri, 31 Jul 2026 10:49:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E6=8E=A8=E9=80=81):=20=E9=9A=94=E7=A6=BB?= =?UTF-8?q?=E8=AE=BE=E5=A4=87=E8=B4=A6=E5=8F=B7=E7=BB=91=E5=AE=9A=E5=B9=B6?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E5=AE=9A=E5=90=91=E8=81=94=E8=B0=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 11 +- alembic/versions/push_binding_isolation.py | 89 ++++++++++++++ app/api/v1/auth.py | 2 +- app/api/v1/device.py | 33 ++++++ app/core/config.py | 21 +++- app/core/test_account.py | 31 +++-- app/integrations/vendor_push.py | 7 +- app/models/device.py | 10 ++ app/repositories/device.py | 131 ++++++++++++++++++--- app/schemas/auth.py | 18 ++- app/schemas/device.py | 11 ++ app/services/notification_events.py | 70 +++++++++-- docs/api/push-vendor-test.md | 1 + scripts/fire_push_events.py | 76 ++++++++++-- scripts/test_push_invite_order_reward.py | 39 +++++- 15 files changed, 487 insertions(+), 63 deletions(-) create mode 100644 alembic/versions/push_binding_isolation.py diff --git a/.env.example b/.env.example index b77fd01..56c0181 100644 --- a/.env.example +++ b/.env.example @@ -69,8 +69,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 +178,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= diff --git a/alembic/versions/push_binding_isolation.py b/alembic/versions/push_binding_isolation.py new file mode 100644 index 0000000..12e9b2b --- /dev/null +++ b/alembic/versions/push_binding_isolation.py @@ -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") diff --git a/app/api/v1/auth.py b/app/api/v1/auth.py index a8c3141..454c5fc 100644 --- a/app/api/v1/auth.py +++ b/app/api/v1/auth.py @@ -256,7 +256,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( diff --git a/app/api/v1/device.py b/app/api/v1/device.py index 893c6c5..5deb727 100644 --- a/app/api/v1/device.py +++ b/app/api/v1/device.py @@ -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() diff --git a/app/core/config.py b/app/core/config.py index bc41b29..8ce3de8 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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" @@ -191,14 +193,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 +506,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——它们沿用默认值 diff --git a/app/core/test_account.py b/app/core/test_account.py index ca64e06..ee22eb3 100644 --- a/app/core/test_account.py +++ b/app/core/test_account.py @@ -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 = {} diff --git a/app/integrations/vendor_push.py b/app/integrations/vendor_push.py index 3675418..cf90316 100644 --- a/app/integrations/vendor_push.py +++ b/app/integrations/vendor_push.py @@ -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, diff --git a/app/models/device.py b/app/models/device.py index 324e176..9b7b7ae 100644 --- a/app/models/device.py +++ b/app/models/device.py @@ -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) diff --git a/app/repositories/device.py b/app/repositories/device.py index e151738..ab8ad21 100644 --- a/app/repositories/device.py +++ b/app/repositories/device.py @@ -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 diff --git a/app/schemas/auth.py b/app/schemas/auth.py index b5c185c..c232b3e 100644 --- a/app/schemas/auth.py +++ b/app/schemas/auth.py @@ -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 ===== diff --git a/app/schemas/device.py b/app/schemas/device.py index 3f8cda1..c42a04a 100644 --- a/app/schemas/device.py +++ b/app/schemas/device.py @@ -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(无告警)。""" diff --git a/app/services/notification_events.py b/app/services/notification_events.py index 0b6e917..3ddf126 100644 --- a/app/services/notification_events.py +++ b/app/services/notification_events.py @@ -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, ) diff --git a/docs/api/push-vendor-test.md b/docs/api/push-vendor-test.md index db05ebc..8c7d01f 100644 --- a/docs/api/push-vendor-test.md +++ b/docs/api/push-vendor-test.md @@ -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 才有效。 diff --git a/scripts/fire_push_events.py b/scripts/fire_push_events.py index 7948491..69924de 100644 --- a/scripts/fire_push_events.py +++ b/scripts/fire_push_events.py @@ -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=失败)。") diff --git a/scripts/test_push_invite_order_reward.py b/scripts/test_push_invite_order_reward.py index f88cc3c..fb9e8ed 100644 --- a/scripts/test_push_invite_order_reward.py +++ b/scripts/test_push_invite_order_reward.py @@ -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 -- 2.52.0