From a6f55a00b82aad00b9749db430404b107355c737 Mon Sep 17 00:00:00 2001 From: chenshirui Date: Tue, 23 Jun 2026 17:33:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(device):=20=E6=97=A0=E9=9A=9C=E7=A2=8D?= =?UTF-8?q?=E4=BF=9D=E6=8A=A4=E5=AD=98=E6=B4=BB=E5=BF=83=E8=B7=B3=E6=A3=80?= =?UTF-8?q?=E6=B5=8B=20+=20=E6=8E=89=E7=BA=BF=E5=90=8E=E7=BD=AE=E6=A3=80?= =?UTF-8?q?=E6=B5=8B(pull)=20(#65)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: 陈世睿 <2839904623@qq.com> Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/65 Co-authored-by: chenshirui Co-committed-by: chenshirui --- .env.example | 5 + alembic/versions/ad_feed_reward_trace_id.py | 42 +++++ alembic/versions/device_kill_alert_pending.py | 39 +++++ alembic/versions/device_table.py | 53 +++++++ app/api/v1/ad.py | 17 ++ app/api/v1/device.py | 96 ++++++++++++ app/core/config.py | 5 + app/core/heartbeat_monitor_worker.py | 148 ++++++++++++++++++ app/main.py | 8 + app/models/__init__.py | 1 + app/models/ad_feed_reward.py | 3 + app/models/device.py | 91 +++++++++++ app/repositories/ad_feed_reward.py | 56 ++++--- app/repositories/comparison.py | 38 ++++- app/repositories/device.py | 124 +++++++++++++++ app/schemas/ad.py | 14 ++ app/schemas/compare_record.py | 12 ++ app/schemas/device.py | 48 ++++++ 18 files changed, 771 insertions(+), 29 deletions(-) create mode 100644 alembic/versions/ad_feed_reward_trace_id.py create mode 100644 alembic/versions/device_kill_alert_pending.py create mode 100644 alembic/versions/device_table.py create mode 100644 app/api/v1/device.py create mode 100644 app/core/heartbeat_monitor_worker.py create mode 100644 app/models/device.py create mode 100644 app/repositories/device.py create mode 100644 app/schemas/device.py diff --git a/.env.example b/.env.example index 1f53db8..646b275 100644 --- a/.env.example +++ b/.env.example @@ -27,6 +27,11 @@ 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 +# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)===== +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/ad_feed_reward_trace_id.py b/alembic/versions/ad_feed_reward_trace_id.py new file mode 100644 index 0000000..83703f2 --- /dev/null +++ b/alembic/versions/ad_feed_reward_trace_id.py @@ -0,0 +1,42 @@ +"""ad_feed_reward_record.trace_id (比价看广告金币归属到比价记录) + +Revision ID: ad_feed_reward_trace_id +Revises: device_kill_alert_pending +Create Date: 2026-06-21 + +比价等候期信息流广告(feed_ad_reward)结算时,客户端带上本场比价 trace_id 落此列; +比价记录页按 trace_id 聚合本场实发金币,显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'ad_feed_reward_trace_id' +down_revision: Union[str, Sequence[str], None] = 'device_kill_alert_pending' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # SQLite 下 ADD COLUMN(可空) 与 CREATE INDEX 均原生支持,无需 batch_alter_table。 + op.add_column( + 'ad_feed_reward_record', + sa.Column('trace_id', sa.String(length=64), nullable=True), + ) + op.create_index( + op.f('ix_ad_feed_reward_record_trace_id'), + 'ad_feed_reward_record', + ['trace_id'], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index( + op.f('ix_ad_feed_reward_record_trace_id'), + table_name='ad_feed_reward_record', + ) + op.drop_column('ad_feed_reward_record', 'trace_id') diff --git a/alembic/versions/device_kill_alert_pending.py b/alembic/versions/device_kill_alert_pending.py new file mode 100644 index 0000000..b25b219 --- /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: device_liveness_table +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] = 'device_liveness_table' +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_liveness', 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_liveness', 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..ffe1e71 --- /dev/null +++ b/alembic/versions/device_table.py @@ -0,0 +1,53 @@ +"""device table (无障碍保护存活检测 + 极光推送) + +Revision ID: device_liveness_table +Revises: ceb286289426 +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] = 'ceb286289426' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'device_liveness', + 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_liveness_user_device'), + ) + with op.batch_alter_table('device_liveness', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_device_liveness_user_id'), ['user_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_liveness_device_id'), ['device_id'], unique=False) + batch_op.create_index(batch_op.f('ix_device_liveness_last_heartbeat_at'), ['last_heartbeat_at'], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table('device_liveness', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_device_liveness_last_heartbeat_at')) + batch_op.drop_index(batch_op.f('ix_device_liveness_device_id')) + batch_op.drop_index(batch_op.f('ix_device_liveness_user_id')) + + op.drop_table('device_liveness') diff --git a/app/api/v1/ad.py b/app/api/v1/ad.py index 79dbcfd..7f203a8 100644 --- a/app/api/v1/ad.py +++ b/app/api/v1/ad.py @@ -32,6 +32,7 @@ from app.schemas.ad import ( EcpmReportOut, FeedRewardIn, FeedRewardOut, + FeedRewardUnitsOut, PangleCallbackOut, RewardNoShowIn, RewardNoShowOut, @@ -373,6 +374,7 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed adn=payload.adn, slot_id=payload.slot_id, feed_scene=payload.feed_scene, + trace_id=payload.trace_id, app_env=payload.app_env, our_code_id=payload.our_code_id, aborted=payload.aborted, @@ -390,6 +392,21 @@ def feed_reward(payload: FeedRewardIn, user: CurrentUser, db: DbSession) -> Feed ) +@router.get( + "/feed-reward/units", + response_model=FeedRewardUnitsOut, + summary="查账号累计信息流发奖份数(前端算因子2 LT、做实时金币进度条用)", + dependencies=[Depends(rate_limit(120, 60, "ad-feed-units"))], +) +def feed_reward_units(user: CurrentUser, db: DbSession) -> FeedRewardUnitsOut: + """返回账号累计已发奖份数(因子2 LT 基线)。 + + 信息流金币进度条:前端 show 时拉一次,之后每满 10 秒一份、逐份用 (granted_units + offset) + 查 LT 因子,精确复刻发奖公式 → 进度条数值 ≈ 实际到账。 + """ + return FeedRewardUnitsOut(granted_units=crud_feed.granted_unit_total(db, user.id)) + + @router.post( "/reward-noshow", response_model=RewardNoShowOut, 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 0520923..8b3fe68 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -66,6 +66,11 @@ class Settings(BaseSettings): JG_VERIFY_ENDPOINT: str = "https://api.verification.jpush.cn/v1/web/loginTokenVerify" JG_REQUEST_TIMEOUT_SEC: int = 15 + # 无障碍保护存活监控后台任务(pull 后置检测;本期不接推送) + HEARTBEAT_MONITOR_ENABLED: bool = True # 总开关 + HEARTBEAT_TIMEOUT_MINUTES: int = 10 # 多久没心跳算掉线(≈3 个客户端心跳周期) + HEARTBEAT_SCAN_INTERVAL_SEC: int = 60 # 扫描周期 + # ===== 短信 ===== 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/main.py b/app/main.py index af011f8..5a4a6bd 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.daily_exchange_worker import ( start_daily_exchange_worker, stop_daily_exchange_worker, @@ -63,10 +68,12 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: settings.DATABASE_URL.split("://", 1)[0], ) reconcile_task = start_withdraw_reconcile_worker() + heartbeat_task = start_heartbeat_monitor() daily_exchange_task = start_daily_exchange_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) logger.info("shutting down") @@ -100,6 +107,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..09ce54a 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 DeviceLiveness # noqa: F401 from app.models.coupon_state import ( # noqa: F401 CouponClaimRecord, CouponDailyCompletion, diff --git a/app/models/ad_feed_reward.py b/app/models/ad_feed_reward.py index 55acbee..f213c10 100644 --- a/app/models/ad_feed_reward.py +++ b/app/models/ad_feed_reward.py @@ -33,6 +33,9 @@ class AdFeedRewardRecord(Base): # 点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页)。比价与领券共用同一信息流 # 代码位,slot_id/our_code_id 分不出,只能客户端各调用点显式打标;NULL=历史/未升级客户端=未分类。 feed_scene: Mapped[str | None] = mapped_column(String(16), nullable=True) + # 本次比价 trace_id(仅 comparison 场景由客户端带上):把这场广告金币归属到对应比价记录, + # 比价记录页按 trace_id 聚合本场实发金币显示「比价赚 N 金币」。领券/福利/旧客户端 = NULL。 + trace_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True) # 来源(广告收益报表用):我们的应用环境 prod/test + 我们配置的代码位 104xxx,由客户端 feed-reward 上报带上。旧数据为 NULL。 app_env: Mapped[str | None] = mapped_column(String(16), nullable=True) our_code_id: Mapped[str | None] = mapped_column(String(64), nullable=True) diff --git a/app/models/device.py b/app/models/device.py new file mode 100644 index 0000000..339d6d9 --- /dev/null +++ b/app/models/device.py @@ -0,0 +1,91 @@ +"""设备表(无障碍保护存活检测 + 极光推送)。 + +每条 = 一个用户的一台设备(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 + ) + # 最近一次 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/ad_feed_reward.py b/app/repositories/ad_feed_reward.py index f5912ac..5b8595d 100644 --- a/app/repositories/ad_feed_reward.py +++ b/app/repositories/ad_feed_reward.py @@ -42,21 +42,20 @@ def _granted_today(db: Session, user_id: int, reward_date: str) -> int: ).scalar_one() -def _unit_reward_total(db: Session, user_id: int, ecpm: str, unit_count: int) -> int: - """按每个 10 秒单位逐份计算奖励,LT 使用**账号累计**奖励份序号(不按天重置)。""" - if unit_count <= 0: - return 0 - existing_units = db.execute( - select(func.coalesce(func.sum(AdFeedRewardRecord.unit_count), 0)) - .where( - AdFeedRewardRecord.user_id == user_id, - AdFeedRewardRecord.status == "granted", - ) - ).scalar_one() - total = 0 - for offset in range(1, unit_count + 1): - total += rewards.calculate_ad_reward_coin(ecpm, int(existing_units) + offset) - return total +def granted_unit_total(db: Session, user_id: int) -> int: + """账号累计已发奖**条**数(COUNT status=granted),不按天重置 = 因子2(LT)基线。 + + 口径:**一条广告 = 一份额度(一个单次公式值)**,故"已发条数"即"已发份数"。供前端拉取后精确复刻 + 金币公式做实时金币进度条(前端因子2 用 granted_unit_total + 本场已结算条数 + 1 取档,与本表发奖一致)。 + """ + return int( + db.execute( + select(func.count()).select_from(AdFeedRewardRecord).where( + AdFeedRewardRecord.user_id == user_id, + AdFeedRewardRecord.status == "granted", + ) + ).scalar_one() + ) def grant_feed_reward( @@ -70,20 +69,21 @@ def grant_feed_reward( adn: str | None = None, slot_id: str | None = None, feed_scene: str | None = None, + trace_id: str | None = None, app_env: str | None = None, our_code_id: str | None = None, aborted: bool = False, ) -> AdFeedRewardRecord: - """比价/领券一整场信息流(轮播多条)结束后结算奖励。client_event_id 幂等,同号重试不重复发。 + """**每条**信息流广告(客户端每条各上报一次)结算奖励。client_event_id 幂等,同号重试不重复发。 - 发奖规则:**比价全程不关广告才发**,金额按整场**总观看时长**折份(每 10 秒 1 份)。 - - aborted=True(用户中途 ✕ 关闭):整场不发,记 status='closed_early' 留痕(原因可查)。 - - 总时长不足 10 秒(unit_count==0):记 status='too_short' 不发。 + 发奖规则:**一条广告 = 一个单次公式值**(rewards.calculate_ad_reward_coin),因子2(LT)按账号累计 + **条**数递进;看满一份时长(unit_count>=1, 即 ≥10 秒)才发,**不逐份累加**。 + - aborted=True(用户中途 ✕ 关闭这条):本条不发,记 status='closed_early'。 + - 时长不足 10 秒(unit_count==0):记 status='too_short' 不发。 - 命中当日条数上限:记 status='capped' 不发。 - duration_seconds 是整场累计秒数。一期 eCPM/时长均由客户端上报,故服务端两道硬闸防刷:时长钳到 - FEED_MAX_DURATION_SECONDS 限单场份数,eCPM 在 rewards.calculate_ad_reward_coin 内钳到 - AD_ECPM_MAX_FEN 限单份金额;叠加每日 get_ad_daily_limit 条数上限,把单用户日产出锁进有限区间。 - feed_scene:点位场景(comparison/coupon/welfare),仅做归类落库,不参与发奖计算。 + duration_seconds 是**这一条**的观看秒数。服务端两道硬闸防刷:时长钳到 FEED_MAX_DURATION_SECONDS、 + eCPM 在 calculate_ad_reward_coin 内钳到 AD_ECPM_MAX_FEN;叠加每日 get_ad_daily_limit 条数上限。 + feed_scene:点位场景(comparison/coupon/welfare),仅归类落库,不参与计算。 """ existing = _find_by_event(db, client_event_id) if existing is not None: @@ -107,6 +107,7 @@ def grant_feed_reward( adn=adn, slot_id=slot_id, feed_scene=feed_scene, + trace_id=trace_id, app_env=app_env, our_code_id=our_code_id, coin=0, @@ -126,6 +127,7 @@ def grant_feed_reward( adn=adn, slot_id=slot_id, feed_scene=feed_scene, + trace_id=trace_id, app_env=app_env, our_code_id=our_code_id, coin=0, @@ -146,6 +148,7 @@ def grant_feed_reward( adn=adn, slot_id=slot_id, feed_scene=feed_scene, + trace_id=trace_id, app_env=app_env, our_code_id=our_code_id, coin=0, @@ -153,12 +156,14 @@ def grant_feed_reward( ) return _commit_record(db, rec, client_event_id) - coin = _unit_reward_total(db, user_id, ecpm, unit_count) + # 一条广告 = 一个「单次公式值」(因子2 按账号累计**条**数, 即第 existing_ads+1 条);看满一份(unit_count>=1)即发,不逐份累加。 + existing_ads = granted_unit_total(db, user_id) + coin = rewards.calculate_ad_reward_coin(ecpm, existing_ads + 1) if coin > 0: crud_wallet.grant_coins( db, user_id, coin, biz_type="feed_ad_reward", ref_id=client_event_id, - remark=f"信息流广告奖励 {unit_count}份", + remark="信息流广告奖励", ) rec = AdFeedRewardRecord( client_event_id=client_event_id, @@ -171,6 +176,7 @@ def grant_feed_reward( adn=adn, slot_id=slot_id, feed_scene=feed_scene, + trace_id=trace_id, app_env=app_env, our_code_id=our_code_id, coin=coin, diff --git a/app/repositories/comparison.py b/app/repositories/comparison.py index 09fa555..d675822 100644 --- a/app/repositories/comparison.py +++ b/app/repositories/comparison.py @@ -11,6 +11,7 @@ from sqlalchemy import func, select from sqlalchemy.orm import Session from app.core.rewards import CN_TZ +from app.models.ad_feed_reward import AdFeedRewardRecord from app.models.comparison import ComparisonRecord from app.models.savings import SavingsRecord from app.schemas.compare_record import ComparisonRecordIn @@ -155,6 +156,29 @@ def _ordered_shop_names(db: Session, user_id: int) -> set[str]: return {s for s in rows if s} +def _ad_coins_by_trace(db: Session, user_id: int, trace_ids: list[str]) -> dict[str, int]: + """各 trace_id「看广告赚的金币」:按 trace_id 聚合该用户已发放(granted)的信息流广告金币。 + + 广告金币来自比价等候期信息流(biz_type=feed_ad_reward);客户端结算时把本场 trace_id 带上, + 落到 ad_feed_reward_record.trace_id。此处按 trace_id 求和 = 这次比价看广告实发的金币。 + trace_id 仅比价场景客户端带(领券/福利/旧客户端为 NULL),故按 trace_id 过滤天然只算比价广告。 + 空集合直接返回(避免 IN () 非法)。 + """ + if not trace_ids: + return {} + rows = db.execute( + select( + AdFeedRewardRecord.trace_id, + func.coalesce(func.sum(AdFeedRewardRecord.coin), 0), + ).where( + AdFeedRewardRecord.user_id == user_id, + AdFeedRewardRecord.status == "granted", + AdFeedRewardRecord.trace_id.in_(trace_ids), + ).group_by(AdFeedRewardRecord.trace_id) + ).all() + return {tid: int(coin) for tid, coin in rows if tid} + + def list_records( db: Session, user_id: int, @@ -162,7 +186,7 @@ def list_records( limit: int = 20, cursor: int | None = None, ) -> tuple[list[ComparisonRecord], int | None]: - """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记(瞬态,不写库)。""" + """比价记录分页(按创建时间倒序、id 兜底,游标式)。附「已下单」店级标记 + 「看广告赚的金币」(瞬态,不写库)。""" stmt = select(ComparisonRecord).where(ComparisonRecord.user_id == user_id) if cursor is not None: stmt = stmt.where(ComparisonRecord.id < cursor) @@ -172,10 +196,13 @@ def list_records( next_cursor = items[-1].id if len(items) == limit else None # 「已下单」标记:本页记录的 store_name 若落在该用户真实下单的店名集合里即 True。 - # ordered 非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。 + # ordered / ad_coins_earned 均非 ORM 列,仅挂实例上供 ComparisonRecordOut(from_attributes) 读出,不持久化。 ordered_shops = _ordered_shop_names(db, user_id) + # 「本次比价看广告赚的金币」:按本页 trace_id 一次性聚合(同 ordered 范式)。 + ad_coins = _ad_coins_by_trace(db, user_id, [it.trace_id for it in items]) for it in items: it.ordered = bool(it.store_name and it.store_name in ordered_shops) + it.ad_coins_earned = ad_coins.get(it.trace_id, 0) return items, next_cursor @@ -210,10 +237,13 @@ def get_stats(db: Session, user_id: int) -> tuple[int, int]: def get_record(db: Session, user_id: int, record_id: int) -> ComparisonRecord | None: - """取单条(限本人,避免越权读他人记录)。""" - return db.execute( + """取单条(限本人,避免越权读他人记录)。附「看广告赚的金币」瞬态属性(同 list_records)。""" + rec = db.execute( select(ComparisonRecord).where( ComparisonRecord.id == record_id, ComparisonRecord.user_id == user_id, ) ).scalar_one_or_none() + if rec is not None: + rec.ad_coins_earned = _ad_coins_by_trace(db, user_id, [rec.trace_id]).get(rec.trace_id, 0) + return rec diff --git a/app/repositories/device.py b/app/repositories/device.py new file mode 100644 index 0000000..e24fe35 --- /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 DeviceLiveness + + +def _get(db: Session, *, user_id: int, device_id: str) -> DeviceLiveness | None: + stmt = select(DeviceLiveness).where( + DeviceLiveness.user_id == user_id, DeviceLiveness.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, +) -> DeviceLiveness: + """注册设备或更新其 registration_id / 元信息。upsert by (user_id, device_id)。""" + device = _get(db, user_id=user_id, device_id=device_id) + if device is None: + device = DeviceLiveness( + 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, +) -> DeviceLiveness: + """处理一次心跳(心跳也能自注册)。 + + 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 = DeviceLiveness(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[DeviceLiveness]: + """掉线设备:曾经保护过、当前 alive、心跳超时。 + + 本期只做终端打印检测、不推送 → 不再要求有 registration_id(没接极光 token 的设备也要检出)。 + """ + cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes) + stmt = select(DeviceLiveness).where( + DeviceLiveness.ever_protected.is_(True), + DeviceLiveness.liveness_state == "alive", + DeviceLiveness.last_heartbeat_at.is_not(None), + DeviceLiveness.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(DeviceLiveness, 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) -> DeviceLiveness | 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/ad.py b/app/schemas/ad.py index b7feccf..d4e91fd 100644 --- a/app/schemas/ad.py +++ b/app/schemas/ad.py @@ -141,6 +141,12 @@ class FeedRewardIn(BaseModel): description="点位场景:comparison(比价等待) / coupon(领券) / welfare(福利页);" "比价与领券共用同一信息流代码位,需客户端在各调用点显式标注,缺省=未分类", ) + trace_id: str | None = Field( + None, + max_length=64, + description="本次比价 trace_id(比价场景必带):把这场广告金币归属到对应比价记录," + "比价记录页据此显示「比价赚 N 金币」。领券/福利场景为空", + ) app_env: str | None = Field( None, max_length=16, description="我们的穿山甲应用环境:prod(傻瓜比价正式) / test(测试应用)" ) @@ -160,6 +166,14 @@ class FeedRewardOut(BaseModel): daily_limit: int = Field(..., description="每日信息流展示次数上限") +class FeedRewardUnitsOut(BaseModel): + granted_units: int = Field( + ..., + description="账号累计已发奖的信息流**条**数(一条广告=一份额度),不按天重置;" + "前端据此算因子2(LT)做实时金币进度条,与后端发奖公式对齐", + ) + + class RewardNoShowIn(BaseModel): """激励视频展示了但用户提前关闭/跳过、未触发发奖——上报留痕(不发金币)。 diff --git a/app/schemas/compare_record.py b/app/schemas/compare_record.py index 23c0fea..9fa196e 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): @@ -73,6 +80,8 @@ class ComparisonRecordIn(BaseModel): # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列), # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是 # 按平台 id 索引的 dict 而非 list,只作 debug 展示)。 + # ⚠️ 早先误声明成 list → 每条成功记录 422 被拒、记录页只剩失败/终止 + # (2026-06-17 引入 platform_results 上报后复现);改 dict 后修复。 platform_results: dict = Field(default_factory=dict) skipped_dish_count: int | None = None skipped_dish_names: list[str] = Field(default_factory=list) @@ -138,6 +147,9 @@ class ComparisonRecordOut(BaseModel): # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。 # 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。 ordered: bool = False + # 「本次比价看广告赚的金币」(feed_ad_reward,按 trace_id 聚合 granted 金币;非 DB 列, + # list_records/get_record 动态挂在 ORM 实例上,from_attributes 读出)。0=没赚到 → 记录页不显示金币标。 + ad_coins_earned: int = 0 created_at: datetime diff --git a/app/schemas/device.py b/app/schemas/device.py new file mode 100644 index 0000000..dd7a7c4 --- /dev/null +++ b/app/schemas/device.py @@ -0,0 +1,48 @@ +"""设备注册 / 心跳相关 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 版)。客户端只需这一个布尔判断要不要弹「开启自启动」引导, + 故只返回 kill_alert_pending(不暴露设备详情 / 内部 liveness_state 等)。从未注册过 → 默认 False(无告警)。""" + model_config = ConfigDict(from_attributes=True) + + kill_alert_pending: bool = False + + +class LivenessAckRequest(BaseModel): + device_id: str