diff --git a/.env.example b/.env.example index 1f53db8..367226a 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/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py b/alembic/versions/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py new file mode 100644 index 0000000..7d25e70 --- /dev/null +++ b/alembic/versions/3a8fa058789c_merge_cps_wx_user_与_device_kill_alert_.py @@ -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 diff --git a/alembic/versions/86852351f5bb_merge_device_与_launch_confirm_sample_双_.py b/alembic/versions/86852351f5bb_merge_device_与_launch_confirm_sample_双_.py new file mode 100644 index 0000000..4b8c799 --- /dev/null +++ b/alembic/versions/86852351f5bb_merge_device_与_launch_confirm_sample_双_.py @@ -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 diff --git a/alembic/versions/cps_v2_platforms.py b/alembic/versions/cps_v2_platforms.py index e80c1a6..8311a58 100644 --- a/alembic/versions/cps_v2_platforms.py +++ b/alembic/versions/cps_v2_platforms.py @@ -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") 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/alembic/versions/device_table.py b/alembic/versions/device_table.py new file mode 100644 index 0000000..795864c --- /dev/null +++ b/alembic/versions/device_table.py @@ -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') diff --git a/alembic/versions/merge_cps_link_and_device_liveness_heads.py b/alembic/versions/merge_cps_link_and_device_liveness_heads.py new file mode 100644 index 0000000..1346b50 --- /dev/null +++ b/alembic/versions/merge_cps_link_and_device_liveness_heads.py @@ -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 diff --git a/app/api/v1/device.py b/app/api/v1/device.py new file mode 100644 index 0000000..56a5a85 --- /dev/null +++ b/app/api/v1/device.py @@ -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() diff --git a/app/core/config.py b/app/core/config.py index 76621e1..900d3e8 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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 diff --git a/app/core/heartbeat_monitor_worker.py b/app/core/heartbeat_monitor_worker.py new file mode 100644 index 0000000..6072d26 --- /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 已被杀/无障碍已停。【已置 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 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 17ff332..f4fde37 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.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) diff --git a/app/models/__init__.py b/app/models/__init__.py index d88882f..f3491f1 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -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, diff --git a/app/models/device.py b/app/models/device.py new file mode 100644 index 0000000..290bc9f --- /dev/null +++ b/app/models/device.py @@ -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"" + ) diff --git a/app/repositories/device.py b/app/repositories/device.py new file mode 100644 index 0000000..b1c40d8 --- /dev/null +++ b/app/repositories/device.py @@ -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() diff --git a/app/schemas/compare_record.py b/app/schemas/compare_record.py index 5f7306c..d4fae31 100644 --- a/app/schemas/compare_record.py +++ b/app/schemas/compare_record.py @@ -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 diff --git a/app/schemas/device.py b/app/schemas/device.py new file mode 100644 index 0000000..3254e06 --- /dev/null +++ b/app/schemas/device.py @@ -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