Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8adad30ff2 |
+13
-1
@@ -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")
|
||||
@@ -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
@@ -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}
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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}>"
|
||||
)
|
||||
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user