From 8adad30ff2257ad1ec477cfbdcb21e08db16ba8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E4=B8=96=E7=9D=BF?= <2839904623@qq.com> Date: Mon, 15 Jun 2026 22:16:39 +0800 Subject: [PATCH 1/7] =?UTF-8?q?feat(device):=20=E6=97=A0=E9=9A=9C=E7=A2=8D?= =?UTF-8?q?=E4=BF=9D=E6=8A=A4=E5=AD=98=E6=B4=BB=E5=BF=83=E8=B7=B3=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=20+=20=E6=8E=89=E7=BA=BF=E7=BB=88=E7=AB=AF=E5=91=8A?= =?UTF-8?q?=E8=AD=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 device 表 + /api/v1/device/{register,heartbeat} + 迁移 device_table - heartbeat_monitor_worker 周期扫描心跳超时(App 被杀/无障碍停)→ 服务器终端打印告警 (推送本期未接,先用 logger 终端打印代替真实通知;integrations/jpush.py 已备,后续直接替换) - config / .env.example 增 JPUSH_* / HEARTBEAT_* - 见 spec(仓库外 e:\codes\spec\accessibility-liveness-push.md) 注:本提交同时快照了并行 session 未提交的 coupon_state 改动 + 相关 migration (与本功能同迁移链耦合,无法单独拆分,经确认一并提交)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .env.example | 14 +- ...9_merge_coupon_engage_app_pkg_and_coin_.py | 26 +++ .../versions/coupon_engagement_app_package.py | 47 ++++++ alembic/versions/device_table.py | 53 +++++++ ...2_merge_store_mapping_cols_branch_into_.py | 26 +++ app/api/v1/coupon.py | 56 +++++-- app/api/v1/device.py | 66 ++++++++ app/core/config.py | 26 +++ app/core/heartbeat_monitor_worker.py | 148 ++++++++++++++++++ app/integrations/jpush.py | 94 +++++++++++ app/main.py | 8 + app/models/__init__.py | 1 + app/models/coupon_state.py | 23 ++- app/models/device.py | 83 ++++++++++ app/repositories/coupon_state.py | 62 ++++++-- app/repositories/device.py | 106 +++++++++++++ app/schemas/coupon_state.py | 22 ++- app/schemas/device.py | 36 +++++ 18 files changed, 863 insertions(+), 34 deletions(-) create mode 100644 alembic/versions/3a9941e76909_merge_coupon_engage_app_pkg_and_coin_.py create mode 100644 alembic/versions/coupon_engagement_app_package.py create mode 100644 alembic/versions/device_table.py create mode 100644 alembic/versions/f3d0a16bb4c2_merge_store_mapping_cols_branch_into_.py create mode 100644 app/api/v1/device.py create mode 100644 app/core/heartbeat_monitor_worker.py create mode 100644 app/integrations/jpush.py create mode 100644 app/models/device.py create mode 100644 app/repositories/device.py create mode 100644 app/schemas/device.py diff --git a/.env.example b/.env.example index 7178490..ff8642f 100644 --- a/.env.example +++ b/.env.example @@ -11,7 +11,7 @@ DATABASE_URL=sqlite:///./data/app.db # ===== JWT ===== # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))" -JWT_SECRET_KEY= +JWT_SECRET_KEY=change-me-in-prod-please-use-a-long-random-string JWT_ALGORITHM=HS256 # access token 有效期(分钟),默认 2 小时 JWT_ACCESS_TOKEN_EXPIRE_MINUTES=120 @@ -27,6 +27,18 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify JG_REQUEST_TIMEOUT_SEC=15 +# ===== 极光推送 JPush(无障碍保护掉线告警)===== +# 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。若推送与一键登录/短信是同一个极光应用, +# JPUSH_* 留空即自动回退到上面的 JG_APP_KEY/JG_MASTER_SECRET;否则单独填那个 push 应用的密钥。 +# 运维清单(厂商通道等)见 spec/accessibility-liveness-push.md §7。 +JPUSH_APP_KEY= +JPUSH_MASTER_SECRET= +JPUSH_PUSH_ENDPOINT=https://api.jpush.cn/v3/push +# 无障碍保护存活监控后台任务 +HEARTBEAT_MONITOR_ENABLED=true +HEARTBEAT_TIMEOUT_MINUTES=10 +HEARTBEAT_SCAN_INTERVAL_SEC=60 + # ===== 短信 (mock 模式) ===== # mock = true 时,任意 6 位数字均通过,且 /sms/send 不真发短信(只 log)。 # 后续接阿里云/腾讯云短信时,改成 false 并填供应商相关 key。 diff --git a/alembic/versions/3a9941e76909_merge_coupon_engage_app_pkg_and_coin_.py b/alembic/versions/3a9941e76909_merge_coupon_engage_app_pkg_and_coin_.py new file mode 100644 index 0000000..9fe1535 --- /dev/null +++ b/alembic/versions/3a9941e76909_merge_coupon_engage_app_pkg_and_coin_.py @@ -0,0 +1,26 @@ +"""merge coupon_engage_app_pkg and coin_txn_task_ref_uq heads + +Revision ID: 3a9941e76909 +Revises: coin_txn_task_ref_uq, coupon_engage_app_pkg +Create Date: 2026-06-13 10:09:37.557466 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '3a9941e76909' +down_revision: Union[str, Sequence[str], None] = ('coin_txn_task_ref_uq', 'coupon_engage_app_pkg') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/alembic/versions/coupon_engagement_app_package.py b/alembic/versions/coupon_engagement_app_package.py new file mode 100644 index 0000000..ac3b518 --- /dev/null +++ b/alembic/versions/coupon_engagement_app_package.py @@ -0,0 +1,47 @@ +"""coupon_prompt_engagement 加 app_package(弹窗频控改按 App 为单位) + +2026-06-12 产品确认:领券引导窗「一个 app 一天只能弹一次」,没领完进其他平台要再弹, +彻底领完(coupon_daily_completion)才全局不弹。频控行从 (device, 日) 唯一改为 +(device, app_package, 日) 唯一;旧数据 app_package 回填 NULL(= 全局兜底行, +按 App 查询时忽略,不带包名的旧式查询仍生效)。 + +⚠️ downgrade 有损:同设备同日多 App 各一行时,重建 (device, 日) 唯一约束会撞; +仅开发/测试库可降级(频控行本就是当日临时数据,coupon_claim_record 资产不受影响)。 + +Revision ID: coupon_engage_app_pkg +Revises: 9b894f5fff05 +Create Date: 2026-06-12 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "coupon_engage_app_pkg" +down_revision: str | Sequence[str] | None = "9b894f5fff05" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # batch 模式兼容 SQLite(本地)与 PG(线上):SQLite 改约束要重建表,batch 自动处理。 + with op.batch_alter_table("coupon_prompt_engagement", schema=None) as batch_op: + batch_op.add_column(sa.Column("app_package", sa.String(length=64), nullable=True)) + batch_op.drop_constraint("uq_coupon_engage_device_date", type_="unique") + batch_op.create_unique_constraint( + "uq_coupon_engage_device_app_date", + ["device_id", "app_package", "engage_date"], + ) + + +def downgrade() -> None: + with op.batch_alter_table("coupon_prompt_engagement", schema=None) as batch_op: + batch_op.drop_constraint("uq_coupon_engage_device_app_date", type_="unique") + batch_op.create_unique_constraint( + "uq_coupon_engage_device_date", ["device_id", "engage_date"] + ) + batch_op.drop_column("app_package") diff --git a/alembic/versions/device_table.py b/alembic/versions/device_table.py new file mode 100644 index 0000000..b38fa17 --- /dev/null +++ b/alembic/versions/device_table.py @@ -0,0 +1,53 @@ +"""device table (无障碍保护存活检测 + 极光推送) + +Revision ID: device_liveness_table +Revises: f3d0a16bb4c2 +Create Date: 2026-06-15 12:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'device_liveness_table' +down_revision: Union[str, Sequence[str], None] = 'f3d0a16bb4c2' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'device', + sa.Column('id', sa.Integer(), autoincrement=True, nullable=False), + sa.Column('user_id', sa.Integer(), nullable=False), + sa.Column('device_id', sa.String(length=128), nullable=False), + sa.Column('registration_id', sa.String(length=64), nullable=True), + sa.Column('platform', sa.String(length=16), nullable=False), + sa.Column('app_version', sa.String(length=32), nullable=True), + sa.Column('ever_protected', sa.Boolean(), nullable=False), + sa.Column('last_heartbeat_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('last_report_protection_on', sa.Boolean(), nullable=False), + sa.Column('liveness_state', sa.String(length=16), nullable=False), + sa.Column('notified_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('user_id', 'device_id', name='uq_device_user_device'), + ) + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_device_user_id'), ['user_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_device_id'), ['device_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_last_heartbeat_at'), ['last_heartbeat_at'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_device_last_heartbeat_at')) + batch_op.drop_index(batch_op.f('ix_device_device_id')) + batch_op.drop_index(batch_op.f('ix_device_user_id')) + + op.drop_table('device') diff --git a/alembic/versions/f3d0a16bb4c2_merge_store_mapping_cols_branch_into_.py b/alembic/versions/f3d0a16bb4c2_merge_store_mapping_cols_branch_into_.py new file mode 100644 index 0000000..4b3217e --- /dev/null +++ b/alembic/versions/f3d0a16bb4c2_merge_store_mapping_cols_branch_into_.py @@ -0,0 +1,26 @@ +"""merge store_mapping cols branch into coupon/coin mergepoint + +Revision ID: f3d0a16bb4c2 +Revises: 3a9941e76909, store_mapping_jd_cols +Create Date: 2026-06-14 14:31:25.504510 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'f3d0a16bb4c2' +down_revision: Union[str, Sequence[str], None] = ('3a9941e76909', 'store_mapping_jd_cols') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass diff --git a/app/api/v1/coupon.py b/app/api/v1/coupon.py index e44adb5..85f53b7 100644 --- a/app/api/v1/coupon.py +++ b/app/api/v1/coupon.py @@ -27,6 +27,7 @@ from app.schemas.coupon_state import ( CouponCompletedTodayOut, CouponPromptDismissIn, CouponPromptShouldShowOut, + CouponPromptShownIn, ) logger = logging.getLogger("shagua.coupon") @@ -112,8 +113,10 @@ async def coupon_step( user_id = _to_int(meta.get("user_id")) # 登录态才带;判断不靠它,资产留痕用 trace_id = meta.get("trace_id") - # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started), - # 今天这台设备不再弹引导窗(对齐前台"点一键领取即 markEngaged")。写库失败绝不能 + # 领券任务首帧(step=0)= 用户已发起领券 → 记一条今日 engagement(claim_started)。 + # 透传链路拿不到发起 App 的包名 → 写 app_package=NULL 的全局兜底行:只对旧式不带 + # package 的 should-show 查询生效;按 App 频控(2026-06-12)由客户端 /prompt/shown + # 负责,本行不堵其他 App 的弹窗(领一半终止,其他平台还要弹)。写库失败绝不能 # 连累领券主流程,整段吞掉。 if device_id and meta.get("step") == 0: try: @@ -188,37 +191,66 @@ async def coupon_step( return resp_json +@router.post("/prompt/shown", summary="领券引导窗已对某 App 弹出(按 App 频控主判据)") +def coupon_prompt_shown(payload: CouponPromptShownIn, db: DbSession) -> dict[str, bool]: + """客户端弹窗一亮即调用 → 记一条今日该 App 的 engagement(shown)。 + + 2026-06-12 频控按 App 为单位(方案文档「一个 app 一天只能弹一次,时机=每天第一次 + 进入」):以"弹出"为频控锚点,用户领/拒/无视都只算这一次;dismiss/claim_started + 后续只升级同一行的 engage_type。MVP 不鉴权,按 device_id 记。 + """ + coupon_repo.mark_engagement( + db, payload.device_id, payload.user_id, "shown", payload.package + ) + return {"ok": True} + + @router.post("/prompt/dismiss", summary="用户拒绝/关闭领券引导窗(记今日已 engage)") def coupon_prompt_dismiss(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]: - """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天不再弹。 + """客户端点关闭引导窗时调用 → 记一条今日 engagement(dismissed),今天该 App 不再弹。 server 在透传链路里看不到"用户拒绝"(拒绝不发起领券),故必须客户端通知。 - MVP 不鉴权,按 device_id 记。 + 频控以 /prompt/shown 为主判据,本端点把同一行升级成 dismissed(记录用); + 旧客户端不带 package → 写 NULL 全局兜底行(旧语义)。MVP 不鉴权,按 device_id 记。 """ - coupon_repo.mark_engagement(db, payload.device_id, payload.user_id, "dismissed") + coupon_repo.mark_engagement( + db, payload.device_id, payload.user_id, "dismissed", payload.package + ) return {"ok": True} @router.get( "/prompt/should-show", response_model=CouponPromptShouldShowOut, - summary="切到外卖 App 时是否还应弹领券引导窗", + summary="切到某外卖 App 时是否还应弹领券引导窗(按 App 频控)", ) def coupon_prompt_should_show( - device_id: str, db: DbSession + device_id: str, db: DbSession, package: str | None = None ) -> CouponPromptShouldShowOut: - """今天这台设备已 engage(领或拒)过 → should_show=false。客户端据此决定弹不弹 - (纯后台判据,客户端不再做前台 SP 缓存判断)。""" + """should_show = 今天没跑完整轮领券 AND 该 App 今天没弹过。 + + 2026-06-12 按 App 为单位(产品确认): + - 该 App 今天弹过(shown/dismissed/claim_started 任一)→ false; + - **其他 App** 弹过/领了一半不影响本 App → 仍 true(没领完进其他平台要再弹); + - 今天已跑完整轮(coupon_daily_completion,"彻底领完")→ 全局 false。 + package 不传 = 旧客户端,退回旧全局语义(任意一行算 engage)+ 同样吃 completion 闸。 + 客户端据此决定弹不弹(纯后台判据,客户端不做前台 SP 缓存判断)。""" + if coupon_repo.has_completed_today(db, device_id): + return CouponPromptShouldShowOut(should_show=False) return CouponPromptShouldShowOut( - should_show=not coupon_repo.has_engaged_today(db, device_id) + should_show=not coupon_repo.has_engaged_today(db, device_id, package) ) -@router.post("/prompt/reset", summary="重置今日领券引导窗 engagement(开发测频控用)") +@router.post("/prompt/reset", summary="重置今日领券状态:弹窗 engagement + 完成记录(开发测试用)") def coupon_prompt_reset(payload: CouponPromptDismissIn, db: DbSession) -> dict[str, bool]: - """删这台设备今天的 engagement → has_engaged_today 变 false,今天又能弹。 + """删这台设备今天的 engagement + coupon_daily_completion:弹窗又能弹,且首页 + 「去领取」卡/弹窗「一键自动领取」CTA 恢复可点(只删 engagement 的话, 重置后 CTA + 仍被"今日已跑完整轮"置灰, 2026-06-12)。配合客户端本地重置 = 等效重装。 + 领券记录(coupon_claim_record)是资产沉淀、不参与任何门控判断, 不删。 开发设置「重置今日领券弹窗状态」按钮调。MVP 不鉴权,按 device_id。""" coupon_repo.reset_today_engagement(db, payload.device_id) + coupon_repo.reset_today_completion(db, payload.device_id) return {"ok": True} diff --git a/app/api/v1/device.py b/app/api/v1/device.py new file mode 100644 index 0000000..d56839d --- /dev/null +++ b/app/api/v1/device.py @@ -0,0 +1,66 @@ +"""设备注册 / 心跳 endpoint(无障碍保护存活检测)。 + +路由前缀 /api/v1/device,需 Bearer 鉴权(设备绑登录用户)。 + POST /register 注册设备 / 更新 registration_id(App 前台、拿到 push token 时调) + POST /heartbeat 上报心跳(无障碍服务存活时周期调,刷新存活) + +后端 heartbeat_monitor_worker 据此发现心跳超时的设备并极光推送告警。 +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter + +from app.api.deps import CurrentUser, DbSession +from app.repositories import device as device_repo +from app.schemas.device import ( + DeviceOut, + DeviceRegisterRequest, + HeartbeatRequest, + OkResponse, +) + +logger = logging.getLogger("shagua.device") + +router = APIRouter(prefix="/api/v1/device", tags=["device"]) + + +@router.post("/register", response_model=DeviceOut, summary="注册设备/更新推送token") +def register_device( + req: DeviceRegisterRequest, + user: CurrentUser, + db: DbSession, +) -> DeviceOut: + device = device_repo.register_or_update( + db, + user_id=user.id, + device_id=req.device_id, + registration_id=req.registration_id, + platform=req.platform, + app_version=req.app_version, + ) + logger.info( + "device register user_id=%d device_id=%s reg=%s", + user.id, + req.device_id, + bool(req.registration_id), + ) + return DeviceOut.model_validate(device) + + +@router.post("/heartbeat", response_model=OkResponse, summary="上报心跳") +def report_heartbeat( + req: HeartbeatRequest, + user: CurrentUser, + db: DbSession, +) -> OkResponse: + device_repo.touch_heartbeat( + db, + user_id=user.id, + device_id=req.device_id, + accessibility_enabled=req.accessibility_enabled, + registration_id=req.registration_id, + ) + return OkResponse() diff --git a/app/core/config.py b/app/core/config.py index a64c3a8..901dec2 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -61,6 +61,32 @@ class Settings(BaseSettings): JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify" JG_REQUEST_TIMEOUT_SEC: int = 15 + # ===== 极光推送 JPush(无障碍保护掉线告警)===== + # 客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d(android build.gradle manifestPlaceholder)。 + # 若推送与一键登录/短信是同一个极光应用(大概率),JPUSH_* 留空即自动回退到 JG_*。 + # 否则在 .env 单独配 JPUSH_APP_KEY / JPUSH_MASTER_SECRET(对应那个 push appkey)。 + JPUSH_APP_KEY: str = "" + JPUSH_MASTER_SECRET: str = "" + JPUSH_PUSH_ENDPOINT: str = "https://api.jpush.cn/v3/push" + + # 无障碍保护存活监控后台任务 + HEARTBEAT_MONITOR_ENABLED: bool = True # 总开关 + HEARTBEAT_TIMEOUT_MINUTES: int = 10 # 多久没心跳算掉线(≈3 个客户端心跳周期) + HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期 + + @property + def jpush_app_key(self) -> str: + return self.JPUSH_APP_KEY or self.JG_APP_KEY + + @property + def jpush_master_secret(self) -> str: + return self.JPUSH_MASTER_SECRET or self.JG_MASTER_SECRET + + @property + def jpush_configured(self) -> bool: + """推送凭证齐全(缺则 monitor 只扫不发,不报错)。""" + return bool(self.jpush_app_key and self.jpush_master_secret) + # ===== 短信 ===== SMS_MOCK: bool = True SMS_CODE_TTL_SEC: int = 300 diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py new file mode 100644 index 0000000..607fe8d --- /dev/null +++ b/app/core/heartbeat_monitor_worker.py @@ -0,0 +1,148 @@ +"""无障碍保护存活监控后台任务。 + +周期扫描「曾经保护过、当前 alive、心跳超时」的设备 = App 被彻底杀掉/无障碍已停(心跳断了), +**命中即在服务器终端打印告警**(本期先不接推送,工程量大,用终端打印代替真实通知);并把状态机 +推进到 notified 防每轮重复打印(心跳恢复时由 repositories.device.touch_heartbeat 重置回 alive)。 +结构仿 withdraw_reconcile_worker(单实例锁 + asyncio 轮询 + 优雅退出)。 + +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +import time +from collections.abc import Iterator +from datetime import datetime, timezone +from pathlib import Path + +from sqlalchemy.exc import SQLAlchemyError + +from app.core.config import settings +from app.db.session import SessionLocal +from app.repositories import device as device_repo + +logger = logging.getLogger("shagua.heartbeat_monitor") +_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "heartbeat_monitor.lock" + + +def _touch_lock() -> None: + with contextlib.suppress(FileNotFoundError): + os.utime(_LOCK_PATH, None) + + +@contextlib.contextmanager +def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]: + """同机多进程保护:同一时间只允许一个监控 worker 运行。""" + _LOCK_PATH.parent.mkdir(parents=True, exist_ok=True) + fd: int | None = None + try: + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + try: + age = time.time() - _LOCK_PATH.stat().st_mtime + except FileNotFoundError: + age = stale_after_sec + 1 + if age > stale_after_sec: + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + try: + fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + except FileExistsError: + fd = None + + if fd is None: + yield False + return + + os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii")) + yield True + finally: + if fd is not None: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + _LOCK_PATH.unlink() + + +def _silent_seconds(last: datetime | None) -> int | None: + """距上次心跳的秒数(兼容 sqlite 取回的 naive datetime)。""" + if last is None: + return None + ref = datetime.now(timezone.utc) if last.tzinfo is not None else datetime.utcnow() + return int((ref - last).total_seconds()) + + +def _scan_once(timeout_minutes: int) -> dict: + """扫描一轮:找出心跳超时(App 被彻底杀掉/无障碍已停)的设备,在**服务器终端打印**告警代替真实推送。 + + 本期不接推送(极光/厂商通道工程量大),只做服务端掉线检测:命中即 logger.warning 打印到终端, + 并把状态机推进到 notified 防每轮重复打印(心跳恢复时 touch_heartbeat 会重置回 alive)。 + """ + notified = 0 + with SessionLocal() as db: + overdue = device_repo.list_overdue(db, timeout_minutes=timeout_minutes) + for device in overdue: + silent = _silent_seconds(device.last_heartbeat_at) + logger.warning( + "🔴 [掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)" + " → 判定 App 已被杀/无障碍已停。【本应推送通知提醒用户重开;推送暂未接,先终端打印代替】", + device.user_id, + device.device_id, + silent if silent is not None else "?", + timeout_minutes, + ) + device_repo.mark_notified(db, device_id_pk=device.id) + notified += 1 + return {"checked": len(overdue), "notified": notified} + + +async def _run_loop() -> None: + interval = max(10, int(settings.HEARTBEAT_SCAN_INTERVAL_SEC)) + timeout_minutes = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES)) + lock_stale_after = max(interval * 3, 600) + with _single_instance_lock(lock_stale_after) as lock_acquired: + if not lock_acquired: + logger.warning("heartbeat monitor skipped: another worker owns lock") + return + await _run_locked_loop(interval, timeout_minutes) + + +async def _run_locked_loop(interval: int, timeout_minutes: int) -> None: + logger.info( + "heartbeat monitor started interval=%ss timeout=%sm", + interval, + timeout_minutes, + ) + try: + while True: + try: + _touch_lock() + result = await asyncio.to_thread(_scan_once, timeout_minutes) + if result["notified"]: + logger.info("heartbeat monitor result=%s", result) + except SQLAlchemyError: + logger.exception("heartbeat monitor db error") + except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出 + logger.exception("heartbeat monitor unexpected error") + await asyncio.sleep(interval) + except asyncio.CancelledError: + logger.info("heartbeat monitor stopped") + raise + + +def start_heartbeat_monitor() -> asyncio.Task | None: + if not settings.HEARTBEAT_MONITOR_ENABLED: + logger.info("heartbeat monitor disabled") + return None + return asyncio.create_task(_run_loop(), name="heartbeat-monitor") + + +async def stop_heartbeat_monitor(task: asyncio.Task | None) -> None: + if task is None: + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task diff --git a/app/integrations/jpush.py b/app/integrations/jpush.py new file mode 100644 index 0000000..5266ac9 --- /dev/null +++ b/app/integrations/jpush.py @@ -0,0 +1,94 @@ +"""极光推送 JPush(无障碍保护掉线告警)。 + +调极光 Push REST v3 /v3/push 给指定 registration_id 推一条通知。鉴权复用极光 Basic Auth +(appKey:masterSecret),与一键登录/短信同模式(见 integrations/jiguang.py、sms.py)。 + +凭证:settings.jpush_app_key / jpush_master_secret(JPUSH_* 留空时自动回退到 JG_*, +若推送与一键登录是同一极光应用)。客户端 push appkey 已知 = 966b451a8d9cfe12d173ea9d。 +厂商通道(App 被杀也能到达)由极光后台 + 客户端插件负责,本服务只管调 push 接口。 +""" +from __future__ import annotations + +import base64 +import logging + +import httpx + +from app.core.config import settings + +logger = logging.getLogger("shagua.jpush") + + +class JPushError(Exception): + """极光推送调用失败。""" + + +class JPushNotConfiguredError(JPushError): + """缺 appKey / masterSecret。""" + + +def push_to_registration_ids( + registration_ids: list[str], + *, + title: str, + alert: str, + extras: dict | None = None, +) -> dict: + """给一批 registration_id 推送通知 + 透传消息。失败抛 JPushError。 + + Returns: 极光响应 JSON(含 sendno / msg_id)。 + """ + if not settings.jpush_configured: + raise JPushNotConfiguredError( + "JPush 未配置(缺 JPUSH_APP_KEY/JPUSH_MASTER_SECRET,且 JG_* 也为空)" + ) + reg_ids = [r for r in registration_ids if r] + if not reg_ids: + raise JPushError("registration_ids 为空") + + auth_b64 = base64.b64encode( + f"{settings.jpush_app_key}:{settings.jpush_master_secret}".encode() + ).decode() + payload = { + "platform": ["android"], + "audience": {"registration_id": reg_ids}, + "notification": { + "android": { + "alert": alert, + "title": title, + "priority": 1, + "extras": extras or {}, + }, + }, + "message": { + "msg_content": alert, + "title": title, + "content_type": "text", + "extras": extras or {}, + }, + "options": {"time_to_live": 86400, "apns_production": True}, + } + + try: + resp = httpx.post( + settings.JPUSH_PUSH_ENDPOINT, + json=payload, + headers={ + "Content-Type": "application/json", + "Authorization": f"Basic {auth_b64}", + }, + timeout=settings.JG_REQUEST_TIMEOUT_SEC, + ) + except httpx.HTTPError as e: + raise JPushError(f"jpush 网络错误: {e}") from e + + if resp.status_code != 200: + body = resp.text[:300] + logger.error("[JPush] http=%s body=%s", resp.status_code, body) + raise JPushError(f"jpush http {resp.status_code}") + + data = resp.json() + if not data.get("sendno") and not data.get("msg_id"): + logger.error("[JPush] unexpected response: %s", data) + raise JPushError(f"jpush 响应异常: {data}") + return data diff --git a/app/main.py b/app/main.py index cea4568..d8cac0a 100644 --- a/app/main.py +++ b/app/main.py @@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router from app.api.v1.compare_milestone import router as compare_milestone_router from app.api.v1.compare_record import router as compare_record_router from app.api.v1.coupon import router as coupon_router +from app.api.v1.device import router as device_router from app.api.internal.price import router as internal_price_router from app.api.internal.store import router as internal_store_router from app.api.v1.feedback import router as feedback_router @@ -35,6 +36,10 @@ from app.api.v1.user import router as user_router from app.api.v1.wallet import router as wallet_router from app.api.v1.wxpay import router as wxpay_router from app.core.config import settings +from app.core.heartbeat_monitor_worker import ( + start_heartbeat_monitor, + stop_heartbeat_monitor, +) from app.core.logging import setup_logging from app.core.withdraw_reconcile_worker import ( start_withdraw_reconcile_worker, @@ -56,9 +61,11 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: settings.DATABASE_URL.split("://", 1)[0], ) reconcile_task = start_withdraw_reconcile_worker() + heartbeat_task = start_heartbeat_monitor() try: yield finally: + await stop_heartbeat_monitor(heartbeat_task) await stop_withdraw_reconcile_worker(reconcile_task) logger.info("shutting down") @@ -91,6 +98,7 @@ app.include_router(user_router) app.include_router(feedback_router) app.include_router(invite_router) app.include_router(coupon_router) +app.include_router(device_router) app.include_router(compare_router) app.include_router(compare_record_router) app.include_router(compare_milestone_router) diff --git a/app/models/__init__.py b/app/models/__init__.py index a089a8a..6dfd8b1 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -7,6 +7,7 @@ from app.models.admin import AdminAuditLog, AdminUser # noqa: F401 from app.models.app_config import AppConfig # noqa: F401 from app.models.comparison import ComparisonRecord # noqa: F401 from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401 +from app.models.device import Device # noqa: F401 from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, CouponDailyCompletion, diff --git a/app/models/coupon_state.py b/app/models/coupon_state.py index bd2aa2d..1ec6424 100644 --- a/app/models/coupon_state.py +++ b/app/models/coupon_state.py @@ -137,14 +137,23 @@ class CouponDailyCompletion(Base): class CouponPromptEngagement(Base): - """按 (device, 自然日) 记"今天是否对领券引导窗表达过意向"——弹窗频控源。""" + """按 (device, App, 自然日) 记"今天该 App 是否弹过/engage 过领券引导窗"——弹窗频控源。 + + 2026-06-12 改为**按 App 为单位**(对齐方案文档「一个 app 一天只能弹出弹窗一次」+ 产品确认: + 没领完就进其他平台要再弹,彻底领完才全局不弹——"彻底领完"由 coupon_daily_completion 负责): + - app_package 非空:该 App 当天的频控行(shown/dismissed 由客户端带包名上报)。 + - app_package 为 NULL:全局兜底行(step=0 透传链路自动记的 claim_started 拿不到包名 + + 旧版客户端不带包名)。**按 App 查询时忽略 NULL 行**(领券中途终止 ≠ 其他 App engage 过); + 不带包名的旧式查询仍把 NULL 行算作"今天 engage 过"(旧客户端行为不变)。 + """ __tablename__ = "coupon_prompt_engagement" __table_args__ = ( - # 一台设备一天一条:今天 engage 过(领或拒)就不再弹。 + # 一台设备、一个 App、一天一条(app_package=NULL 的全局兜底行不受唯一约束限制—— + # SQL 标准里 NULL 互不相等;写入走 select-first upsert,正常不会堆重复行)。 UniqueConstraint( - "device_id", "engage_date", - name="uq_coupon_engage_device_date", + "device_id", "app_package", "engage_date", + name="uq_coupon_engage_device_app_date", ), ) @@ -152,10 +161,12 @@ class CouponPromptEngagement(Base): device_id: Mapped[str] = mapped_column(String(64), nullable=False) user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + # 在哪个外卖 App 弹的窗(Android 包名,如 com.sankuai.meituan)。NULL = 全局兜底行(见类注释)。 + app_package: Mapped[str | None] = mapped_column(String(64), nullable=True) # Asia/Shanghai 自然日。 engage_date: Mapped[date] = mapped_column(Date, nullable=False) - # claim_started(点了一键领取)/ dismissed(点了拒绝/关闭)。仅记录区分, - # 判断只看"今天有没有这条",type 不影响弹不弹。 + # shown(弹窗已对该 App 弹出,频控主判据)/ claim_started(点了一键领取)/ + # dismissed(点了拒绝/关闭)。判断只看"今天该 App 有没有这条",type 仅记录区分。 engage_type: Mapped[str] = mapped_column(String(16), nullable=False) created_at: Mapped[datetime] = mapped_column( diff --git a/app/models/device.py b/app/models/device.py new file mode 100644 index 0000000..520b325 --- /dev/null +++ b/app/models/device.py @@ -0,0 +1,83 @@ +"""设备表(无障碍保护存活检测 + 极光推送)。 + +每条 = 一个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)。 +客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报 +registration_id(极光推送目标)。后端 heartbeat_monitor_worker 扫描「曾经保护过、 +现在心跳超时」的设备,通过极光推送提醒用户重开无障碍。 + +liveness_state 状态机(防刷屏,一次掉线只推一条): + unknown → alive(收到 service 心跳)→ silent/notified(扫描发现超时并已推送) + 心跳恢复时 handler 重置回 alive。 +见 spec: spec/accessibility-liveness-push.md。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Integer, + String, + UniqueConstraint, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + + +class Device(Base): + __tablename__ = "device" + __table_args__ = ( + UniqueConstraint("user_id", "device_id", name="uq_device_user_device"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column( + Integer, ForeignKey("user.id"), index=True, nullable=False + ) + # 客户端 DeviceId.get() 生成的 per-install id(如 device_Pixel_ab12cd34) + device_id: Mapped[str] = mapped_column(String(128), index=True, nullable=False) + # 极光推送 registration id;拿到才填(JCollectionAuth 同意后才下发) + registration_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android") + app_version: Mapped[str | None] = mapped_column(String(32), nullable=True) + + # 收到过 service 心跳即 true(=该设备开过无障碍,功能对它有意义) + ever_protected: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + # 最近一次 service 心跳时间(存活证明);超时即视为保护掉线 + last_heartbeat_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), index=True, nullable=True + ) + # 最近一次上报的无障碍开关状态(观测用) + last_report_protection_on: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) + # unknown / alive / silent / notified + liveness_state: Mapped[str] = mapped_column( + String(16), nullable=False, default="unknown" + ) + # 最近一次推送告警时间 + notified_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + def __repr__(self) -> str: # pragma: no cover + return ( + f"" + ) diff --git a/app/repositories/coupon_state.py b/app/repositories/coupon_state.py index c5012b2..a6d9ef7 100644 --- a/app/repositories/coupon_state.py +++ b/app/repositories/coupon_state.py @@ -31,26 +31,46 @@ def today_cn() -> date: # ===== 弹窗频控(coupon_prompt_engagement)===== -def has_engaged_today(db: Session, device_id: str) -> bool: - """这台设备今天是否已对领券引导窗表达过意向(领或拒)。有 = 不再弹。""" - row = db.execute( - select(CouponPromptEngagement.id).where( - CouponPromptEngagement.device_id == device_id, - CouponPromptEngagement.engage_date == today_cn(), - ) - ).first() +def has_engaged_today( + db: Session, device_id: str, app_package: str | None = None +) -> bool: + """这台设备今天是否已弹过/engage 过领券引导窗。有 = 不再弹。 + + 2026-06-12 按 App 为单位: + - app_package 给定(新客户端):只看**该 App** 当天的行;app_package=NULL 的全局兜底行 + (step=0 claim_started / 旧客户端写的)**不算**——领券中途终止不该堵死其他 App 的弹窗, + "彻底领完才全局不弹"由 has_completed_today 在 should-show 端点单独把关。 + - app_package=None(旧客户端不带包名):保持旧全局语义,当天任意一行都算 engage 过。 + """ + cond = [ + CouponPromptEngagement.device_id == device_id, + CouponPromptEngagement.engage_date == today_cn(), + ] + if app_package is not None: + cond.append(CouponPromptEngagement.app_package == app_package) + row = db.execute(select(CouponPromptEngagement.id).where(*cond)).first() return row is not None def mark_engagement( - db: Session, device_id: str, user_id: int | None, engage_type: str + db: Session, + device_id: str, + user_id: int | None, + engage_type: str, + app_package: str | None = None, ) -> None: - """记今日意向(claim_started / dismissed)。(device, 今天) 唯一,幂等 upsert。""" + """记今日意向(shown / claim_started / dismissed)。(device, App, 今天) 唯一,幂等 upsert。 + + app_package=None = 全局兜底行(step=0 透传链路拿不到包名 / 旧客户端),与各 App 行互不覆盖。 + 同 (device, App, 日) 重复上报走更新(shown → dismissed 升级 engage_type)。 + """ today = today_cn() row = db.execute( select(CouponPromptEngagement).where( CouponPromptEngagement.device_id == device_id, CouponPromptEngagement.engage_date == today, + # SQLAlchemy 的 == None 会生成 IS NULL,NULL 兜底行与 App 行各自独立 upsert。 + CouponPromptEngagement.app_package == app_package, ) ).scalar_one_or_none() if row is not None: @@ -59,19 +79,20 @@ def mark_engagement( row.user_id = user_id else: db.add(CouponPromptEngagement( - device_id=device_id, user_id=user_id, + device_id=device_id, user_id=user_id, app_package=app_package, engage_date=today, engage_type=engage_type, )) try: db.commit() except IntegrityError: - # 并发下另一请求刚插了同 (device, 日) → 唯一约束撞,回滚忽略(本就幂等)。 + # 并发下另一请求刚插了同 (device, App, 日) → 唯一约束撞,回滚忽略(本就幂等)。 db.rollback() def reset_today_engagement(db: Session, device_id: str) -> int: """删这台设备今天的 engagement(开发设置「重置今日领券弹窗状态」调,测频控用)。 - 删后 has_engaged_today → false,今天又能弹。返回删除行数。""" + 按 device+日 删,**所有 App 的频控行 + NULL 兜底行一并清**(频控按 App 拆行后语义不变: + 重置 = 等效重装,每个 App 今天都又能弹)。返回删除行数。""" result = db.execute( delete(CouponPromptEngagement).where( CouponPromptEngagement.device_id == device_id, @@ -84,6 +105,21 @@ def reset_today_engagement(db: Session, device_id: str) -> int: # ===== 今日跑完整轮(coupon_daily_completion)===== +def reset_today_completion(db: Session, device_id: str) -> int: + """删这台设备今天的「已跑完整轮」记录(开发设置「重置今日领券弹窗状态」调)。 + 删后 has_completed_today → false:首页「去领取」卡恢复可点、领券弹窗 CTA 不再置灰 + (只清 engagement 不清这条的话, 重置后弹窗能弹但「一键自动领取」仍是灰的, 2026-06-12)。 + 返回删除行数。""" + result = db.execute( + delete(CouponDailyCompletion).where( + CouponDailyCompletion.device_id == device_id, + CouponDailyCompletion.complete_date == today_cn(), + ) + ) + db.commit() + return result.rowcount or 0 + + def has_completed_today(db: Session, device_id: str) -> bool: """这台设备今天是否已跑完整轮领券(到 done 帧)。有 = 首页置灰、不能再领。""" row = db.execute( diff --git a/app/repositories/device.py b/app/repositories/device.py new file mode 100644 index 0000000..3e7c35c --- /dev/null +++ b/app/repositories/device.py @@ -0,0 +1,106 @@ +"""device 表读写(设备注册 / 心跳 / 超时扫描)。""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models.device import Device + + +def _get(db: Session, *, user_id: int, device_id: str) -> Device | None: + stmt = select(Device).where( + Device.user_id == user_id, Device.device_id == device_id + ) + return db.execute(stmt).scalar_one_or_none() + + +def register_or_update( + db: Session, + *, + user_id: int, + device_id: str, + registration_id: str | None, + platform: str = "android", + app_version: str | None = None, +) -> Device: + """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。""" + device = _get(db, user_id=user_id, device_id=device_id) + if device is None: + device = Device( + user_id=user_id, + device_id=device_id, + registration_id=registration_id, + platform=platform or "android", + app_version=app_version, + ) + db.add(device) + else: + if registration_id: + device.registration_id = registration_id + if platform: + device.platform = platform + if app_version: + device.app_version = app_version + db.commit() + db.refresh(device) + return device + + +def touch_heartbeat( + db: Session, + *, + user_id: int, + device_id: str, + accessibility_enabled: bool, + registration_id: str | None, +) -> Device: + """处理一次心跳(心跳也能自注册)。 + + service 心跳或 accessibility_enabled=true 时,刷新存活并把状态机重置回 alive、 + 清掉 notified_at(掉线恢复 → 下次再断才会再推一条)。 + """ + now = datetime.now(timezone.utc) + device = _get(db, user_id=user_id, device_id=device_id) + if device is None: + device = Device(user_id=user_id, device_id=device_id) + db.add(device) + + if registration_id: + device.registration_id = registration_id + device.last_report_protection_on = accessibility_enabled + + if accessibility_enabled: + device.last_heartbeat_at = now + device.ever_protected = True + device.liveness_state = "alive" + device.notified_at = None + + db.commit() + db.refresh(device) + return device + + +def list_overdue(db: Session, *, timeout_minutes: int) -> list[Device]: + """掉线设备:曾经保护过、当前 alive、心跳超时。 + + 本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。 + """ + cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes) + stmt = select(Device).where( + Device.ever_protected.is_(True), + Device.liveness_state == "alive", + Device.last_heartbeat_at.is_not(None), + Device.last_heartbeat_at < cutoff, + ) + return list(db.execute(stmt).scalars().all()) + + +def mark_notified(db: Session, *, device_id_pk: int) -> None: + """标记已推送告警(状态机进入 notified,避免重复推送)。""" + device = db.get(Device, device_id_pk) + if device is not None: + device.liveness_state = "notified" + device.notified_at = datetime.now(timezone.utc) + db.commit() diff --git a/app/schemas/coupon_state.py b/app/schemas/coupon_state.py index fd66cb8..1b429f6 100644 --- a/app/schemas/coupon_state.py +++ b/app/schemas/coupon_state.py @@ -7,16 +7,34 @@ from pydantic import BaseModel class CouponPromptDismissIn(BaseModel): """客户端拒绝/关闭领券引导窗的通知体。 - server 据此记一条今日 engagement(dismissed)→ 今天这台设备不再弹引导窗。 + server 据此记一条今日 engagement(dismissed)。2026-06-12 频控按 App 为单位: + package = 在哪个外卖 App 关的窗(Android 包名);旧客户端不带 → None(全局兜底行)。 MVP 不鉴权,按 device_id 判断;user_id 登录态带上就一并记(资产),可空。 """ device_id: str user_id: int | None = None + package: str | None = None + + +class CouponPromptShownIn(BaseModel): + """领券引导窗已对某 App 弹出(shown)的通知体——按 App 频控的主判据。 + + 客户端弹窗一亮即上报:该 App 今天不再弹(「一个 app 一天只能弹一次, + 时机=每天第一次进入」);领/拒后续上报只升级 engage_type,不影响频控。 + """ + + device_id: str + user_id: int | None = None + package: str | None = None class CouponPromptShouldShowOut(BaseModel): - """切到外卖 App 时是否还应弹领券引导窗。今天已 engage(领或拒)过 → false。""" + """切到外卖 App 时是否还应弹领券引导窗。 + + false 的两种来源:该 App 今天已弹过(per-App engagement)/ 今天已跑完整轮领券 + (completion,彻底领完全局不弹)。 + """ should_show: bool diff --git a/app/schemas/device.py b/app/schemas/device.py new file mode 100644 index 0000000..6343dc8 --- /dev/null +++ b/app/schemas/device.py @@ -0,0 +1,36 @@ +"""设备注册 / 心跳相关 schema(无障碍保护存活检测)。""" +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class DeviceRegisterRequest(BaseModel): + device_id: str + registration_id: str | None = None + platform: str = "android" + app_version: str | None = None + + +class HeartbeatRequest(BaseModel): + device_id: str + source: str = "service" # service | app + accessibility_enabled: bool = True + registration_id: str | None = None + + +class DeviceOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: int + device_id: str + registration_id: str | None + ever_protected: bool + liveness_state: str + last_heartbeat_at: datetime | None + updated_at: datetime | None + + +class OkResponse(BaseModel): + ok: bool = True From 7fd70a264d281752b2c0ef7c7bbcabf77d17036f Mon Sep 17 00:00:00 2001 From: marco Date: Fri, 19 Jun 2026 11:44:11 +0800 Subject: [PATCH 2/7] =?UTF-8?q?feat(cps):=20=E5=BE=AE=E4=BF=A1=E7=BD=91?= =?UTF-8?q?=E9=A1=B5=E6=8E=88=E6=9D=83=E6=8E=A5=E5=85=A5=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=90=BD=E5=9C=B0=E9=A1=B5=E6=8B=BF=20openid/=E6=98=B5?= =?UTF-8?q?=E7=A7=B0=E5=A4=B4=E5=83=8F,=E7=94=A8=E6=88=B7=E7=BA=A7?= =?UTF-8?q?=E7=BE=A4=E7=BB=9F=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 服务号网页授权(WX_MP_APPID/SECRET,区别于 App 的 WECHAT_APP_ID): 落地页 base 静默拿 openid;点「领券」补 userinfo 拿昵称头像(交互触发避快照页) - 新表 cps_wx_user + cps_click 加 openid + 迁移;integrations/wx_oauth - /c/{code} 授权链路 + 回调 /wx/oauth/cb + cookie 免重复授权;click 带 openid - admin 群详情加「群内微信用户」(领券画像)端点 - 下单归因到人需 user-level sid,本期只做身份+领券侧 Co-Authored-By: Claude Opus 4.8 --- alembic/versions/cps_wx_user.py | 48 +++++++++++++++ app/admin/repositories/cps.py | 47 ++++++++++++++- app/admin/routers/cps.py | 8 +++ app/api/v1/cps_redirect.py | 100 ++++++++++++++++++++++++++++---- app/core/config.py | 11 ++++ app/integrations/wx_oauth.py | 75 ++++++++++++++++++++++++ app/models/__init__.py | 1 + app/models/cps_link.py | 2 + app/models/cps_wx_user.py | 42 ++++++++++++++ app/repositories/cps_link.py | 7 ++- app/repositories/cps_wx_user.py | 43 ++++++++++++++ 11 files changed, 369 insertions(+), 15 deletions(-) create mode 100644 alembic/versions/cps_wx_user.py create mode 100644 app/integrations/wx_oauth.py create mode 100644 app/models/cps_wx_user.py create mode 100644 app/repositories/cps_wx_user.py diff --git a/alembic/versions/cps_wx_user.py b/alembic/versions/cps_wx_user.py new file mode 100644 index 0000000..0e0db37 --- /dev/null +++ b/alembic/versions/cps_wx_user.py @@ -0,0 +1,48 @@ +"""cps 微信用户表 cps_wx_user + cps_click 加 openid + +CPS 落地页微信网页授权拿 openid/昵称头像,做用户级群统计。 + +Revision ID: cps_wx_user +Revises: comparison_debug_fields +Create Date: 2026-06-19 12:00:00.000000 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "cps_wx_user" +down_revision: Union[str, Sequence[str], None] = "comparison_debug_fields" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "cps_wx_user", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("openid", sa.String(64), nullable=False), + sa.Column("unionid", sa.String(64), nullable=True), + sa.Column("nickname", sa.String(128), nullable=True), + sa.Column("headimgurl", sa.String(512), nullable=True), + sa.Column("first_code", sa.String(16), nullable=True), + sa.Column("first_group_id", sa.Integer(), nullable=True), + sa.Column("first_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("last_seen", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + ) + op.create_index("ix_cps_wx_user_openid", "cps_wx_user", ["openid"], unique=True) + op.create_index("ix_cps_wx_user_unionid", "cps_wx_user", ["unionid"]) + op.create_index("ix_cps_wx_user_first_group_id", "cps_wx_user", ["first_group_id"]) + + op.add_column("cps_click", sa.Column("openid", sa.String(64), nullable=True)) + op.create_index("ix_cps_click_openid", "cps_click", ["openid"]) + + +def downgrade() -> None: + op.drop_index("ix_cps_click_openid", table_name="cps_click") + op.drop_column("cps_click", "openid") + op.drop_index("ix_cps_wx_user_first_group_id", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_unionid", table_name="cps_wx_user") + op.drop_index("ix_cps_wx_user_openid", table_name="cps_wx_user") + op.drop_table("cps_wx_user") diff --git a/app/admin/repositories/cps.py b/app/admin/repositories/cps.py index 3271a74..4b1c7c3 100644 --- a/app/admin/repositories/cps.py +++ b/app/admin/repositories/cps.py @@ -9,7 +9,7 @@ from datetime import datetime, timedelta, timezone from decimal import Decimal, InvalidOperation from uuid import uuid4 -from sqlalchemy import desc, select +from sqlalchemy import desc, func, select from sqlalchemy.orm import Session from app.admin.repositories.queries import _as_utc, offset_paginate @@ -19,6 +19,7 @@ from app.models.cps_activity import CpsActivity from app.models.cps_group import CpsGroup from app.models.cps_link import CpsClick from app.models.cps_order import CpsOrder +from app.models.cps_wx_user import CpsWxUser # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账 _INVALID_STATUS = {"4", "5"} @@ -470,3 +471,47 @@ def group_order_daily( "settled_commission_cents": sum(o.commission_cents or 0 for o in settled), } return out + + +def group_wx_users(db: Session, *, group_id: int, limit: int = 200) -> list[dict]: + """群内微信用户(首次来源 first_group_id=group_id)+ 每人在该群的 visit/copy 次数(领券画像)。 + + 按首次进入倒序。copy 次数 = 该用户在该群点了几次「复制口令」(领券意向),visit = 进落地页次数。 + """ + users = ( + db.execute( + select(CpsWxUser) + .where(CpsWxUser.first_group_id == group_id) + .order_by(desc(CpsWxUser.first_seen)) + .limit(limit) + ) + .scalars() + .all() + ) + if not users: + return [] + openids = [u.openid for u in users] + stat: dict[str, dict] = {} + rows = db.execute( + select(CpsClick.openid, CpsClick.event_type, func.count()) + .where(CpsClick.group_id == group_id) + .where(CpsClick.openid.in_(openids)) + .group_by(CpsClick.openid, CpsClick.event_type) + ).all() + for openid, event_type, cnt in rows: + s = stat.setdefault(openid, {"visit": 0, "copy": 0}) + if event_type == "copy": + s["copy"] = cnt + else: + s["visit"] = cnt + return [ + { + "openid": u.openid, + "nickname": u.nickname, + "headimgurl": u.headimgurl, + "first_seen": u.first_seen, + "visit_count": stat.get(u.openid, {}).get("visit", 0), + "copy_count": stat.get(u.openid, {}).get("copy", 0), + } + for u in users + ] diff --git a/app/admin/routers/cps.py b/app/admin/routers/cps.py index 6f892dd..a1fd267 100644 --- a/app/admin/routers/cps.py +++ b/app/admin/routers/cps.py @@ -488,3 +488,11 @@ def group_daily( "days": days, "rows": rows, } + + +@router.get("/groups/{group_id}/wx-users", summary="群内微信用户(领券画像:头像/昵称/领券次数)") +def group_wx_users(group_id: int, db: AdminDb) -> dict: + group = cps_repo.get_group(db, group_id) + if group is None: + raise HTTPException(status_code=404, detail="群不存在") + return {"users": cps_repo.group_wx_users(db, group_id=group_id)} diff --git a/app/api/v1/cps_redirect.py b/app/api/v1/cps_redirect.py index 49a061d..5270586 100644 --- a/app/api/v1/cps_redirect.py +++ b/app/api/v1/cps_redirect.py @@ -8,16 +8,22 @@ from __future__ import annotations import html import json +import logging from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse from sqlalchemy.orm import Session from app.core import media +from app.core.config import settings from app.db.session import get_db +from app.integrations import wx_oauth from app.models.cps_activity import CpsActivity from app.models.cps_link import CpsLink from app.repositories import cps_link as cps_link_repo +from app.repositories import cps_wx_user as cps_wx_user_repo + +logger = logging.getLogger("shagua.cps_redirect") router = APIRouter(tags=["cps-redirect"]) @@ -27,6 +33,16 @@ _FALLBACK_URL = "https://www.meituan.com/" # 淘宝活动未设图时的兜底主视觉(存量已回填,基本只在老 link/异常时触发) _DEFAULT_TAOBAO_IMAGE = "/media/taobao_landing.jpg" +# 微信网页授权拿到的用户标识 cookie(种在 coupon 域,30 天免重复授权) +_WX_OPENID_COOKIE = "wx_openid" +_WX_UINFO_COOKIE = "wx_uinfo" # "1" = 已拿过昵称头像(userinfo),点领券不再跳授权 +_COOKIE_MAX_AGE = 30 * 86400 + + +def _is_wechat(request: Request) -> bool: + """是否微信内置浏览器(网页授权只在微信内有意义,外部浏览器不跳授权)。""" + return "micromessenger" in (request.headers.get("user-agent", "").lower()) + def _client_ip(request: Request) -> str | None: xff = request.headers.get("x-forwarded-for") @@ -35,11 +51,13 @@ def _client_ip(request: Request) -> str | None: return request.client.host if request.client else None -def _record(db: Session, link: CpsLink, request: Request, event_type: str) -> None: +def _record( + db: Session, link: CpsLink, request: Request, event_type: str, openid: str | None = None +) -> None: try: cps_link_repo.record_click( db, link=link, ip=_client_ip(request), - ua=request.headers.get("user-agent"), event_type=event_type, + ua=request.headers.get("user-agent"), event_type=event_type, openid=openid, ) except Exception: pass # 记点击失败不阻断 @@ -53,38 +71,89 @@ def wx_mp_domain_verify() -> PlainTextResponse: return PlainTextResponse("F7wnRQ7xPbhVOWC8") -@router.get("/c/{code}", summary="群发短链落地(记点击 + 跳转/淘宝落地页)") +@router.get("/c/{code}", summary="群发短链落地(微信授权拿 openid + 记点击 + 跳转/淘宝落地页)") def cps_landing(code: str, request: Request, db: Session = Depends(get_db)): link = cps_link_repo.get_by_code(db, code) if link is None: return RedirectResponse(_FALLBACK_URL, status_code=302) - _record(db, link, request, "visit") + openid = request.cookies.get(_WX_OPENID_COOKIE) + # 微信内 + 还没拿到 openid + 配了服务号 → 先 base 静默授权拿 openid,再 302 回本页。 + # 非微信 / 已有 cookie / 未配服务号 → 直接走原逻辑(openid 可能为 None,绝不阻断领券)。 + if openid is None and settings.wx_mp_configured and _is_wechat(request): + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}") + return RedirectResponse(auth_url, status_code=302) + _record(db, link, request, "visit", openid=openid) if link.platform == "taobao": activity = db.get(CpsActivity, link.activity_id) image_url = (activity.image_url if activity else None) or _DEFAULT_TAOBAO_IMAGE - return HTMLResponse(_taobao_landing_html(link.target_url, image_url)) + has_uinfo = request.cookies.get(_WX_UINFO_COOKIE) == "1" + return HTMLResponse( + _taobao_landing_html(link.target_url, image_url, code, openid, has_uinfo) + ) # 美团短链 / 京东链接:直接 302 跳 return RedirectResponse(link.target_url, status_code=302) -@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件)") +@router.get("/wx/oauth/cb", include_in_schema=False) +def wx_oauth_cb(code: str, state: str, db: Session = Depends(get_db)): + """微信网页授权回调。state='base:{原code}'(静默拿 openid) 或 'uinfo:{原code}'(补昵称头像)。 + 换 openid(+userinfo)→ upsert 用户 → 种 cookie → 302 回落地页。任何失败兜底回落地页, + 绝不阻断用户领券。""" + kind, _, orig_code = state.partition(":") + try: + token = wx_oauth.exchange_code(code) # {openid, access_token, scope, ...} + openid = token["openid"] + link = cps_link_repo.get_by_code(db, orig_code) + group_id = link.group_id if link else None + nickname = headimgurl = unionid = None + if kind == "uinfo" and "userinfo" in (token.get("scope") or ""): + info = wx_oauth.get_userinfo(token["access_token"], openid) + nickname, headimgurl, unionid = ( + info.get("nickname"), info.get("headimgurl"), info.get("unionid"), + ) + cps_wx_user_repo.upsert( + db, openid=openid, code=orig_code, group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + except Exception: + logger.exception("[wx_oauth] callback failed state=%s", state) + return RedirectResponse(f"/c/{orig_code}" if orig_code else _FALLBACK_URL, status_code=302) + # userinfo 授权回来带 ?authed=1,落地页据此自动触发复制(把"点领券→授权→复制"衔接为一步) + target = f"/c/{orig_code}" + ("?authed=1" if kind == "uinfo" else "") + resp = RedirectResponse(target, status_code=302) + resp.set_cookie(_WX_OPENID_COOKIE, openid, max_age=_COOKIE_MAX_AGE, httponly=True, samesite="lax") + if nickname: + resp.set_cookie(_WX_UINFO_COOKIE, "1", max_age=_COOKIE_MAX_AGE, samesite="lax") + return resp + + +@router.post("/c/{code}/copy", summary="淘宝落地页点了「复制口令」(记 copy 事件,带 openid)") def cps_copy(code: str, request: Request, db: Session = Depends(get_db)) -> dict: link = cps_link_repo.get_by_code(db, code) if link is not None: - _record(db, link, request, "copy") + _record(db, link, request, "copy", openid=request.cookies.get(_WX_OPENID_COOKIE)) return {"ok": True} -def _taobao_landing_html(token: str, image_url: str) -> str: +def _taobao_landing_html( + token: str, image_url: str, code: str, openid: str | None, has_uinfo: bool +) -> str: """淘宝落地页 H5(主视觉图按活动传入,复制淘口令按钮在 75% 屏高)。 - image_url 转绝对(落地页 coupon 域,相对也可达,绝对更稳)后经 html.escape 嵌入 - (防属性注入);token 经 json.dumps 安全嵌入 JS。 + image_url 经 html.escape 嵌入 (防注入);token 经 json.dumps 安全嵌入 JS。 + uinfo_url:已有 openid 但还没拿过昵称头像时,生成 userinfo 授权链接 —— 用户点「领券」 + 时先跳它(交互触发,避免微信快照页),授权回来自动复制。已拿过 / 无 openid 则为空,直接复制。 """ safe_img = html.escape(media.to_abs_media_url(image_url) or image_url, quote=True) + uinfo_url = "" + if openid and not has_uinfo and settings.wx_mp_configured: + redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" + uinfo_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_userinfo", f"uinfo:{code}") return ( _TAOBAO_HTML .replace("__IMAGE_URL__", safe_img) + .replace("__UINFO_URL__", json.dumps(uinfo_url)) .replace("__TOKEN_JS__", json.dumps(token)) ) @@ -112,6 +181,7 @@ body{font-family:-apple-system,"PingFang SC",sans-serif;background:#fff0ef;color
""" diff --git a/app/core/config.py b/app/core/config.py index bfc4139..8862476 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -95,6 +95,17 @@ class Settings(BaseSettings): """美团 CPS 凭证齐全(缺则接口返空,而非 502)。""" return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET) + # ===== 微信服务号(网页授权) ===== + # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。 + # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 + WX_MP_APPID: str = "" + WX_MP_SECRET: str = "" + + @property + def wx_mp_configured(self) -> bool: + """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" + return bool(self.WX_MP_APPID and self.WX_MP_SECRET) + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 diff --git a/app/integrations/wx_oauth.py b/app/integrations/wx_oauth.py new file mode 100644 index 0000000..0af50d6 --- /dev/null +++ b/app/integrations/wx_oauth.py @@ -0,0 +1,75 @@ +"""微信服务号网页授权:构造授权链接 + code 换 openid + 拉 userinfo。 + +用于 CPS 落地页(coupon.shaguabijia.com)在微信内拿用户身份。流程(见微信文档): + 授权链接(snsapi_base|snsapi_userinfo) → 回调带 code → sns/oauth2/access_token 换 openid + → (userinfo 时)sns/userinfo 拿昵称头像。 + +注意: + - 这里的 access_token 是【网页授权 token】,与基础 access_token(cgi-bin/token)不同, + 走 sns/* 接口,不需要 IP 白名单。 + - secret 只在服务器,不下发客户端。 + - 用 WX_MP_APPID/SECRET(已认证服务号),区别于 WECHAT_APP_ID(App 移动应用)。 +""" +from __future__ import annotations + +from urllib.parse import quote + +import httpx + +from app.core.config import settings + +_API = "https://api.weixin.qq.com" +_AUTHORIZE = "https://open.weixin.qq.com/connect/oauth2/authorize" +_TIMEOUT = 10 + + +class WxOauthError(Exception): + """网页授权调用失败(凭证/网络/code 失效等)。调用方兜底为无 openid。""" + + +def build_authorize_url(redirect_uri: str, scope: str, state: str) -> str: + """构造网页授权跳转链接。scope: 'snsapi_base'(静默,仅 openid) | 'snsapi_userinfo'(昵称头像)。 + + 微信要求 redirect_uri urlEncode、参数顺序固定、结尾 #wechat_redirect。 + """ + return ( + f"{_AUTHORIZE}?appid={settings.WX_MP_APPID}" + f"&redirect_uri={quote(redirect_uri, safe='')}" + f"&response_type=code&scope={scope}&state={quote(state, safe='')}" + f"#wechat_redirect" + ) + + +def exchange_code(code: str) -> dict: + """code 换网页授权 access_token + openid。返回含 openid/access_token/scope(/unionid)。 + + code 5 分钟内一次性有效。失败(errcode)抛 WxOauthError。 + """ + resp = httpx.get( + f"{_API}/sns/oauth2/access_token", + params={ + "appid": settings.WX_MP_APPID, + "secret": settings.WX_MP_SECRET, + "code": code, + "grant_type": "authorization_code", + }, + timeout=_TIMEOUT, + ) + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"exchange_code failed: {data}") + return data + + +def get_userinfo(access_token: str, openid: str) -> dict: + """拉用户信息(nickname/headimgurl/unionid)。需 scope=snsapi_userinfo 的 access_token。""" + resp = httpx.get( + f"{_API}/sns/userinfo", + params={"access_token": access_token, "openid": openid, "lang": "zh_CN"}, + timeout=_TIMEOUT, + ) + resp.encoding = "utf-8" # 昵称含中文/emoji,强制 utf-8 避免乱码 + data = resp.json() + if "openid" not in data: + raise WxOauthError(f"get_userinfo failed: {data}") + return data diff --git a/app/models/__init__.py b/app/models/__init__.py index 1955e31..768d192 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -10,6 +10,7 @@ from app.models.cps_activity import CpsActivity # noqa: F401 from app.models.cps_group import CpsGroup # noqa: F401 from app.models.cps_link import CpsClick, CpsLink # noqa: F401 from app.models.cps_order import CpsOrder # noqa: F401 +from app.models.cps_wx_user import CpsWxUser # noqa: F401 from app.models.comparison_milestone import ComparisonMilestoneClaim # noqa: F401 from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, diff --git a/app/models/cps_link.py b/app/models/cps_link.py index 2702c8a..bb88e6b 100644 --- a/app/models/cps_link.py +++ b/app/models/cps_link.py @@ -49,6 +49,8 @@ class CpsClick(Base): ) ip: Mapped[str | None] = mapped_column(String(64), nullable=True) ua: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计 + openid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) clicked_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True, nullable=False ) diff --git a/app/models/cps_wx_user.py b/app/models/cps_wx_user.py new file mode 100644 index 0000000..669b26c --- /dev/null +++ b/app/models/cps_wx_user.py @@ -0,0 +1,42 @@ +"""CPS 落地页微信用户(cps_wx_user)。 + +用户在微信内打开群发短链落地页 /c/{code},经服务号网页授权: + - base 静默:拿 openid(唯一标识,统计主力) + - userinfo(点领券触发):补 nickname/headimgurl/unionid + +按 openid 唯一,记录首次来源群(first_group_id),用于群内用户级统计(谁领了券)。 +下单归因到人需 user-level sid(另一期),本表只承载身份 + 领券/点击侧。 +""" +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 CpsWxUser(Base): + __tablename__ = "cps_wx_user" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + # 服务号下用户唯一标识(网页授权拿到) + openid: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False) + # 开放平台 unionid(服务号绑开放平台 + scope=userinfo 才有);跨 App/服务号统一用户 + unionid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + # 昵称/头像:userinfo 授权后才有(base 阶段为空) + nickname: Mapped[str | None] = mapped_column(String(128), nullable=True) + headimgurl: Mapped[str | None] = mapped_column(String(512), nullable=True) + # 首次进入来源(从哪个群的链接授权进来) + first_code: Mapped[str | None] = mapped_column(String(16), nullable=True) + first_group_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True) + first_seen: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + last_seen: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False + ) + + def __repr__(self) -> str: # pragma: no cover + return f"" diff --git a/app/repositories/cps_link.py b/app/repositories/cps_link.py index 2e34970..00b9c1f 100644 --- a/app/repositories/cps_link.py +++ b/app/repositories/cps_link.py @@ -52,12 +52,13 @@ def create_link( def record_click( db: Session, *, link: CpsLink, ip: str | None = None, ua: str | None = None, - event_type: str = "visit", + event_type: str = "visit", openid: str | None = None, ) -> None: - """记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。""" + """记一条点击。event_type: visit(进落地页/被跳转) | copy(淘宝点了"复制口令")。 + openid: 微信网页授权拿到的用户标识(非微信/未授权为空),做用户级群统计。""" db.add(CpsClick( link_id=link.id, group_id=link.group_id, sid=link.sid, event_type=event_type, - ip=ip, ua=(ua[:500] if ua else None), + ip=ip, ua=(ua[:500] if ua else None), openid=openid, )) db.commit() diff --git a/app/repositories/cps_wx_user.py b/app/repositories/cps_wx_user.py new file mode 100644 index 0000000..9fde31e --- /dev/null +++ b/app/repositories/cps_wx_user.py @@ -0,0 +1,43 @@ +"""cps_wx_user 数据访问:按 openid upsert。""" +from __future__ import annotations + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app.models.cps_wx_user import CpsWxUser + + +def get_by_openid(db: Session, openid: str) -> CpsWxUser | None: + return db.execute( + select(CpsWxUser).where(CpsWxUser.openid == openid) + ).scalar_one_or_none() + + +def upsert( + db: Session, *, openid: str, code: str | None = None, group_id: int | None = None, + nickname: str | None = None, headimgurl: str | None = None, unionid: str | None = None, +) -> CpsWxUser: + """按 openid upsert。 + + - 首次:建行,记来源 code/group。 + - 已存在:仅在传入非 None 时更新昵称/头像/unionid(base 阶段为 None,不覆盖已有画像); + 并刷新 last_seen(显式 set,因 onupdate 仅在字段有变更时触发)。 + """ + u = get_by_openid(db, openid) + if u is None: + u = CpsWxUser( + openid=openid, first_code=code, first_group_id=group_id, + nickname=nickname, headimgurl=headimgurl, unionid=unionid, + ) + db.add(u) + else: + if nickname is not None: + u.nickname = nickname + if headimgurl is not None: + u.headimgurl = headimgurl + if unionid is not None: + u.unionid = unionid + u.last_seen = func.now() + db.commit() + db.refresh(u) + return u From 733a51260f79bb58be1df281a17d11311017b35a Mon Sep 17 00:00:00 2001 From: marco Date: Fri, 19 Jun 2026 12:03:58 +0800 Subject: [PATCH 3/7] =?UTF-8?q?feat(cps):=20=E5=BE=AE=E4=BF=A1=E7=BD=91?= =?UTF-8?q?=E9=A1=B5=E6=8E=88=E6=9D=83=E5=8A=A0=E6=80=BB=E5=BC=80=E5=85=B3?= =?UTF-8?q?=20WX=5FMP=5FOAUTH=5FENABLED(=E9=BB=98=E8=AE=A4=E5=85=B3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 服务号认证审核中,先关掉落地页授权(走原逻辑不跳授权/不拿 openid),整套微信代码保留。 审核通过后 .env 置 WX_MP_OAUTH_ENABLED=true + 重启即启用,无需改代码。 Co-Authored-By: Claude Opus 4.8 --- app/api/v1/cps_redirect.py | 4 ++-- app/core/config.py | 8 ++++++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/app/api/v1/cps_redirect.py b/app/api/v1/cps_redirect.py index 5270586..ee7aa0b 100644 --- a/app/api/v1/cps_redirect.py +++ b/app/api/v1/cps_redirect.py @@ -79,7 +79,7 @@ def cps_landing(code: str, request: Request, db: Session = Depends(get_db)): openid = request.cookies.get(_WX_OPENID_COOKIE) # 微信内 + 还没拿到 openid + 配了服务号 → 先 base 静默授权拿 openid,再 302 回本页。 # 非微信 / 已有 cookie / 未配服务号 → 直接走原逻辑(openid 可能为 None,绝不阻断领券)。 - if openid is None and settings.wx_mp_configured and _is_wechat(request): + if openid is None and settings.wx_oauth_active and _is_wechat(request): redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" auth_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_base", f"base:{code}") return RedirectResponse(auth_url, status_code=302) @@ -147,7 +147,7 @@ def _taobao_landing_html( """ safe_img = html.escape(media.to_abs_media_url(image_url) or image_url, quote=True) uinfo_url = "" - if openid and not has_uinfo and settings.wx_mp_configured: + if openid and not has_uinfo and settings.wx_oauth_active: redirect_uri = f"{settings.CPS_REDIRECT_BASE.rstrip('/')}/wx/oauth/cb" uinfo_url = wx_oauth.build_authorize_url(redirect_uri, "snsapi_userinfo", f"uinfo:{code}") return ( diff --git a/app/core/config.py b/app/core/config.py index 8862476..76621e1 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -100,12 +100,20 @@ class Settings(BaseSettings): # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。 WX_MP_APPID: str = "" WX_MP_SECRET: str = "" + # 落地页微信网页授权【总开关】。默认关:服务号认证审核中/未就绪时,落地页走原逻辑 + # (不跳授权、不拿 openid),整套微信代码保留。审核通过后 .env 置 true + 重启即启用,无需改代码。 + WX_MP_OAUTH_ENABLED: bool = False @property def wx_mp_configured(self) -> bool: """服务号网页授权凭证齐全(缺则落地页不发起授权,降级为无 openid)。""" return bool(self.WX_MP_APPID and self.WX_MP_SECRET) + @property + def wx_oauth_active(self) -> bool: + """落地页是否真正发起微信授权 = 凭证齐全 且 总开关开。""" + return self.wx_mp_configured and self.WX_MP_OAUTH_ENABLED + # ===== 微信支付(商家转账到零钱 / 提现)===== # 真实凭证放 .env(已 gitignore),证书 .pem 放 secrets/。WECHAT_APP_ID 同时用于 # 微信登录(code 换 openid)与转账,必须与 App 端开放平台 appid 一致。 From 4d1162053ecda47618754b092f593b0521571654 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E4=B8=96=E7=9D=BF?= <2839904623@qq.com> Date: Fri, 19 Jun 2026 22:35:12 +0800 Subject: [PATCH 4/7] =?UTF-8?q?feat(device):=20=E6=97=A0=E9=9A=9C=E7=A2=8D?= =?UTF-8?q?=E6=8E=89=E7=BA=BF=E5=90=8E=E7=BD=AE=E6=A3=80=E6=B5=8B=20?= =?UTF-8?q?=E2=80=94=20kill=5Falert=5Fpending=20+=20liveness=20=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2/ack=20=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 推送暂缓改 pull:worker 检出掉线即置 kill_alert_pending(与 liveness_state 解耦,心跳恢复 touch_heartbeat 不清,规避竞态),客户端进 App 拉 GET /device/liveness 命中→弹自启动引导,POST /device/liveness/ack 清。含 alembic 加列(临时 sqlite 验通)。spec: accessibility-liveness-pull-prompt.md Co-Authored-By: Claude Opus 4.8 (1M context) --- alembic/versions/device_kill_alert_pending.py | 39 +++++++++++++++++++ app/api/v1/device.py | 30 ++++++++++++++ app/core/heartbeat_monitor_worker.py | 2 +- app/models/device.py | 6 +++ app/repositories/device.py | 20 +++++++++- app/schemas/device.py | 14 +++++++ 6 files changed, 109 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/device_kill_alert_pending.py diff --git a/alembic/versions/device_kill_alert_pending.py b/alembic/versions/device_kill_alert_pending.py new file mode 100644 index 0000000..f4bc6fc --- /dev/null +++ b/alembic/versions/device_kill_alert_pending.py @@ -0,0 +1,39 @@ +"""device.kill_alert_pending (无障碍掉线「后置检测」待提醒标记) + +Revision ID: device_kill_alert_pending +Revises: merge_cps_device_liveness +Create Date: 2026-06-19 + +后置检测 pull 版(spec/accessibility-liveness-pull-prompt.md):worker 检出心跳掉线即置此标记, +客户端下次进 App 拉取到 → 弹「开启自启动」引导,交互后 ack 清回。与 liveness_state 解耦,故单列。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'device_kill_alert_pending' +down_revision: Union[str, Sequence[str], None] = 'merge_cps_device_liveness' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # server_default=sa.false() 跨方言: SQLite 渲染 0 / PG 渲染 false(别用 sa.text("0"),PG 布尔严格会报错), + # 给存量行回填 False。 + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.add_column( + sa.Column( + 'kill_alert_pending', + sa.Boolean(), + nullable=False, + server_default=sa.false(), + ) + ) + + +def downgrade() -> None: + with op.batch_alter_table('device', schema=None) as batch_op: + batch_op.drop_column('kill_alert_pending') diff --git a/app/api/v1/device.py b/app/api/v1/device.py index d56839d..56a5a85 100644 --- a/app/api/v1/device.py +++ b/app/api/v1/device.py @@ -19,6 +19,8 @@ from app.schemas.device import ( DeviceOut, DeviceRegisterRequest, HeartbeatRequest, + LivenessAckRequest, + LivenessOut, OkResponse, ) @@ -64,3 +66,31 @@ def report_heartbeat( registration_id=req.registration_id, ) return OkResponse() + + +@router.get("/liveness", response_model=LivenessOut, summary="查询本机掉线告警(后置检测)") +def get_liveness( + device_id: str, + user: CurrentUser, + db: DbSession, +) -> LivenessOut: + """客户端进 App 时拉取: 本机是否被判定掉线过(kill_alert_pending)。 + + 设备从未注册过 → 返回默认(无告警)。命中 → 客户端弹「开启自启动」引导, 之后调 /liveness/ack 清。 + 见 spec: spec/accessibility-liveness-pull-prompt.md。 + """ + device = device_repo.get_device(db, user_id=user.id, device_id=device_id) + if device is None: + return LivenessOut() + return LivenessOut.model_validate(device) + + +@router.post("/liveness/ack", response_model=OkResponse, summary="确认已提醒(清掉线告警)") +def ack_liveness( + req: LivenessAckRequest, + user: CurrentUser, + db: DbSession, +) -> OkResponse: + """客户端已弹过「开启自启动」引导 → 清掉「待提醒」标记(下次真掉线 worker 再置)。""" + device_repo.ack_kill_alert(db, user_id=user.id, device_id=req.device_id) + return OkResponse() diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py index 607fe8d..d5c6a16 100644 --- a/app/core/heartbeat_monitor_worker.py +++ b/app/core/heartbeat_monitor_worker.py @@ -88,7 +88,7 @@ def _scan_once(timeout_minutes: int) -> dict: silent = _silent_seconds(device.last_heartbeat_at) logger.warning( "🔴 [掉线检测] user_id=%s device_id=%s 已 %s 秒无心跳(阈值 %d 分钟)" - " → 判定 App 已被杀/无障碍已停。【本应推送通知提醒用户重开;推送暂未接,先终端打印代替】", + " → 判定 App 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 App 将弹「开启自启动」引导(后置检测);推送本期未接】", device.user_id, device.device_id, silent if silent is not None else "?", diff --git a/app/models/device.py b/app/models/device.py index 520b325..290bc9f 100644 --- a/app/models/device.py +++ b/app/models/device.py @@ -65,6 +65,12 @@ class Device(Base): notified_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) + # 掉线告警「待客户端提醒」标记(后置检测 pull 版, 见 spec accessibility-liveness-pull-prompt.md)。 + # 与 liveness_state 解耦: worker 检出掉线即置 True; touch_heartbeat(心跳恢复)不动它 + # → 规避「服务随 App 重启先发心跳、state 被重置回 alive → 客户端进 App 漏看」竞态; 只由客户端 ack 清。 + kill_alert_pending: Mapped[bool] = mapped_column( + Boolean, nullable=False, default=False + ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False diff --git a/app/repositories/device.py b/app/repositories/device.py index 3e7c35c..b1c40d8 100644 --- a/app/repositories/device.py +++ b/app/repositories/device.py @@ -98,9 +98,27 @@ def list_overdue(db: Session, *, timeout_minutes: int) -> list[Device]: def mark_notified(db: Session, *, device_id_pk: int) -> None: - """标记已推送告警(状态机进入 notified,避免重复推送)。""" + """标记掉线已检出(状态机进入 notified,避免每轮重复处理)+ 置「待客户端提醒」标记。 + + kill_alert_pending 与 state 解耦(见 spec accessibility-liveness-pull-prompt.md): + 心跳恢复(touch_heartbeat)只重置 state、不动此标记,故客户端下次进 App 必能拉到这次掉线。 + """ device = db.get(Device, device_id_pk) if device is not None: device.liveness_state = "notified" device.notified_at = datetime.now(timezone.utc) + device.kill_alert_pending = True + db.commit() + + +def get_device(db: Session, *, user_id: int, device_id: str) -> Device | None: + """按 (user_id, device_id) 取设备(liveness 查询用)。""" + return _get(db, user_id=user_id, device_id=device_id) + + +def ack_kill_alert(db: Session, *, user_id: int, device_id: str) -> None: + """客户端已弹过「开启自启动」引导 → 清「待提醒」标记(幂等; 下次真掉线 worker 再置)。""" + device = _get(db, user_id=user_id, device_id=device_id) + if device is not None and device.kill_alert_pending: + device.kill_alert_pending = False db.commit() diff --git a/app/schemas/device.py b/app/schemas/device.py index 6343dc8..3254e06 100644 --- a/app/schemas/device.py +++ b/app/schemas/device.py @@ -34,3 +34,17 @@ class DeviceOut(BaseModel): class OkResponse(BaseModel): ok: bool = True + + +class LivenessOut(BaseModel): + """本机掉线告警状态(后置检测 pull 版)。设备从未注册过 → 全默认(= 无告警)。""" + model_config = ConfigDict(from_attributes=True) + + kill_alert_pending: bool = False + liveness_state: str = "unknown" + ever_protected: bool = False + last_heartbeat_at: datetime | None = None + + +class LivenessAckRequest(BaseModel): + device_id: str From 95a07f75b75d0ce632526f669435f6621e12e7cc Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 20 Jun 2026 01:35:17 +0800 Subject: [PATCH 5/7] =?UTF-8?q?feat(ad):=20=E7=A9=BF=E5=B1=B1=E7=94=B2?= =?UTF-8?q?=E5=B9=BF=E5=91=8A=E9=85=8D=E7=BD=AE=E5=90=8E=E5=8F=B0=E5=8C=96?= =?UTF-8?q?(app=5Fid/=E5=90=84=E4=BD=8DID/=E9=AA=8C=E7=AD=BE=E5=AF=86?= =?UTF-8?q?=E9=92=A5/=E5=90=84=E5=9C=BA=E6=99=AF=E5=BC=80=E5=85=B3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - app_config 存 ad_config(默认=客户端当前内置 ID,空库维持现状) - GET /platform/ad-config 下发客户端(不含验签密钥 reward_mkey) - admin GET/PATCH /admin/api/ad-config 读写(operator/finance + 审计,mkey 不记明文) - pangle 回调验签 m-key 优先读后台配置、.env 兜底兼容 - 比价/领券信息流位拆分(compare/coupon_feed_code_id) - 客户端需发版接入 /platform/ad-config 才生效;本次后端就绪、不影响现状 Co-Authored-By: Claude Opus 4.8 --- app/admin/main.py | 2 ++ app/admin/routers/ad_config.py | 45 +++++++++++++++++++++++++++++++ app/admin/schemas/ad_config.py | 30 +++++++++++++++++++++ app/api/v1/ad.py | 13 +++++++-- app/api/v1/platform.py | 17 ++++++++++++ app/repositories/app_config.py | 48 ++++++++++++++++++++++++++++++++++ app/schemas/platform.py | 15 +++++++++++ 7 files changed, 168 insertions(+), 2 deletions(-) create mode 100644 app/admin/routers/ad_config.py create mode 100644 app/admin/schemas/ad_config.py diff --git a/app/admin/main.py b/app/admin/main.py index b718441..eb46a7c 100644 --- a/app/admin/main.py +++ b/app/admin/main.py @@ -14,6 +14,7 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.admin.routers.ad_audit import router as ad_audit_router +from app.admin.routers.ad_config import router as ad_config_router from app.admin.routers.ad_revenue import router as ad_revenue_router from app.admin.routers.admins import router as admins_router from app.admin.routers.audit import router as audit_router @@ -95,4 +96,5 @@ admin_app.include_router(config_router) admin_app.include_router(comparison_router) admin_app.include_router(cps_router) admin_app.include_router(ad_audit_router) +admin_app.include_router(ad_config_router) admin_app.include_router(ad_revenue_router) diff --git a/app/admin/routers/ad_config.py b/app/admin/routers/ad_config.py new file mode 100644 index 0000000..c46068c --- /dev/null +++ b/app/admin/routers/ad_config.py @@ -0,0 +1,45 @@ +"""admin 广告配置:读/写穿山甲 app_id / 各场景广告位ID / 验签密钥 / 各场景开关。 + +存在 app_config 表的 ad_config dict(见 repositories/app_config.get_ad_config/set_ad_config)。 +客户端经 /api/v1/platform/ad-config 拉取(不含 reward_mkey)。权限 operator/finance + 审计。 +""" +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Depends, Request + +from app.admin.audit import write_audit +from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role +from app.admin.schemas.ad_config import AdConfigOut, AdConfigUpdate +from app.models.admin import AdminUser +from app.repositories import app_config + +router = APIRouter( + prefix="/admin/api/ad-config", + tags=["admin-ad-config"], + dependencies=[Depends(get_current_admin)], +) + + +@router.get("", response_model=AdConfigOut, summary="广告配置(穿山甲ID/验签密钥/各场景开关)") +def get_ad_config(db: AdminDb) -> AdConfigOut: + return AdConfigOut(**app_config.get_ad_config(db)) + + +@router.patch("", response_model=AdConfigOut, summary="改广告配置(部分更新,带审计)") +def update_ad_config( + body: AdConfigUpdate, + request: Request, + admin: Annotated[AdminUser, Depends(require_role("operator", "finance"))], + db: AdminDb, +) -> AdConfigOut: + data = body.model_dump(exclude_none=True) + app_config.set_ad_config(db, data, admin_id=admin.id, commit=False) + # 审计只记改了哪些字段,不记 mkey 明文(机密) + write_audit( + db, admin, action="ad_config.set", target_type="ad_config", target_id="ad_config", + detail={"changed": sorted(data.keys())}, ip=get_client_ip(request), commit=False, + ) + db.commit() + return AdConfigOut(**app_config.get_ad_config(db)) diff --git a/app/admin/schemas/ad_config.py b/app/admin/schemas/ad_config.py new file mode 100644 index 0000000..3f5c033 --- /dev/null +++ b/app/admin/schemas/ad_config.py @@ -0,0 +1,30 @@ +"""admin 广告配置 schemas(穿山甲 app_id/各位ID/验签密钥/各场景开关)。""" +from __future__ import annotations + +from pydantic import BaseModel + + +class AdConfigOut(BaseModel): + """admin 读到的完整广告配置(含 reward_mkey;admin 有权见,但绝不经 /platform 下发客户端)。""" + + app_id: str + reward_code_id: str + compare_feed_code_id: str + coupon_feed_code_id: str + reward_mkey: str + reward_enabled: bool + compare_ad_enabled: bool + coupon_ad_enabled: bool + + +class AdConfigUpdate(BaseModel): + """部分更新:只改传入(非 None)字段。空串是合法值(如清空 mkey 回退 .env)。""" + + app_id: str | None = None + reward_code_id: str | None = None + compare_feed_code_id: str | None = None + coupon_feed_code_id: str | None = None + reward_mkey: str | None = None + reward_enabled: bool | None = None + compare_ad_enabled: bool | None = None + coupon_ad_enabled: bool | None = None diff --git a/app/api/v1/ad.py b/app/api/v1/ad.py index bb05891..cf866b2 100644 --- a/app/api/v1/ad.py +++ b/app/api/v1/ad.py @@ -24,6 +24,7 @@ from app.repositories import ad_ecpm as crud_ecpm from app.repositories import ad_feed_reward as crud_feed from app.repositories import ad_reward as crud_ad from app.repositories import ad_watch as crud_watch +from app.repositories import app_config from app.repositories import signin as crud_signin from app.schemas.ad import ( AdRewardStatusOut, @@ -80,14 +81,22 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut: 验签失败 403(留给真请求重试);参数缺/坏或 user 不存在 → is_verify=false + reason(不发,不重试)。 granted / capped → is_verify=true + reason=0。 """ - if not settings.pangle_callback_configured: + if not settings.PANGLE_CALLBACK_ENABLED: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="pangle callback not configured" ) params = dict(request.query_params) - if not pangle.verify_callback_sign_any(params, settings.pangle_reward_secrets): + # 验签密钥:admin 后台配的 reward_mkey 优先,.env 的 PANGLE_REWARD_SECRET* 兜底(兼容/过渡)。 + # 换激励位时后台同步换 mkey 即可,无需改 .env。两者都空才视为未配置。 + ad_mkey = app_config.get_ad_config(db).get("reward_mkey") or "" + secrets = ([ad_mkey] if ad_mkey else []) + settings.pangle_reward_secrets + if not secrets: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="pangle callback not configured" + ) + if not pangle.verify_callback_sign_any(params, secrets): logger.warning("pangle callback bad sign trans_id=%s", params.get("trans_id")) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="bad sign") diff --git a/app/api/v1/platform.py b/app/api/v1/platform.py index f839a58..31908bd 100644 --- a/app/api/v1/platform.py +++ b/app/api/v1/platform.py @@ -17,6 +17,7 @@ from app.repositories import app_config from app.repositories import ops_marquee as marquee_crud from app.repositories import ops_stat as crud from app.schemas.platform import ( + AdConfigPublicOut, AppFlagsOut, AppVersionOut, PlatformStatsOut, @@ -54,6 +55,22 @@ def flags(db: DbSession) -> AppFlagsOut: ) +@router.get("/ad-config", response_model=AdConfigPublicOut, summary="客户端拉广告配置(穿山甲ID+场景开关,不鉴权)") +def ad_config(db: DbSession) -> AdConfigPublicOut: + """客户端启动/每场广告前拉,缓存后用:app_id + 各位ID + 各场景开关。 + 不含验签密钥;空库回退默认(=客户端内置值,维持现状)。""" + c = app_config.get_ad_config(db) + return AdConfigPublicOut( + app_id=c["app_id"], + reward_code_id=c["reward_code_id"], + compare_feed_code_id=c["compare_feed_code_id"], + coupon_feed_code_id=c["coupon_feed_code_id"], + reward_enabled=c["reward_enabled"], + compare_ad_enabled=c["compare_ad_enabled"], + coupon_ad_enabled=c["coupon_ad_enabled"], + ) + + @router.get("/app-version", response_model=AppVersionOut, summary="最新 App 版本(OTA 检查更新,不鉴权)") def app_version(db: DbSession) -> AppVersionOut: """客户端启动 / 手动检查更新时拉取。不鉴权:版本信息非敏感,且检查更新可能在登录前。 diff --git a/app/repositories/app_config.py b/app/repositories/app_config.py index b6f67a4..40bfc80 100644 --- a/app/repositories/app_config.py +++ b/app/repositories/app_config.py @@ -91,3 +91,51 @@ def set_app_version(db: Session, data: dict, *, commit: bool = True) -> AppConfi else: db.flush() return row + + +# ── 穿山甲广告配置(admin 可配,客户端动态拉)────────────────────────────────── +# 同 app_version:结构化 dict,复用 AppConfig 表不进 CONFIG_DEFS(它是一组关联配置,不是散项)。 +# 默认值 = 客户端当前内置的硬编码 ID(AdConfig.kt),空库时下发默认 = 维持现状。 +# reward_mkey 是 GroMore S2S 验签密钥(机密),只后端验签用、绝不下发客户端(见 platform/ad-config)。 +AD_CONFIG_KEY = "ad_config" +_AD_CONFIG_DEFAULTS: dict[str, Any] = { + "app_id": "5830519", # 穿山甲应用ID(正式) + "reward_code_id": "104099389", # 福利页激励视频位 + "compare_feed_code_id": "104090333", # 比价信息流位 + "coupon_feed_code_id": "104090333", # 领券信息流位(初始同比价,运营可拆) + "reward_mkey": "", # 激励位 GroMore 验签密钥(空则回退 .env PANGLE_REWARD_SECRET*) + "reward_enabled": True, # 福利激励视频开关 + "compare_ad_enabled": True, # 比价广告开关 + "coupon_ad_enabled": True, # 领券广告开关 +} + + +def get_ad_config(db: Session) -> dict: + """读广告配置。DB 无则返回默认(=客户端内置值,维持现状);DB 有则与默认 merge + (新增字段向后兼容,老记录缺的字段用默认补)。""" + row = db.get(AppConfig, AD_CONFIG_KEY) + merged = dict(_AD_CONFIG_DEFAULTS) + if row is not None and isinstance(row.value, dict): + merged.update(row.value) + return merged + + +def set_ad_config(db: Session, data: dict, *, admin_id: int, commit: bool = True) -> AppConfig: + """admin 写广告配置。与现有值 merge(部分更新),只接受已知字段(防脏键)。""" + row = db.get(AppConfig, AD_CONFIG_KEY) + merged = dict(_AD_CONFIG_DEFAULTS) + if row is not None and isinstance(row.value, dict): + merged.update(row.value) + merged.update({k: v for k, v in data.items() if k in _AD_CONFIG_DEFAULTS}) + if row is None: + row = AppConfig(key=AD_CONFIG_KEY, value=merged, updated_by_admin_id=admin_id) + db.add(row) + else: + row.value = merged + row.updated_by_admin_id = admin_id + if commit: + db.commit() + db.refresh(row) + else: + db.flush() + return row diff --git a/app/schemas/platform.py b/app/schemas/platform.py index 6968271..4c97bd6 100644 --- a/app/schemas/platform.py +++ b/app/schemas/platform.py @@ -30,6 +30,21 @@ class AppFlagsOut(BaseModel): comparing_ad_enabled: bool # 比价/领券期是否展示信息流广告(远程 kill-switch) +class AdConfigPublicOut(BaseModel): + """下发给客户端的穿山甲广告配置(不鉴权,客户端缓存后用)。 + + ⚠️ 不含 reward_mkey(GroMore 验签密钥,机密,只后端验签用,绝不下发)。 + """ + + app_id: str # 穿山甲应用ID(改了客户端需冷启才生效,SDK init 一次性读) + reward_code_id: str # 福利页激励视频位 + compare_feed_code_id: str # 比价信息流位 + coupon_feed_code_id: str # 领券信息流位 + reward_enabled: bool # 福利激励视频开关 + compare_ad_enabled: bool # 比价广告开关 + coupon_ad_enabled: bool # 领券广告开关 + + class AppVersionOut(BaseModel): """最新 App 版本信息(OTA 检查更新,不鉴权)。 From d8358a6816fc6a359f13c0aaabb0cbb341e54280 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 20 Jun 2026 02:03:33 +0800 Subject: [PATCH 6/7] =?UTF-8?q?fix(user):=20=E6=B3=A8=E9=94=80=E8=B4=A6?= =?UTF-8?q?=E5=8F=B7=E5=89=8D=E7=BD=AE=E6=A0=A1=E9=AA=8C=E8=B5=84=E9=87=91?= =?UTF-8?q?=E4=BD=99=E9=A2=9D=20+=20=E9=87=8A=E6=94=BE=E5=BE=AE=E4=BF=A1/?= =?UTF-8?q?=E9=82=80=E8=AF=B7=E7=A0=81=E5=94=AF=E4=B8=80=E7=BA=A6=E6=9D=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - delete_account 端点: 有可提现现金余额 / 在审提现单(reviewing)时拒绝注销(409), 防余额锁死在 deleted 行无法退出; pending 提现已在途不拦(对账 worker 不依赖 user.status) - soft_delete_account: 注销连带清 wechat_openid/nickname/avatar(释放微信绑定槽, 否则该微信永久无法再被任何账号绑定→提现通道锁死) + invite_code(释放邀请码唯一槽) - wallet 新增 get_cash_balance_cents(只读, 不建账户) 供注销前置校验; 复用 has_reviewing_withdraw - 新增 tests/test_delete_account.py Co-Authored-By: Claude Opus 4.8 --- app/api/v1/user.py | 12 +++ app/repositories/user.py | 17 ++++- app/repositories/wallet.py | 9 +++ tests/test_delete_account.py | 141 +++++++++++++++++++++++++++++++++++ 4 files changed, 177 insertions(+), 2 deletions(-) create mode 100644 tests/test_delete_account.py diff --git a/app/api/v1/user.py b/app/api/v1/user.py index 6be689f..dc23388 100644 --- a/app/api/v1/user.py +++ b/app/api/v1/user.py @@ -18,6 +18,7 @@ from app.api.deps import CurrentUser, DbSession from app.core import media from app.repositories import onboarding as onboarding_repo from app.repositories import user as user_repo +from app.repositories import wallet as wallet_repo from app.schemas.auth import UserOut from app.schemas.user import ( OkResponse, @@ -87,6 +88,17 @@ def onboarding_status( @router.delete("", response_model=OkResponse, summary="注销账号(软删除)") def delete_account(user: CurrentUser, db: DbSession) -> OkResponse: + # 资金前置校验:注销是软删 + 匿名化,余额一旦留在 deleted 行就锁死(既退不出、新号也 + # 拿不回)。有可提现现金 / 在审提现单 → 拒绝注销,引导用户先把钱处理掉。 + # (pending 提现转账已在途、对账 worker 不依赖 user.status 照常到账,故不拦 pending。) + if wallet_repo.get_cash_balance_cents(db, user.id) > 0: + raise HTTPException( + status_code=409, detail="账户还有未提现的现金余额,请先提现后再注销" + ) + if wallet_repo.has_reviewing_withdraw(db, user.id): + raise HTTPException( + status_code=409, detail="有提现正在审核中,请等审核完成后再注销" + ) media.delete_avatar(user.avatar_url) user_repo.soft_delete_account(db, user) logger.info("delete account user_id=%d", user.id) diff --git a/app/repositories/user.py b/app/repositories/user.py index 33c53ae..7753e33 100644 --- a/app/repositories/user.py +++ b/app/repositories/user.py @@ -111,14 +111,27 @@ def set_avatar_url(db: Session, user: User, *, avatar_url: str) -> User: def soft_delete_account(db: Session, user: User) -> None: - """注销账号:软删除 + 匿名化。 + """注销账号:软删除 + 匿名化 + 释放唯一约束/外部绑定。 - 把 phone 改成 `deleted_` 释放唯一约束,允许同号码重新注册成全新账号; + 把 phone 改成 `deleted_` 释放手机号唯一约束,允许同号码重新注册成全新账号; 清空 PII(昵称/头像),保留行用于审计。status=deleted 后将无法再登录该行 (登录接口校验 status==active)。 + + ⚠️ 注销必须连同释放 user 上其余唯一约束/外部身份,否则 deleted 行永久占着槽位, + 对应资源再也无法被复用: + - wechat_openid: 不清 → 这个微信再也无法被任何账号绑定(提现通道彻底锁死); + - invite_code: 不清 → 邀请码槽位被占 + 裸查仍命中死号(发奖已被 bind 的 + status==active 校验兜住,清掉更干净)。 + 资金安全(未提现现金 / 在审提现单)由调用方 delete_account 端点前置校验,这里只匿名化。 """ user.status = "deleted" user.phone = f"deleted_{user.id}" user.nickname = None user.avatar_url = None + # 等价"解绑微信":释放 openid 唯一槽 + 清微信昵称头像 + user.wechat_openid = None + user.wechat_nickname = None + user.wechat_avatar_url = None + # 释放邀请码唯一槽 + user.invite_code = None db.commit() diff --git a/app/repositories/wallet.py b/app/repositories/wallet.py index 615f0eb..a38ac88 100644 --- a/app/repositories/wallet.py +++ b/app/repositories/wallet.py @@ -340,6 +340,15 @@ def bind_wechat_openid(db: Session, user_id: int, code: str) -> dict: return info +def get_cash_balance_cents(db: Session, user_id: int) -> int: + """只读查现金余额(分)。账户不存在返回 0,**不创建账户**——注销前置校验用, + 不该给一个即将被删除的用户凭空建一行空账户(get_or_create_account 会建)。""" + val = db.execute( + select(CoinAccount.cash_balance_cents).where(CoinAccount.user_id == user_id) + ).scalar_one_or_none() + return val or 0 + + def has_reviewing_withdraw(db: Session, user_id: int) -> bool: """用户是否有「待审核(reviewing)」的提现单。 diff --git a/tests/test_delete_account.py b/tests/test_delete_account.py new file mode 100644 index 0000000..464b02c --- /dev/null +++ b/tests/test_delete_account.py @@ -0,0 +1,141 @@ +"""注销账号(DELETE /api/v1/user):软删 + 释放唯一约束 + 资金前置校验。 + +覆盖: + - 干净注销:status=deleted + phone 匿名化 + wechat_openid/invite_code 等唯一槽全部释放 + - 回归原始 bug:注销释放 openid 后,别的账号能绑同一个微信 + - 资金前置闸:有可提现现金 / 在审提现单 → 409 拒绝注销,账号不被删 + - 注销后旧 token 即时失效(status==active 闸) +""" +from __future__ import annotations + +from sqlalchemy import select + +from app.db.session import SessionLocal +from app.models.user import User +from app.models.wallet import CoinAccount + + +def _login(client, phone: str) -> str: + client.post("/api/v1/auth/sms/send", json={"phone": phone}) + r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"}) + assert r.status_code == 200, r.text + return r.json()["access_token"] + + +def _auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +def _patch_userinfo(monkeypatch, openid: str) -> None: + monkeypatch.setattr( + "app.integrations.wxpay.code_to_userinfo", + lambda code: {"openid": openid, "nickname": "微信昵称", "avatar_url": "https://x/a.png", "raw": {}}, + ) + + +def _seed_cash(client, token: str, phone: str, cents: int) -> None: + """先访问 /account 触发建账户,再用 DB 直接灌现金余额。""" + client.get("/api/v1/wallet/account", headers=_auth(token)) + db = SessionLocal() + try: + user = db.execute(select(User).where(User.phone == phone)).scalar_one() + acc = db.get(CoinAccount, user.id) + acc.cash_balance_cents = cents + db.commit() + finally: + db.close() + + +def _get_user(phone: str) -> User | None: + db = SessionLocal() + try: + return db.execute(select(User).where(User.phone == phone)).scalar_one_or_none() + finally: + db.close() + + +def test_delete_clean_account_releases_all_unique_slots(client) -> None: + phone = "13900000001" + token = _login(client, phone) + # 预置:给该用户造昵称 + 邀请码(占 invite_code 唯一槽),验证注销时一并释放 + db = SessionLocal() + try: + u = db.execute(select(User).where(User.phone == phone)).scalar_one() + u.nickname = "张三" + u.invite_code = "INVCODE0001" + db.commit() + uid = u.id + finally: + db.close() + + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 200, r.text + + u = _get_user(f"deleted_{uid}") + assert u is not None + assert u.status == "deleted" + assert u.phone == f"deleted_{uid}" + assert u.nickname is None + assert u.avatar_url is None + assert u.wechat_openid is None + assert u.wechat_nickname is None + assert u.wechat_avatar_url is None + assert u.invite_code is None + + +def test_delete_releases_wechat_so_another_account_can_bind(client, monkeypatch) -> None: + """原始 bug 回归:A 绑微信 → 注销,B 必须能绑同一个 openid(修复前会 409)。""" + openid = "openid_delreuse_a1" # 独特值,避免跨测试共享库污染(如 test_withdraw 也用 openid_shared_xyz) + # A 绑微信 + _patch_userinfo(monkeypatch, openid) + token_a = _login(client, "13900000002") + r = client.post("/api/v1/wallet/bind-wechat", json={"code": "ca"}, headers=_auth(token_a)) + assert r.status_code == 200, r.text + # A 注销(无现金、无提现单 → 允许) + r = client.delete("/api/v1/user", headers=_auth(token_a)) + assert r.status_code == 200, r.text + # B 绑同一个 openid → 必须成功(修复前会 409 "该微信已绑定其他账号") + token_b = _login(client, "13900000003") + r = client.post("/api/v1/wallet/bind-wechat", json={"code": "cb"}, headers=_auth(token_b)) + assert r.status_code == 200, r.text + assert r.json()["bound"] is True + + +def test_delete_blocked_when_cash_balance_positive(client) -> None: + phone = "13900000004" + token = _login(client, phone) + _seed_cash(client, token, phone, 100) # 1 元未提现 + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 409, r.text + assert "现金" in r.json()["detail"] + u = _get_user(phone) # 账号未被删 + assert u is not None and u.status == "active" + + +def test_delete_blocked_when_reviewing_withdraw_exists(client, monkeypatch) -> None: + """cash 提光(=0)但有在审提现单 → 仍拒绝(命中 reviewing 闸,非 cash 闸)。""" + phone = "13900000005" + _patch_userinfo(monkeypatch, "openid_delrev_b2") + token = _login(client, phone) + _seed_cash(client, token, phone, 50) + client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token)) + # 提光 50 → cash 归 0,单进 reviewing + r = client.post("/api/v1/wallet/withdraw", json={"amount_cents": 50}, headers=_auth(token)) + assert r.status_code == 200 and r.json()["status"] == "reviewing", r.text + assert r.json()["cash_balance_cents"] == 0 + + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 409, r.text + assert "审核" in r.json()["detail"] + u = _get_user(phone) + assert u is not None and u.status == "active" + + +def test_deleted_user_old_token_is_rejected(client) -> None: + phone = "13900000006" + token = _login(client, phone) + r = client.delete("/api/v1/user", headers=_auth(token)) + assert r.status_code == 200, r.text + # 注销后旧 token 立即失效(get_current_user 校验 status==active) + r = client.get("/api/v1/user/onboarding/status", params={"device_id": "d"}, headers=_auth(token)) + assert r.status_code == 401, r.text From f7a3ef2e0bd2f8bc80377732761d4a6eca78d032 Mon Sep 17 00:00:00 2001 From: marco Date: Sat, 20 Jun 2026 13:39:45 +0800 Subject: [PATCH 7/7] =?UTF-8?q?feat(internal):=20=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E7=A1=AE=E8=AE=A4=E7=AA=97=20agent=20=E5=85=9C=E5=BA=95?= =?UTF-8?q?=E6=A0=B7=E6=9C=AC=E8=90=BD=E7=9B=98(pricebot=20=E5=9B=9E?= =?UTF-8?q?=E5=86=99)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增内部端点 POST /internal/launch-confirm-sample(X-Internal-Secret 鉴权): pricebot 对没见过文案的启动确认窗用 LLM 兜底放行后,把样本(host 包 / 弹窗树 / LLM plan / 设备 locale 机型)回写落 launch_confirm_sample 表,供研发人工沉淀回 pricebot 规则 yaml - 配套: model LaunchConfirmSample + schema + repo(都上报不去重) + alembic 迁移 launch_confirm_sample_table(down_revision=cps_wx_user) + main 注册 router + models/__init__ 登记 - docs: 后端技术实现.md 加 §6.5 pricebot 回写内部端点说明 Co-Authored-By: Claude Opus 4.8 --- .../versions/launch_confirm_sample_table.py | 53 ++++++++++++++++ app/api/internal/launch_confirm.py | 63 +++++++++++++++++++ app/main.py | 2 + app/models/__init__.py | 1 + app/models/launch_confirm_sample.py | 61 ++++++++++++++++++ app/repositories/launch_confirm_sample.py | 35 +++++++++++ app/schemas/launch_confirm_sample.py | 26 ++++++++ docs/后端技术实现.md | 10 +++ 8 files changed, 251 insertions(+) create mode 100644 alembic/versions/launch_confirm_sample_table.py create mode 100644 app/api/internal/launch_confirm.py create mode 100644 app/models/launch_confirm_sample.py create mode 100644 app/repositories/launch_confirm_sample.py create mode 100644 app/schemas/launch_confirm_sample.py diff --git a/alembic/versions/launch_confirm_sample_table.py b/alembic/versions/launch_confirm_sample_table.py new file mode 100644 index 0000000..d9ff6fe --- /dev/null +++ b/alembic/versions/launch_confirm_sample_table.py @@ -0,0 +1,53 @@ +"""launch_confirm_sample table (启动确认窗 agent 兜底样本: LLM 放行的弹窗样本沉淀) + +Revision ID: launch_confirm_sample_table +Revises: cps_wx_user +Create Date: 2026-06-20 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "launch_confirm_sample_table" +down_revision: Union[str, Sequence[str], None] = "cps_wx_user" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "launch_confirm_sample", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False), + sa.Column("trace_id", sa.String(length=64), nullable=True), + sa.Column("device_id", sa.String(length=64), nullable=True), + sa.Column("host_package", sa.String(length=128), nullable=True), + sa.Column("target_app", sa.String(length=128), nullable=True), + sa.Column("system_locale", sa.String(length=32), nullable=True), + sa.Column("exec_success", sa.Boolean(), nullable=False), + sa.Column("dialog_title", sa.String(length=256), nullable=True), + # PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐 + sa.Column("payload", sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), "postgresql"), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("launch_confirm_sample", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_created_at"), ["created_at"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_trace_id"), ["trace_id"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_device_id"), ["device_id"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_host_package"), ["host_package"], unique=False) + batch_op.create_index(batch_op.f("ix_launch_confirm_sample_system_locale"), ["system_locale"], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table("launch_confirm_sample", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_system_locale")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_host_package")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_device_id")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_trace_id")) + batch_op.drop_index(batch_op.f("ix_launch_confirm_sample_created_at")) + + op.drop_table("launch_confirm_sample") diff --git a/app/api/internal/launch_confirm.py b/app/api/internal/launch_confirm.py new file mode 100644 index 0000000..5f24d3f --- /dev/null +++ b/app/api/internal/launch_confirm.py @@ -0,0 +1,63 @@ +"""启动确认窗兜底样本内部上报端点(pricebot → app-server)。 + +pricebot 的 launch_confirm_agent 每次靠 LLM 兜底放行一个"静态 PROFILES 没认出"的跨 App +启动确认窗时,把完整样本 POST 到这里落库。**不是给客户端的接口**:不走用户 JWT,靠 +server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。密钥未配置 +(默认空)时直接 503,避免裸奔写端点。 + +研发定期人工把 exec_success=true 的样本沉淀进 pricebot 的 launch_confirm.PROFILES。 +""" +from __future__ import annotations + +import hmac +import logging +from typing import Annotated + +from fastapi import APIRouter, Header, HTTPException, status + +from app.api.deps import DbSession +from app.core.config import settings +from app.repositories import launch_confirm_sample as repo +from app.schemas.launch_confirm_sample import ( + LaunchConfirmSampleIn, + LaunchConfirmSampleOut, +) + +logger = logging.getLogger("shagua.internal.launch_confirm") + +router = APIRouter(prefix="/internal", tags=["internal"]) + + +def _check_secret(x_internal_secret: str | None) -> None: + """共享密钥校验。未配置 → 503(挡裸奔写端点);不匹配 → 401(常量时间比较)。""" + configured = settings.INTERNAL_API_SECRET + if not configured: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="internal api not configured", + ) + if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="invalid internal secret", + ) + + +@router.post( + "/launch-confirm-sample", + response_model=LaunchConfirmSampleOut, + summary="启动确认窗兜底样本上报(pricebot→app-server,落 launch_confirm_sample)", +) +def report_launch_confirm_sample( + payload: LaunchConfirmSampleIn, + db: DbSession, + x_internal_secret: Annotated[str | None, Header()] = None, +) -> LaunchConfirmSampleOut: + _check_secret(x_internal_secret) + sid = repo.insert_sample(db, payload) + logger.info( + "launch_confirm_sample id=%d host=%s locale=%s success=%s trace=%s", + sid, payload.host_package, payload.system_locale, + payload.exec_success, payload.trace_id, + ) + return LaunchConfirmSampleOut(id=sid) diff --git a/app/main.py b/app/main.py index 702f19b..17ff332 100644 --- a/app/main.py +++ b/app/main.py @@ -22,6 +22,7 @@ from app.api.v1.compare_record import router as compare_record_router from app.api.v1.coupon import router as coupon_router from app.api.v1.cps_redirect import router as cps_redirect_router from app.api.internal.app_version import router as internal_app_version_router +from app.api.internal.launch_confirm import router as internal_launch_confirm_router from app.api.internal.price import router as internal_price_router from app.api.internal.store import router as internal_store_router from app.api.v1.feedback import router as feedback_router @@ -109,6 +110,7 @@ app.include_router(report_router) app.include_router(internal_price_router) app.include_router(internal_store_router) app.include_router(internal_app_version_router) +app.include_router(internal_launch_confirm_router) app.include_router(platform_router) # CPS 群发短链跳转 /c/{code}(公网无鉴权:记点击 → 302 跳美团) app.include_router(cps_redirect_router) diff --git a/app/models/__init__.py b/app/models/__init__.py index 768d192..d88882f 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -20,6 +20,7 @@ from app.models.coupon_state import ( # noqa: F401 from app.models.feedback import Feedback # noqa: F401 from app.models.invite import InviteRelation # noqa: F401 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.ops_marquee_seed import OpsMarqueeSeed # noqa: F401 diff --git a/app/models/launch_confirm_sample.py b/app/models/launch_confirm_sample.py new file mode 100644 index 0000000..de225b0 --- /dev/null +++ b/app/models/launch_confirm_sample.py @@ -0,0 +1,61 @@ +"""启动确认窗 agent 兜底样本表(launch_confirm_sample)。 + +pricebot 的 launch_confirm_agent 每次"静态 PROFILES 没认出、靠 LLM 兜底放行"一个跨 App +启动确认窗(繁体 / 英文 / 小语种 / ROM 改版)时,把完整样本 server→server 落这里。研发定期 +**人工**把 exec_success=true 的样本沉淀进 pricebot 的 launch_confirm.PROFILES(静态快路径), +让以后同款窗不必再调 LLM。 + +设计: +- 与登录无关、不鉴权(server→server 共享密钥头 X-Internal-Secret)。 +- **都上报、不去重**(by design):同一种窗多台机器上报多条,研发按 host_package / locale + 聚合后人工筛。 +- 大字段(弹窗树快照 + LLM plan + 设备信息)塞 payload(JSONB)兜底,免得加字段就迁移 schema。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import JSON, Boolean, DateTime, Integer, String, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.base import Base + +# PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON(同 price_observation)。 +_JSON = JSON().with_variant(JSONB(), "postgresql") + + +class LaunchConfirmSample(Base): + __tablename__ = "launch_confirm_sample" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), index=True, nullable=False + ) + + # pricebot 侧 trace_id:回指原始 trace,能去 price.shaguabijia.com 查完整上下文。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + device_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) + + # 弹窗宿主包(= 触发判据, 也是研发沉淀进 PROFILES.host_packages 的键)。按它聚合人工筛。 + host_package: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True) + # 被打开的目标 App 包名(标题里"想要打开 X"的 X 是变量, 据此把它从 sign 里挖空)。 + target_app: Mapped[str | None] = mapped_column(String(128), nullable=True) + # 系统语言地区(zh-TW / en-US…)—— 多语言问题的核心维度。 + # ⚠️ 前端 AgentStepRequest 暂未上报设备信息 → 现恒 None,待前端加 device_info 字段后回填。 + system_locale: Mapped[str | None] = mapped_column(String(32), index=True, nullable=True) + + # 这次兜底是否成功放行(弹窗消失=True;放弃 / LLM 没认出=False)。研发只沉淀 True 的。 + exec_success: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + # 弹窗标题全文(研发据此提 launch_sign),如"「傻瓜比价」想要開啟「京東」"。 + dialog_title: Mapped[str | None] = mapped_column(String(256), nullable=True) + + # 大 JSON:完整样本。{is_launch_confirm, llm_model, llm_reason, plan, dialog_tree, + # device_info: {screen, locale, brand, model, android_version, rom_version}} + payload: Mapped[dict | None] = mapped_column(_JSON, nullable=True) + + def __repr__(self) -> str: # pragma: no cover + return ( + f"" + ) diff --git a/app/repositories/launch_confirm_sample.py b/app/repositories/launch_confirm_sample.py new file mode 100644 index 0000000..4d0b226 --- /dev/null +++ b/app/repositories/launch_confirm_sample.py @@ -0,0 +1,35 @@ +"""启动确认窗兜底样本落库:单条 insert(都上报、不去重,by design)。""" +from __future__ import annotations + +import logging +from typing import Optional + +from sqlalchemy.orm import Session + +from app.models.launch_confirm_sample import LaunchConfirmSample +from app.schemas.launch_confirm_sample import LaunchConfirmSampleIn + +logger = logging.getLogger("shagua.launch_confirm") + + +def _trunc(s: Optional[str], n: int) -> Optional[str]: + """截断到列长上限(PG varchar 超长会报错;繁体长标题等需要)。空 → None。""" + return s[:n] if s else None + + +def insert_sample(db: Session, payload: LaunchConfirmSampleIn) -> int: + """写入一条样本,返回新行 id。""" + row = LaunchConfirmSample( + trace_id=_trunc(payload.trace_id, 64), + device_id=_trunc(payload.device_id, 64), + host_package=_trunc(payload.host_package, 128), + target_app=_trunc(payload.target_app, 128), + system_locale=_trunc(payload.system_locale, 32), + exec_success=bool(payload.exec_success), + dialog_title=_trunc(payload.dialog_title, 256), + payload=payload.payload, + ) + db.add(row) + db.commit() + db.refresh(row) + return row.id diff --git a/app/schemas/launch_confirm_sample.py b/app/schemas/launch_confirm_sample.py new file mode 100644 index 0000000..e2ed223 --- /dev/null +++ b/app/schemas/launch_confirm_sample.py @@ -0,0 +1,26 @@ +"""启动确认窗兜底样本的内部上报模型(pricebot → app-server)。""" +from __future__ import annotations + +from pydantic import BaseModel + + +class LaunchConfirmSampleIn(BaseModel): + """一次 agent 兜底的完整样本。 + + 字段一律宽松可空:落盘是辅助数据,绝不能因个别字段缺失就拒收(否则连累 pricebot 兜底)。 + """ + + trace_id: str | None = None + device_id: str | None = None + host_package: str | None = None + target_app: str | None = None + system_locale: str | None = None + exec_success: bool = False + dialog_title: str | None = None + payload: dict | None = None + + +class LaunchConfirmSampleOut(BaseModel): + """落库结果。""" + + id: int diff --git a/docs/后端技术实现.md b/docs/后端技术实现.md index 8828ed7..e106985 100644 --- a/docs/后端技术实现.md +++ b/docs/后端技术实现.md @@ -222,6 +222,16 @@ POST /api/v1/auth/sms/login { phone, code } → 任意 6 位通过 → upsert --- +## 6.5 pricebot 回写:内部端点(server→server) + +§6 是 app-server 透传给 pricebot;反过来 pricebot 也会把少量数据回写 app-server 落库,走 `app/api/internal/*` —— **不走用户 JWT**,靠共享密钥头 `X-Internal-Secret`(== `INTERNAL_API_SECRET`,未配则 503,防裸奔写端点)校验。 + +**启动确认窗兜底样本**(2026-06):国产 ROM 打开别的 App 时弹"想要打开 XX"确认窗,pricebot 对没见过文案(繁体/英文/ROM 改版)的窗用 LLM 兜底放行后,把样本 POST 到 [`/internal/launch-confirm-sample`](../app/api/internal/launch_confirm.py) 落 `launch_confirm_sample` 表(host 包 + 弹窗树 + LLM plan + 设备 locale/机型),供研发定期人工沉淀回 pricebot 的规则 yaml(回到快路径)。**需 pricebot 与 app-server 两边 `.env` 配同一 `INTERNAL_API_SECRET` 才生效**(未配则 pricebot 侧跳过上报、不影响比价/领券)。 + +> 同类内部回写端点还有 `/internal/price-observation`(价格观测)、`/internal/store-mapping`(跨平台店铺映射)等 pricebot 比价资产沉淀。 + +--- + ## 7. 数据模型 业务表(下表)+ `alembic_version` 框架表。生产 PG / 开发可回退 SQLite。