Compare commits

...

1 Commits

Author SHA1 Message Date
陈世睿 8adad30ff2 feat(device): 无障碍保护存活心跳检测 + 掉线终端告警
- 新增 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) <noreply@anthropic.com>
2026-06-15 22:16:39 +08:00
18 changed files with 863 additions and 34 deletions
+13 -1
View File
@@ -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。
@@ -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
@@ -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")
+53
View File
@@ -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')
@@ -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
+44 -12
View File
@@ -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}
+66
View File
@@ -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()
+26
View File
@@ -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
+148
View File
@@ -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
+94
View File
@@ -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
+8
View File
@@ -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)
+1
View File
@@ -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,
+17 -6
View File
@@ -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(
+83
View File
@@ -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"<Device id={self.id} user_id={self.user_id} "
f"device_id={self.device_id} state={self.liveness_state}>"
)
+49 -13
View File
@@ -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(
+106
View File
@@ -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()
+20 -2
View File
@@ -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
+36
View File
@@ -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