Compare commits

..

1 Commits

Author SHA1 Message Date
unknown afc55ab910 feat(device): 新增设备信息表 device + 上报接口
- 新建 device 设备表(设备/系统信息 + 网络/时区/模拟器/位置 + 关联用户 + 状态/时间戳),
  以客户端 device_id 唯一标识;登录后回填 user_id,游客态可空
- 原 device 模型改名 device_liveness(实为无障碍存活表),腾出 device 名给新表并保持
  表名与实体名一致;同步更新 __init__ / repositories / admin queries / 文档引用
- 新增 POST /api/v1/device/report:软鉴权(带 JWT 绑用户、游客也收),按 device_id upsert;
  设备信息非空才覆盖,位置 latitude/longitude 整字段覆盖含清空(以本次上报实际为准)
- 新增 alembic migration bb47051068c8 建 device 表(裁掉 autogenerate 带出的无关表漂移)
- 补 /device/report 端点与 upsert 单元测试 4 例
2026-07-17 09:46:59 +08:00
15 changed files with 462 additions and 384 deletions
@@ -0,0 +1,59 @@
"""device table(设备档案 / 终端注册)
Revision ID: bb47051068c8
Revises: comparison_llm_cost
Create Date: 2026-07-16 14:22:17.770307
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'bb47051068c8'
down_revision: Union[str, Sequence[str], None] = 'comparison_llm_cost'
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('device_id', sa.String(length=128), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.Column('platform', sa.String(length=16), nullable=False),
sa.Column('oem', sa.String(length=32), nullable=True),
sa.Column('model', sa.String(length=64), nullable=True),
sa.Column('os_version', sa.String(length=32), nullable=True),
sa.Column('app_version', sa.String(length=32), nullable=True),
sa.Column('channel', sa.String(length=32), nullable=True),
sa.Column('screen', sa.String(length=32), nullable=True),
sa.Column('network', sa.String(length=16), nullable=True),
sa.Column('timezone', sa.String(length=64), nullable=True),
sa.Column('is_emulator', sa.Boolean(), nullable=True),
sa.Column('latitude', sa.Float(), nullable=True),
sa.Column('longitude', sa.Float(), nullable=True),
sa.Column('last_ip', sa.String(length=64), nullable=True),
sa.Column('status', sa.String(length=16), nullable=False),
sa.Column('last_active_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'),
)
with op.batch_alter_table('device', schema=None) as batch_op:
batch_op.create_index(batch_op.f('ix_device_device_id'), ['device_id'], unique=True)
batch_op.create_index(batch_op.f('ix_device_last_active_at'), ['last_active_at'], unique=False)
batch_op.create_index(batch_op.f('ix_device_user_id'), ['user_id'], 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_user_id'))
batch_op.drop_index(batch_op.f('ix_device_last_active_at'))
batch_op.drop_index(batch_op.f('ix_device_device_id'))
op.drop_table('device')
+1 -1
View File
@@ -20,7 +20,7 @@ from app.models.admin import AdminAuditLog
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponPromptEngagement
from app.models.device import DeviceLiveness
from app.models.device_liveness import DeviceLiveness
from app.models.feedback import Feedback
from app.models.onboarding import OnboardingCompletion
from app.models.price_report import PriceReport
+51
View File
@@ -0,0 +1,51 @@
"""设备档案上报 endpoint(device 表)。
POST /api/v1/device/report 客户端上报设备当前信息, device_id upsert device
软鉴权:带合法 Bearer 设备关联到该用户;游客态( token)也接受,user_id 暂空
注意与 /api/v1/device/register(app/api/v1/device.py)区分:那个是无障碍存活注册,
device_liveness 硬鉴权;本端点是设备信息档案, device 软鉴权
"""
from __future__ import annotations
import logging
from fastapi import APIRouter, Request
from app.api.deps import DbSession, OptionalUser
from app.repositories import device_profile as device_profile_repo
from app.schemas.device_profile import DeviceReportOut, DeviceReportRequest
logger = logging.getLogger("shagua.device_profile")
router = APIRouter(prefix="/api/v1/device", tags=["device"])
def _client_ip(request: Request) -> str | None:
"""客户端 IP:生产经 nginx 反代优先 X-Forwarded-For 第一段,否则直连 IP。"""
xff = request.headers.get("x-forwarded-for")
if xff:
return xff.split(",")[0].strip()
return request.client.host if request.client else None
@router.post("/report", response_model=DeviceReportOut, summary="上报设备信息(设备档案 upsert)")
def report_device(
req: DeviceReportRequest,
request: Request,
user: OptionalUser,
db: DbSession,
) -> DeviceReportOut:
device = device_profile_repo.upsert_device(
db,
req,
user_id=user.id if user else None,
last_ip=_client_ip(request),
)
logger.info(
"device report device_id=%s user_id=%s has_loc=%s",
req.device_id,
device.user_id,
req.latitude is not None,
)
return DeviceReportOut()
-8
View File
@@ -169,14 +169,6 @@ class Settings(BaseSettings):
# 进程内自动兑换 worker 的检查间隔(秒):每隔这么久醒一次,跨过北京 0 点就跑一轮。
# 默认 600s=10min,即 0 点后最多 10 分钟内兑完(客户端文案已注明「可能存在延迟」)。
AUTO_EXCHANGE_CHECK_INTERVAL_SEC: int = 600
# 连续 N 天无活跃(无比价 / 领券,登录不算)账户金币+现金清零:进程内 worker
# app.core.inactive_clear_worker 跨北京 0 点跑一轮 inactive_clear.clear_inactive_accounts。
# 邀请奖励金物理隔离,不在清空范围。⚠️ 不可逆批量资金操作,口径见 repositories/inactive_clear。
# 运营要临时停在 .env 置 false(worker 不启动)。
INACTIVE_CLEAR_ENABLED: bool = True
# 连续无活跃**满这么多整天之后**才清:15 → 最后活跃在 15 天前当天仍保留,第 16 天起清。
INACTIVE_CLEAR_DAYS: int = 15
INACTIVE_CLEAR_CHECK_INTERVAL_SEC: int = 600
# 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
# 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
# (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
-132
View File
@@ -1,132 +0,0 @@
"""连续 N 天无活跃账户金币 / 现金清零的进程内定时任务。
结构同 app.core.daily_exchange_worker: `INACTIVE_CLEAR_CHECK_INTERVAL_SEC` 醒一次,
跨进北京新的一天(0 点后)就跑一轮 inactive_clear.clear_inactive_accounts
健壮性:
- **天然幂等**:只清有余额的账户,清完余额=0,下一轮 / 重启 / 多次唤醒都不重复清
- **当天首跑即补**:进程起来时若当天还没跑过,立即跑一轮
- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑
- **开关**:settings.INACTIVE_CLEAR_ENABLED=false 时不启动
这是不可逆的批量资金操作(清空用户金币 + 金币现金,邀请金除外)活跃口径与清零
细节见 app.repositories.inactive_clear
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from datetime import date
from pathlib import Path
from sqlalchemy.exc import SQLAlchemyError
from app.core import rewards
from app.core.config import settings
from app.db.session import SessionLocal
from app.repositories import inactive_clear as inactive_clear_repo
logger = logging.getLogger("shagua.inactive_clear")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "inactive_clear.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 _clear_once() -> dict:
with SessionLocal() as db:
return inactive_clear_repo.clear_inactive_accounts(
db, days=settings.INACTIVE_CLEAR_DAYS
)
async def _run_loop() -> None:
interval = max(60, int(settings.INACTIVE_CLEAR_CHECK_INTERVAL_SEC))
lock_stale_after = max(interval * 3, 1800)
with _single_instance_lock(lock_stale_after) as lock_acquired:
if not lock_acquired:
logger.warning("inactive-clear skipped: another worker owns lock")
return
await _run_locked_loop(interval)
async def _run_locked_loop(interval: int) -> None:
logger.info(
"inactive-clear worker started interval=%ss days=%s",
interval,
settings.INACTIVE_CLEAR_DAYS,
)
# 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。
last_run: date | None = None
try:
while True:
try:
_touch_lock()
today = rewards.cn_today()
if last_run != today:
result = await asyncio.to_thread(_clear_once)
last_run = today
logger.info("inactive-clear done date=%s result=%s", today, result)
except SQLAlchemyError:
logger.exception("inactive-clear db error")
except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出
logger.exception("inactive-clear unexpected error")
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("inactive-clear worker stopped")
raise
def start_inactive_clear_worker() -> asyncio.Task | None:
if not settings.INACTIVE_CLEAR_ENABLED:
logger.info("inactive-clear disabled (INACTIVE_CLEAR_ENABLED=false)")
return None
return asyncio.create_task(_run_loop(), name="inactive-clear")
async def stop_inactive_clear_worker(task: asyncio.Task | None) -> None:
if task is None:
return
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
+2 -7
View File
@@ -28,6 +28,7 @@ from app.api.v1.compare_record import router as compare_record_router
from app.api.v1.coupon import router as coupon_router
from app.api.v1.cps_redirect import router as cps_redirect_router
from app.api.v1.device import router as device_router
from app.api.v1.device_profile import router as device_profile_router
from app.api.v1.feedback import router as feedback_router
from app.api.v1.invite import router as invite_router
from app.api.v1.meituan import router as meituan_router
@@ -49,10 +50,6 @@ from app.core.heartbeat_monitor_worker import (
start_heartbeat_monitor,
stop_heartbeat_monitor,
)
from app.core.inactive_clear_worker import (
start_inactive_clear_worker,
stop_inactive_clear_worker,
)
from app.core.logging import setup_logging
from app.core.pricebot_client import aclose_pricebot_client, get_pricebot_client
from app.core.withdraw_reconcile_worker import (
@@ -78,21 +75,18 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
try:
# 预热离线地理库:首次加载 ~2.5M 行 CSV + 建 KDTree,摊到启动、不砸首个按城市过滤的请求
from app.utils import geo
geo.ensure_loaded()
except Exception: # noqa: BLE001
logger.exception("reverse_geocoder 预热失败(城市反查将在首个请求时懒加载)")
reconcile_task = start_withdraw_reconcile_worker()
heartbeat_task = start_heartbeat_monitor()
daily_exchange_task = start_daily_exchange_worker()
inactive_clear_task = start_inactive_clear_worker()
try:
yield
finally:
await stop_heartbeat_monitor(heartbeat_task)
await stop_withdraw_reconcile_worker(reconcile_task)
await stop_daily_exchange_worker(daily_exchange_task)
await stop_inactive_clear_worker(inactive_clear_task)
await aclose_pricebot_client()
logger.info("shutting down")
@@ -127,6 +121,7 @@ app.include_router(analytics_router)
app.include_router(invite_router)
app.include_router(coupon_router)
app.include_router(device_router)
app.include_router(device_profile_router)
app.include_router(compare_router)
app.include_router(compare_record_router)
app.include_router(compare_milestone_router)
+2 -1
View File
@@ -19,7 +19,8 @@ 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 DeviceLiveness # noqa: F401
from app.models.device import Device # noqa: F401
from app.models.device_liveness import DeviceLiveness # noqa: F401
from app.models.coupon_state import ( # noqa: F401
CouponClaimRecord,
CouponDailyCompletion,
+50 -65
View File
@@ -1,83 +1,71 @@
"""设备表(无障碍保护存活检测 + 极光推送)。
"""设备表(设备档案 / 终端注册)。
每条 = 个用户的一台设备(per-install,device_id 由客户端 DeviceId.get() 生成)
客户端的无障碍服务存活时周期上报心跳刷新 last_heartbeat_at;App 前台/登录时上报
registration_id(极光推送目标)后端 heartbeat_monitor_worker 扫描曾经保护过
现在心跳超时的设备,通过极光推送提醒用户重开无障碍
每条 = 台设备,以客户端生成的 device_id 唯一标识(格式 device_<MODEL>_<8hex>,
per-install; pricebot 客户端 PriceBotService.getOrCreateDeviceId)登录前(游客态)
即可建档,登录后回填 user_id 单表只记最近一个登录用户,不保留一机多号历史
liveness_state 状态机(防刷屏,一次掉线只推一条):
unknown alive(收到 service 心跳) silent/notified(扫描发现超时并已推送)
心跳恢复时 handler 重置回 alive
spec: spec/accessibility-liveness-push.md
字段采集来源(逐列见注释):
已在埋点 analytics_event 采集 : oem / os_version / model / app_version / channel / network
服务端补 : last_ip(X-Forwarded-For)last_active_at
客户端需新增上报(无需权限) : timezoneis_emulator
客户端需新增上报 + 定位权限 + 隐私合规(PIPL 敏感信息): latitude / longitude
device_liveness(无障碍存活/极光推送, device_liveness.py)是两张相互独立的表,
靠同一个 device_id 关联,不要合并
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import (
Boolean,
DateTime,
ForeignKey,
Integer,
String,
UniqueConstraint,
func,
)
from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class DeviceLiveness(Base):
# 表名不叫 device:device 易被当成「设备信息(品牌/型号/系统)」表;本表实为**无障碍存活监控状态**
# (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。
__tablename__ = "device_liveness"
__table_args__ = (
UniqueConstraint("user_id", "device_id", name="uq_device_liveness_user_device"),
)
class Device(Base):
__tablename__ = "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
# 客户端生成的设备唯一 ID(全局唯一);与 analytics_event / device_liveness 用同一个值
device_id: Mapped[str] = mapped_column(
String(128), unique=True, index=True, nullable=False
)
# 首次开无障碍(首次收到 accessibility_enabled 心跳)的时刻;ever_protected 第一次翻 true 时记一次,
# 后续心跳不覆盖。老设备(迁移前已 protected)无此值 → NULL。
first_protected_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
# 关联用户:登录后回填「最近一次登录」的 user;游客态为空(可空 → 不阻塞未登录建档)
user_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("user.id"), index=True, nullable=True
)
# 最近一次 service 心跳时间(存活证明);超时即视为保护掉线
last_heartbeat_at: Mapped[datetime | None] = mapped_column(
# ---- 设备 / 系统信息(埋点已采集,注册/上报接口带过来即可) ----
platform: Mapped[str] = mapped_column(String(16), nullable=False, default="android") # android/ios/harmony
oem: Mapped[str | None] = mapped_column(String(32), nullable=True) # 厂商 Build.MANUFACTURER:xiaomi/huawei…
model: Mapped[str | None] = mapped_column(String(64), nullable=True) # 型号 Build.MODEL,如 PJF110
os_version: Mapped[str | None] = mapped_column(String(32), nullable=True) # 系统版本,如 "Android 13"
app_version: Mapped[str | None] = mapped_column(String(32), nullable=True) # app 版本
channel: Mapped[str | None] = mapped_column(String(32), nullable=True) # 安装渠道(应用市场)
screen: Mapped[str | None] = mapped_column(String(32), nullable=True) # 分辨率 "1080x2400"
network: Mapped[str | None] = mapped_column(String(16), nullable=True) # 最近网络类型 wifi/4g/5g
# ---- 需客户端「新增」上报的字段 ----
# 设备时区 TimeZone.getDefault().id,如 "Asia/Shanghai"(客户端需新增上报,无需权限)
timezone: Mapped[str | None] = mapped_column(String(64), nullable=True)
# 是否模拟器(客户端 Build 指纹判断后上报;NULL=未知,无需权限)
is_emulator: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
# 位置(经纬度):App 会申请定位权限,但可能拿不到(首次运行未授权 / 用户拒绝 / 关了 GPS)。
# 规则(leader 定):每次上报以实际为准——能拿到就写,拿不到就置 NULL 清空旧值。
# ⚠️ upsert 时 lat/lng 必须「整字段覆盖(含 None)」,不能像 model/oem 那样 COALESCE 保留旧值,
# 否则会留下一条早已离开该位置的陈旧坐标。
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
# ---- 上下文 / 状态(服务端维护) ----
last_ip: Mapped[str | None] = mapped_column(String(64), nullable=True) # 最近一次上报 IP(服务端从 X-Forwarded-For 取)
status: Mapped[str] = mapped_column(String(16), nullable=False, default="normal") # normal/banned(风控封设备)
# 最近活跃时间(设备维度 DAU / 留存统计用;updated_at 只在字段真变化时跳,故单列一个)
last_active_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
@@ -90,7 +78,4 @@ class DeviceLiveness(Base):
)
def __repr__(self) -> str: # pragma: no cover
return (
f"<DeviceLiveness id={self.id} user_id={self.user_id} "
f"device_id={self.device_id} state={self.liveness_state}>"
)
return f"<Device id={self.id} device_id={self.device_id} user_id={self.user_id}>"
+96
View File
@@ -0,0 +1,96 @@
"""设备表(无障碍保护存活检测 + 极光推送)。
每条 = 一个用户的一台设备(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 DeviceLiveness(Base):
# 表名不叫 device:device 易被当成「设备信息(品牌/型号/系统)」表;本表实为**无障碍存活监控状态**
# (心跳 last_heartbeat_at + liveness_state + kill_alert_pending + 推送目标 registration_id),故名 device_liveness。
__tablename__ = "device_liveness"
__table_args__ = (
UniqueConstraint("user_id", "device_id", name="uq_device_liveness_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
)
# 首次开无障碍(首次收到 accessibility_enabled 心跳)的时刻;ever_protected 第一次翻 true 时记一次,
# 后续心跳不覆盖。老设备(迁移前已 protected)无此值 → NULL。
first_protected_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True
)
# 最近一次 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"<DeviceLiveness id={self.id} user_id={self.user_id} "
f"device_id={self.device_id} state={self.liveness_state}>"
)
+1 -1
View File
@@ -6,7 +6,7 @@ from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.device import DeviceLiveness
from app.models.device_liveness import DeviceLiveness
def _get(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None:
+66
View File
@@ -0,0 +1,66 @@
"""device 表(设备档案)读写:按 device_id upsert。
repositories/device.py(无障碍存活 DeviceLiveness)是不同的表:本文件写 device
"""
from __future__ import annotations
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.device import Device
from app.schemas.device_profile import DeviceReportRequest
# 设备/系统信息:提供了(非 None)才覆盖,某次上报漏带不会把已有值冲掉
_STICKY_STR_FIELDS = ("oem", "model", "os_version", "app_version", "channel", "screen", "network", "timezone")
def upsert_device(
db: Session,
req: DeviceReportRequest,
*,
user_id: int | None,
last_ip: str | None,
) -> Device:
"""按 device_id upsert 一台设备的档案。
写入策略分三类:
- 设备/系统信息(_STICKY_STR_FIELDS + platform + is_emulator):非空才覆盖(sticky)
- 位置 latitude/longitude:**整字段覆盖, None**以本次上报实际为准,拿不到即清空
(leader 规则, model Device.latitude 注释;不能像上面那样 sticky)
- user_id:仅登录态(user_id None)回填为当前登录用户;游客态保留已有关联,不清空
last_active_at / last_ip 每次刷新
"""
now = datetime.now(timezone.utc)
device = db.execute(
select(Device).where(Device.device_id == req.device_id)
).scalar_one_or_none()
if device is None:
device = Device(device_id=req.device_id)
db.add(device)
# 设备/系统信息:非空才覆盖
if req.platform:
device.platform = req.platform
for field in _STICKY_STR_FIELDS:
val = getattr(req, field)
if val is not None:
setattr(device, field, val)
if req.is_emulator is not None:
device.is_emulator = req.is_emulator
# 位置:整字段覆盖(含 None),以本次实际为准
device.latitude = req.latitude
device.longitude = req.longitude
# 关联用户:仅登录态回填,游客态不动
if user_id is not None:
device.user_id = user_id
device.last_ip = last_ip
device.last_active_at = now
db.commit()
db.refresh(device)
return device
-168
View File
@@ -1,168 +0,0 @@
"""连续 N 天无活跃账户的金币 / 现金清零。
规则(2026-07-15 产品定):某用户连续 `days` (默认 15)既没发起比价也没发起领券
判定为流失,清空其金币余额(coin_balance)与金币兑换现金(cash_balance_cents)
**单纯登录(打开 App)不算活跃**光登录不用比价/领券的用户视为流失照清
**邀请奖励金(invite_cash_balance_cents)物理隔离,不在清空范围**(产品红线,
models/wallet.py CoinAccount 注释);total_coin_earned(历史累计赚取,只增不减)也不动
活跃口径:发起比价 real_compare_start + 发起领券 real_coupon_start / claim_started
两路并集(**不含登录 last_login_at**) admin 大盘 DAU / 后台最近活跃**不同**
那两处含登录,此处刻意去掉,只认真正用了核心功能;若两处口径要同步须留意此差异
新用户保护:从没发起过比价/领券的用户永远不在活跃集合,故用**注册时间兜底**注册未满
days (created_at today-days 之后)的用户豁免清零,新用户有 days 天宽限,不会刚注册
(有礼金)就被清(登录已不算活跃,故不能再靠 last_login 兜底新用户)
清零走 wallet.grant_coins / grant_cash(负数出账入口):更新余额快照 + 写一笔
biz_type=inactive_clear 的流水( balance_after),可逐笔回溯 / 人工恢复;grant_coins
对负数不累加 total_coin_earned
"""
from __future__ import annotations
from datetime import UTC, date, datetime, timedelta
from sqlalchemy import or_, select
from sqlalchemy.orm import Session
from app.core import rewards
from app.models.analytics_event import AnalyticsEvent
from app.models.coupon_state import CouponPromptEngagement
from app.models.user import User
from app.models.wallet import CoinAccount
from app.repositories.wallet import get_or_create_account, grant_cash, grant_coins
CLEAR_BIZ_TYPE = "inactive_clear"
# 「活跃」计入的埋点事件:发起比价 + 发起领券(登录不算)
_ACTIVE_EVENTS = ("real_compare_start", "real_coupon_start")
def _cn_date_start_utc(d: date) -> datetime:
"""北京自然日 d 的 00:00 → UTC aware 下界。
analytics_event.created_at / user.created_at UTC(见各自模型),须用 UTC 边界比较;
CN_TZ 是固定 +8 偏移(rewards.CN_TZ), DST 歧义
"""
return datetime(d.year, d.month, d.day, tzinfo=rewards.CN_TZ).astimezone(UTC)
def _active_user_ids_since(db: Session, since_cn: date) -> set[int]:
"""北京自然日 since_cn(含)当天 0 点起 **发起过比价 / 领券** 的 user_id 集合。
**登录不算活跃**(产品定):只认核心功能行为两路并集:
- 发起比价 / 领券:analytics_event.event in _ACTIVE_EVENTS created_at >= since(UTC 边界)
- 发起领券:coupon_prompt_engagement.engage_type=claim_started engage_date >= since(北京日列)
"""
since_utc = _cn_date_start_utc(since_cn)
ids: set[int] = set()
ids.update(
uid
for uid in db.execute(
select(AnalyticsEvent.user_id).where(
AnalyticsEvent.user_id.is_not(None),
AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
AnalyticsEvent.created_at >= since_utc,
)
).scalars()
if uid is not None
)
ids.update(
uid
for uid in db.execute(
select(CouponPromptEngagement.user_id).where(
CouponPromptEngagement.user_id.is_not(None),
CouponPromptEngagement.engage_type == "claim_started",
CouponPromptEngagement.engage_date >= since_cn,
)
).scalars()
if uid is not None
)
return ids
def _new_user_ids_since(db: Session, since_cn: date) -> set[int]:
"""注册时间 >= since_cn 北京日 0 点的 user_id——注册未满 days 天的新用户,豁免清零。"""
since_utc = _cn_date_start_utc(since_cn)
return set(db.execute(select(User.id).where(User.created_at >= since_utc)).scalars())
def clear_inactive_accounts(db: Session, *, days: int, dry_run: bool = False) -> dict:
"""连续无活跃**满 `days` 整天之后**把用户金币 + 金币现金清零(邀请金不动),逐用户独立事务。
- 活跃 = 发起比价 / 发起领券(**登录不算**)last_active >= today-days 视为活跃;连续无
比价无领券满 days (且注册也满 days )才清days=15 最后一次比价/领券在 15 天前
当天仍留 16 天起清today 一律北京时(rewards.cn_today())
- 新用户保护:注册未满 days (created_at today-days 之后)豁免,不会刚注册就被清
- 只扫描有余额(coin_balance>0 cash_balance_cents>0)的账户:清零后余额=0,下一轮
自然跳过 天然幂等,重启 / 多次唤醒 / 补跑都安全
- 逐用户独立事务:单用户异常 rollback 不影响其他人
- dry_run=True:只统计不写库(供运营上线前预演:看会清哪些人清多少)
返回统计 dict(scanned/cleared/skipped_active/skipped_new_user/coin_cleared/
cents_cleared/failed)
"""
# days<1 会把 active_since 推到未来 → 全员判流失清空(灾难);防御性钳到 >=1。
days = max(1, days)
today = rewards.cn_today()
# 连续无活跃满 days 整天之后才清:active_since=today-days,活跃/新用户都以此为边界。
active_since = today - timedelta(days=days)
active_ids = _active_user_ids_since(db, active_since)
new_user_ids = _new_user_ids_since(db, active_since)
# 只取候选 user_id;清零金额在循环内读**实时**余额(不用此处快照)——活跃集合算完到逐行
# 清零之间余额可能变(如并发看广告发币),用快照 grant 会清不干净、balance_after≠0。
candidate_ids = (
db.execute(
select(CoinAccount.user_id).where(
or_(CoinAccount.coin_balance > 0, CoinAccount.cash_balance_cents > 0)
)
)
.scalars()
.all()
)
stats = {
"scanned": 0,
"cleared": 0,
"skipped_active": 0,
"skipped_new_user": 0,
"coin_cleared": 0,
"cents_cleared": 0,
"failed": 0,
}
remark = f"连续{days}天无活跃清零"
for user_id in candidate_ids:
stats["scanned"] += 1
if user_id in active_ids:
stats["skipped_active"] += 1
continue
if user_id in new_user_ids: # 注册未满 days 天,新用户宽限
stats["skipped_new_user"] += 1
continue
if dry_run:
acc = db.get(CoinAccount, user_id)
if acc is not None:
stats["cleared"] += 1
stats["coin_cleared"] += acc.coin_balance
stats["cents_cleared"] += acc.cash_balance_cents
continue
try:
# lock=True 对该账户行加 FOR UPDATE(PG 生效,SQLite no-op),读-清-写串行化防并发
acc = get_or_create_account(db, user_id, commit=False, lock=True)
coin, cents = acc.coin_balance, acc.cash_balance_cents
if coin <= 0 and cents <= 0:
continue # 快照后被并发清空 / 变动,无需再清
if coin > 0:
grant_coins(db, user_id, -coin, biz_type=CLEAR_BIZ_TYPE, remark=remark)
if cents > 0:
grant_cash(db, user_id, -cents, biz_type=CLEAR_BIZ_TYPE, remark=remark)
db.commit()
stats["cleared"] += 1
stats["coin_cleared"] += coin
stats["cents_cleared"] += cents
except Exception: # noqa: BLE001 - 批处理不因单用户异常中断
db.rollback()
stats["failed"] += 1
return stats
+35
View File
@@ -0,0 +1,35 @@
"""设备档案上报 schema(device 表)。
schemas/device.py(无障碍存活 DeviceLiveness register/heartbeat)是不同用途:
本文件对应 device (设备信息/档案),schemas/device.py 对应 device_liveness
"""
from __future__ import annotations
from pydantic import BaseModel, Field
class DeviceReportRequest(BaseModel):
"""客户端上报的一台设备的当前信息(每次以实际为准)。"""
device_id: str = Field(max_length=128)
platform: str = Field(default="android", max_length=16)
# 设备 / 系统信息:提供了才覆盖(见 repo:非空 sticky)
oem: str | None = Field(default=None, max_length=32)
model: str | None = Field(default=None, max_length=64)
os_version: str | None = Field(default=None, max_length=32)
app_version: str | None = Field(default=None, max_length=32)
channel: str | None = Field(default=None, max_length=32)
screen: str | None = Field(default=None, max_length=32)
network: str | None = Field(default=None, max_length=16)
timezone: str | None = Field(default=None, max_length=64)
is_emulator: bool | None = None
# 位置:客户端每次带「本次实际」的经纬度,拿不到就传 null(或不传)→ 服务端清空。
# 见 model Device.latitude 注释:这两个字段整字段覆盖,不做 sticky。
latitude: float | None = None
longitude: float | None = None
class DeviceReportOut(BaseModel):
ok: bool = True
+1 -1
View File
@@ -1,6 +1,6 @@
# device_liveness — 无障碍存活监控(心跳 + 掉线召回)
> 模型 `app/models/device.py`(`DeviceLiveness`) · 仓库 `app/repositories/device.py`(`register_or_update` / `touch_heartbeat` / `list_overdue` / `mark_notified` / `get_device` / `ack_kill_alert`) · 接口 用户 `POST /api/v1/device/register``POST /api/v1/device/heartbeat``GET /api/v1/device/liveness``POST /api/v1/device/liveness/ack`(`app/api/v1/device.py`);后台 worker `heartbeat_monitor_worker` · [← 索引](./README.md) · [总览](./OVERVIEW.md)
> 模型 `app/models/device_liveness.py`(`DeviceLiveness`) · 仓库 `app/repositories/device.py`(`register_or_update` / `touch_heartbeat` / `list_overdue` / `mark_notified` / `get_device` / `ack_kill_alert`) · 接口 用户 `POST /api/v1/device/register``POST /api/v1/device/heartbeat``GET /api/v1/device/liveness``POST /api/v1/device/liveness/ack`(`app/api/v1/device.py`);后台 worker `heartbeat_monitor_worker` · [← 索引](./README.md) · [总览](./OVERVIEW.md)
每行 = 一个用户的一台设备(per-install,`(user_id, device_id)` 唯一)。客户端无障碍服务存活时周期上报心跳刷新 `last_heartbeat_at`;App 前台/登录拿到极光 push token 时上报 `registration_id`。后端 `heartbeat_monitor_worker` 扫「曾保护过、现已心跳超时」的设备,推送(或本期仅终端打印)提醒用户重开无障碍。**表名不叫 `device`**:它存的不是设备信息(品牌/型号),而是**无障碍存活状态**。#65 新增。
+98
View File
@@ -0,0 +1,98 @@
"""设备档案上报端点 /api/v1/device/report + upsert 测试。"""
from __future__ import annotations
from sqlalchemy import select
from app.core.security import create_token
from app.db.session import SessionLocal
from app.models.device import Device
from app.models.user import User
def _payload(**over) -> dict:
base = {
"device_id": "dev-report-1",
"platform": "android",
"oem": "Xiaomi",
"model": "PJF110",
"os_version": "Android 14",
"app_version": "0.2.12(62)",
"channel": "yingyongbao",
"screen": "1080x2400",
"network": "wifi",
"timezone": "Asia/Shanghai",
"is_emulator": False,
"latitude": 31.23,
"longitude": 121.47,
}
base.update(over)
return base
def _get(device_id: str) -> Device | None:
with SessionLocal() as db:
return db.execute(
select(Device).where(Device.device_id == device_id)
).scalar_one_or_none()
def test_report_creates_device_as_guest(client) -> None:
r = client.post("/api/v1/device/report", json=_payload(device_id="dev-guest"))
assert r.status_code == 200, r.text
assert r.json()["ok"] is True
d = _get("dev-guest")
assert d is not None
assert d.user_id is None # 游客态:未绑用户
assert d.oem == "Xiaomi"
assert d.is_emulator is False
assert d.latitude == 31.23
assert d.last_active_at is not None
def test_report_binds_user_when_authed(client) -> None:
with SessionLocal() as db:
u = User(phone="13800000001", username="20000000001")
db.add(u)
db.commit()
uid = u.id
token, _ = create_token(user_id=uid, token_type="access")
r = client.post(
"/api/v1/device/report",
json=_payload(device_id="dev-user"),
headers={"Authorization": f"Bearer {token}"},
)
assert r.status_code == 200, r.text
assert _get("dev-user").user_id == uid
def test_report_clears_location_when_absent(client) -> None:
# 首次带位置
client.post(
"/api/v1/device/report",
json=_payload(device_id="dev-loc", latitude=10.0, longitude=20.0),
)
assert _get("dev-loc").latitude == 10.0
# 再次上报拿不到位置(lat/lng=None)→ 必须清空,不能保留旧坐标
r = client.post(
"/api/v1/device/report",
json=_payload(device_id="dev-loc", latitude=None, longitude=None, oem="HONOR"),
)
assert r.status_code == 200, r.text
d = _get("dev-loc")
assert d.latitude is None # 位置以实际为准 → 清空
assert d.longitude is None
assert d.oem == "HONOR" # 设备信息正常更新
def test_report_sticky_fields_not_wiped_by_missing(client) -> None:
# 首次全量
client.post("/api/v1/device/report", json=_payload(device_id="dev-sticky"))
# 再次只带 device_id + 位置(不带 oem/model)→ 设备信息保留旧值(非空才覆盖)
client.post(
"/api/v1/device/report",
json={"device_id": "dev-sticky", "latitude": 1.0, "longitude": 2.0},
)
d = _get("dev-sticky")
assert d.oem == "Xiaomi" # 未被漏带的 None 冲掉
assert d.model == "PJF110"
assert d.latitude == 1.0 # 位置按本次