Compare commits

...

15 Commits

Author SHA1 Message Date
陈世睿 0bd42ea325 fix(compare): platform_results 接受 dict, 修成功比价记录被 422 拒
客户端 fromComparison 把 done.params.platform_results(按 platform_id 键的 dict)
原样上报, 但 schema 误声明为 list → pydantic 422 → 成功记录从不落库, 记录页只剩
失败/终止 (2026-06-17 加该字段上报后复现)。改 dict | list 宽松存(仅进 raw_payload)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-20 22:16:23 +08:00
陈世睿 0028c07c7b fix(alembic): 合并 device 与 launch_confirm_sample 双 head(分支合并产物)
feat/a11y-liveness-heartbeat 合进 feat/compare-record-status 后,device 链(3a8fa058789c)与 main 来的 launch_confirm_sample_table 成两 head;加合并节点 86852351f5bb 收敛单 head。全链路在临时 sqlite upgrade head 通过。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-20 14:49:46 +08:00
陈世睿 dd8375f822 Merge branch 'feat/a11y-liveness-heartbeat' into feat/compare-record-status 2026-06-20 14:47:54 +08:00
陈世睿 713a0a2afe Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/a11y-liveness-heartbeat 2026-06-20 14:26:45 +08:00
陈世睿 150aa4f81a Merge origin/main into feat/a11y-liveness-heartbeat(并解决 alembic 双 head)
合并 main 引入 cps_wx_user / 穿山甲配置后台化 / 注销账号校验 / 文档重构 等。
合并产生 alembic 双 head(main 的 cps_wx_user + 本分支 device_kill_alert_pending)→ 新增合并节点 3a8fa058789c 收敛为单 head, alembic upgrade head 恢复正常(修 run.sh 的 Multiple head revisions)。
顺手修 heartbeat_monitor_worker 掉线告警日志 🔴 emoji 在 Windows GBK 控制台 UnicodeEncodeError(非致命刷屏)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-20 10:37:46 +08:00
陈世睿 4d1162053e feat(device): 无障碍掉线后置检测 — kill_alert_pending + liveness 查询/ack 端点
推送暂缓改 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) <noreply@anthropic.com>
2026-06-19 22:35:12 +08:00
陈世睿 b949d3b662 Merge remote-tracking branch 'origin/main' into feat/compare-record-status 2026-06-19 10:18:10 +08:00
陈世睿 060cb74a84 fix(alembic): cps_v2_platforms 兼容本地 SQLite
队友这条迁移只在 PG 测过:server_default 用了 PG 专属 ::jsonb(SQLite 报 unrecognized token),
且 op.alter_column 改可空/类型 SQLite 原生不支持。改成:platforms default 分方言(PG 仍 ::jsonb、
SQLite 用纯 JSON 字面量)+ 所有 alter_column 包进 op.batch_alter_table(SQLite 重建表、PG 等价直接 ALTER)。
线上 PG 行为不变,本地 SQLite dev 能正常 upgrade。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-19 10:18:10 +08:00
陈世睿 caee1c355d Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/compare-record-status 2026-06-18 21:42:48 +08:00
陈世睿 9409ee6e05 Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/compare-record-status 2026-06-18 10:19:46 +08:00
陈世睿 020dd143c7 Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/compare-record-status 2026-06-17 20:36:30 +08:00
陈世睿 a59b49fdc9 比价结果状态 2026-06-17 14:56:42 +08:00
陈世睿 07a8710d5a feat(compare-record): 逐平台 status 落库,记录页显示异常文案
ComparisonResultIn 增 status / store_closed 两字段(原本 pydantic 静默丢弃)。
客户端上报前把 done.params 的 comparison_results + platform_results.status
合并写入,记录页据此对无价平台显示「未满起送价/门店打烊/店家未入驻/比价失败」,
而非笼统的「店家未入驻」。纯字段透传,best/saved/status 派生逻辑不变(仍按 price 过滤)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-17 14:41:56 +08:00
陈世睿 b3b731f25d Merge origin/main into feat/a11y-liveness-heartbeat
冲突全在领券「弹窗频控按 App 独立」这一功能——本分支夹带的旧 WIP 草稿(2026-06-12,
app_package 可空+NULL 兜底行)与 main 上已合并的最终版 #52(2026-06-14,package NOT NULL
+"" 占位)是同一功能的两套不兼容实现。

解决:四个 coupon 文件(api/models/repositories/schemas)一律取 main(#52)——它是
权威已合版,且文件未冲突区已自动并入 #52 的签名,只有取 theirs 才自洽。

Alembic 收敛为单 head:
- 删本分支 WIP 迁移 coupon_engagement_app_package(否则与 #52 重复建 per-App 列)
  及其仅用于挂接它的两个空 merge 节点 3a9941e76909 / f3d0a16bb4c2
- device_liveness_table 的 down_revision 由 f3d0a16bb4c2 改挂 main 头 store_mapping_tb_dl_invalid

验证:alembic heads 单头;全链在临时 sqlite upgrade head 通过;
coupon_prompt_engagement 只有 package、无 app_package;app 全量 import 通过。

注:本分支不再保留他人领券 WIP——无损,#52 即该功能的最终版,已在 main。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-16 10:12:49 +08:00
陈世睿 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
17 changed files with 875 additions and 27 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 cps_wx_user 与 device_kill_alert_pending 双 head
Revision ID: 3a8fa058789c
Revises: cps_wx_user, device_kill_alert_pending
Create Date: 2026-06-20 10:27:01.193201
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '3a8fa058789c'
down_revision: Union[str, Sequence[str], None] = ('cps_wx_user', 'device_kill_alert_pending')
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,26 @@
"""merge device 与 launch_confirm_sample 双 head(分支合并产物)
Revision ID: 86852351f5bb
Revises: 3a8fa058789c, launch_confirm_sample_table
Create Date: 2026-06-20 14:48:51.906553
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '86852351f5bb'
down_revision: Union[str, Sequence[str], None] = ('3a8fa058789c', 'launch_confirm_sample_table')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
+38 -24
View File
@@ -23,39 +23,53 @@ branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
# ⚠️ SQLite 不支持 ALTER COLUMN(改可空/类型),必须用 op.batch_alter_table(重建表)模拟;
# env.py 的 render_as_batch 只管 autogenerate 渲染、不会让"手写的 op.alter_column"在运行时自动 batch。
# batch_alter_table 在 Postgres 上等价于直接 ALTER(recreate=auto 不重建),两端通用。
def upgrade() -> None:
# 活动:淘宝淘口令 / 京东链接
# 活动:淘宝淘口令 / 京东链接(纯 add column,SQLite 直接支持)
op.add_column("cps_activity", sa.Column("payload", sa.Text(), nullable=True))
# 群:多平台(现有群都是美团,server_default 回填)+ sid 可空
op.add_column(
"cps_group",
sa.Column(
"platforms",
sa.JSON().with_variant(postgresql.JSONB(), "postgresql"),
nullable=False,
server_default=sa.text("'[\"meituan\"]'::jsonb"),
),
# 群:多平台(现有群都是美团,server_default 回填)+ sid 可空
# server_default 分方言:Postgres 用 ::jsonb 显式转;SQLite 不认 ::jsonb(报 unrecognized token ":"),
# 用纯 JSON 字符串字面量即可(JSON 列在 SQLite 实际是 TEXT)。
platforms_default = (
"'[\"meituan\"]'::jsonb"
if op.get_bind().dialect.name == "postgresql"
else "'[\"meituan\"]'"
)
op.alter_column("cps_group", "sid", existing_type=sa.String(64), nullable=True)
with op.batch_alter_table("cps_group") as batch_op:
batch_op.add_column(
sa.Column(
"platforms",
sa.JSON().with_variant(postgresql.JSONB(), "postgresql"),
nullable=False,
server_default=sa.text(platforms_default),
)
)
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=True)
# link:sid 可空 + target 加长
op.alter_column("cps_link", "sid", existing_type=sa.String(64), nullable=True)
op.alter_column("cps_link", "target_url", existing_type=sa.String(1024), type_=sa.String(2048))
with op.batch_alter_table("cps_link") as batch_op:
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=True)
batch_op.alter_column("target_url", existing_type=sa.String(1024), type_=sa.String(2048))
# click:sid 可空 + 事件类型
op.alter_column("cps_click", "sid", existing_type=sa.String(64), nullable=True)
op.add_column(
"cps_click",
sa.Column("event_type", sa.String(16), nullable=False, server_default="visit"),
)
with op.batch_alter_table("cps_click") as batch_op:
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=True)
batch_op.add_column(
sa.Column("event_type", sa.String(16), nullable=False, server_default="visit")
)
def downgrade() -> None:
op.drop_column("cps_click", "event_type")
op.alter_column("cps_click", "sid", existing_type=sa.String(64), nullable=False)
op.alter_column("cps_link", "target_url", existing_type=sa.String(2048), type_=sa.String(1024))
op.alter_column("cps_link", "sid", existing_type=sa.String(64), nullable=False)
op.drop_column("cps_group", "platforms")
op.alter_column("cps_group", "sid", existing_type=sa.String(64), nullable=False)
with op.batch_alter_table("cps_click") as batch_op:
batch_op.drop_column("event_type")
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=False)
with op.batch_alter_table("cps_link") as batch_op:
batch_op.alter_column("target_url", existing_type=sa.String(2048), type_=sa.String(1024))
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=False)
with op.batch_alter_table("cps_group") as batch_op:
batch_op.alter_column("sid", existing_type=sa.String(64), nullable=False)
batch_op.drop_column("platforms")
op.drop_column("cps_activity", "payload")
@@ -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')
+53
View File
@@ -0,0 +1,53 @@
"""device table (无障碍保护存活检测 + 极光推送)
Revision ID: device_liveness_table
Revises: store_mapping_tb_dl_invalid
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] = 'store_mapping_tb_dl_invalid'
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,31 @@
"""merge cps_link_tables and device_liveness_table heads
Revision ID: merge_cps_device_liveness
Revises: cps_link_tables, device_liveness_table
Create Date: 2026-06-17
并行分支双 head 合并(团队常态:不同 feature 分支各自基于同一父 revision 加迁移,
集成时就出现多 head,`alembic upgrade head` 报 Multiple head revisions):
- cps_link_tables —— main 的 CPS 链路表
- device_liveness_table —— feat/a11y-liveness-heartbeat 的设备存活心跳表
两条链彼此独立、无 schema 冲突,这里合并成单 head。纯合并节点,不改表结构。
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'merge_cps_device_liveness'
down_revision: Union[str, Sequence[str], None] = ('cps_link_tables', 'device_liveness_table')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
+96
View File
@@ -0,0 +1,96 @@
"""设备注册 / 心跳 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,
LivenessAckRequest,
LivenessOut,
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()
@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()
+26
View File
@@ -66,6 +66,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 已被杀/无障碍已停。【已置 kill_alert_pending: 用户下次进 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.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
@@ -38,6 +39,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,
@@ -59,9 +64,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")
@@ -94,6 +101,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
@@ -12,6 +12,7 @@ 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.device import Device # noqa: F401
from app.models.coupon_state import ( # noqa: F401
CouponClaimRecord,
CouponDailyCompletion,
+89
View File
@@ -0,0 +1,89 @@
"""设备表(无障碍保护存活检测 + 极光推送)。
每条 = 一个用户的一台设备(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
)
# 掉线告警「待客户端提醒」标记(后置检测 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
)
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}>"
)
+124
View File
@@ -0,0 +1,124 @@
"""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,避免每轮重复处理)+ 置「待客户端提醒」标记。
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()
+13 -2
View File
@@ -51,6 +51,13 @@ class ComparisonResultIn(BaseModel):
# 跟 coupon_saved 并存, 是更丰富的明细; 空=没抠到 → 前端回退单券路径。
# 必须显式声明: 落库走 model_dump(), pydantic 默认丢未知字段, 不声明这行会被悄悄吞掉。
applied_coupons: list[AppliedCouponIn] = Field(default_factory=list)
# 逐平台状态(客户端上报前把 done.params 的 comparison_results + platform_results.status 合并写入):
# success / below_minimum / store_closed / store_not_found / items_not_found / unsupported / failed。
# 「比价记录」页据此对无价平台显示对应文案(未满起送价/门店打烊/店家未入驻/比价失败)。
# 必须显式声明: 落库走 model_dump(), 不声明会被 pydantic 静默丢弃 → 记录页拿不到 status。
status: str | None = None
# 门店打烊原因(price 为 None 时带): 与 status="store_closed" 等价的更早信号, 一并落库供前端兜底判打烊。
store_closed: str | None = None
class ComparisonRecordIn(BaseModel):
@@ -71,8 +78,12 @@ class ComparisonRecordIn(BaseModel):
comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
# 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
# unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
# admin「卡在哪一步」从这里读。list[dict] 宽松存(结构由 pricebot 定,只作 debug 展示)。
platform_results: list = Field(default_factory=list)
# admin「卡在哪一步」从这里读。
# ⚠️ pricebot done.params.platform_results 是 **dict(按 platform_id 键)**, 不是 list ——
# 客户端 fromComparison 原样 optJSONObject 透传上来 (见 android Protocol.kt)。早先误声明成
# list 导致每条成功记录 422 被拒、记录页只剩失败/终止 (2026-06-17 引入 platform_results 上报后复现)。
# 结构由 pricebot 定、仅作 debug 展示, 宽松存 dict; 兼容老的 list 形态以防回退。
platform_results: dict | list = Field(default_factory=dict)
skipped_dish_count: int | None = None
skipped_dish_names: list[str] = Field(default_factory=list)
total_dish_count: int | None = None
+50
View File
@@ -0,0 +1,50 @@
"""设备注册 / 心跳相关 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
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