feat(welfare): 15 天不活跃清零(金币+折算现金,邀请金不清)+ 预警 + admin 活跃口径统一 (#143)

背景 / 目标
连续 15 天(北京自然日)没有「首页可见 / 发起比价 / 发起领券」行为的用户判定为流失,每日自动清零其金币 + 折算现金;清零前按可配置节奏预警;全程留审计。纯登录不算活跃;邀请奖励金物理隔离、不清(产品红线)。

改了什么
活跃口径共享模块 app/repositories/activity.py:worker(清零/预警)与 admin(最近活跃)单一真源,防漂移。
清零业务逻辑 inactivity.py:选取 / 逐用户清零(行锁 + 幂等)/ 预警分档 + streak 去重 / run_once 组合,预警故障逐用户隔离、绝不阻塞清零。
每日 worker inactivity_reset_worker.py(仿 daily_exchange:文件锁 + 北京日守卫 + RUN_HOUR 门槛 + 总闸)+ main.py lifespan 接线。
可插拔通知器 notifier.py(v1 LogNotifier 日志占位,预留 JPush/短信)。
配置 INACTIVITY_*(阈值 / 预警档 / 执行点 / 通道 / 开关)。
admin 最近活跃口径改用共享模块(移除 last_login_at、纳入 show/home、以 created_at 为基线)。
审计:inactivity_reset_log + inactivity_notification_log 两表 + 钱包流水双写(biz_type=inactivity_reset,ref_id 交叉)。
文档:设计 spec / 实现 plan / docs/database/ 两表字典。
关键产品决策
邀请金不清:只清金币 + 折算现金,invite_cash_balance_cents 原封(仅快照入审计)。
活跃口径 = max(created_at, 首页可见, 比价, 领券),不含 last_login_at;首页可见 = event=show + page=home。
时间边界:北京自然日 0 点对齐(见 activity.reset_cutoff)。
总闸默认关,灰度验证后再开。
数据库变更
新表:inactivity_reset_log、inactivity_notification_log。
analytics_event 新增复合覆盖索引 ix_analytics_event_active (event, page, user_id, created_at)(活跃口径聚合热点)。
修复了 base 上的迁移多头(135e79414fd0 与 phone_rebind_log 同从 comparison_llm_cost 分叉)→ 加空 merge 修订,alembic upgrade head 恢复单头正常。
⚠️ 上线注意(合并后 / 开总闸前)
INACTIVITY_RESET_ENABLED 默认 false;开启前提 = show/home 埋点全量铺满——否则"只登录不操作"且注册满 15 天的老用户会落到 created_at 基线被误清。
前端依赖:Android 端需在首页可见上报 event=show + page=home(携带登录后的 user_id)。
admin「最近活跃」口径变化(去登录、纳入 home_view、created_at 基线):属预期变化,需产品/运营知会;与 DAU(_period_active_user_ids,仍含登录)是两套指标。
清零对 C 端「金币流水」可见(biz_type=inactivity_reset,备注「15天不活跃清零」)。
灰度:先只看预警/清零名单对不对,再开总闸。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #143

app/admin/repositories/queries.py

@@ -11,7 +11,6 @@ from zoneinfo import ZoneInfo
 from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
-from app.admin.repositories.stats import COMPARE_START_EVENT, COUPON_START_EVENT
 from app.core import rewards
 from app.core.config import settings
 from app.models.ad_feed_reward import AdFeedRewardRecord
@@ -32,10 +31,7 @@ from app.models.wallet import (
     InviteCashTransaction,
     WithdrawOrder,
 )
-from app.repositories import ad_ecpm
-
-# 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
-_ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
+from app.repositories import activity, ad_ecpm
 
 # 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
 _NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
@@ -88,49 +84,6 @@ def offset_paginate(
     return items, next_cursor, total
 
 
-def _last_active_parts():
-    """「最近活跃」的两个按 user_id 预聚合派生表(最近开始比价/领券事件、最近领券发起)。
-
-    活跃口径与大盘 DAU/留存一致(2026-07-05 产品定:进入 App≈登录 last_login_at +
-    发起比价 real_compare_start + 发起领券 real_coupon_start/claim_started)。
-    用 LEFT JOIN 预聚合而非相关标量子查询:后者在 PG 上对 users 每行各跑一个 SubPlan
-    (排序键、range 筛选、offset_paginate 的 count 三处叠加),埋点表大了会拖垮列表接口;
-    预聚合借 analytics_event.event 索引只扫两类 start 事件,每次查询聚合一次。
-    """
-    ev_agg = (
-        select(
-            AnalyticsEvent.user_id.label("user_id"),
-            func.max(AnalyticsEvent.created_at).label("last_at"),
-        )
-        .where(
-            AnalyticsEvent.user_id.is_not(None),
-            AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
-        )
-        .group_by(AnalyticsEvent.user_id)
-        .subquery()
-    )
-    eng_agg = (
-        select(
-            CouponPromptEngagement.user_id.label("user_id"),
-            func.max(CouponPromptEngagement.created_at).label("last_at"),
-        )
-        .where(
-            CouponPromptEngagement.user_id.is_not(None),
-            CouponPromptEngagement.engage_type == "claim_started",
-        )
-        .group_by(CouponPromptEngagement.user_id)
-        .subquery()
-    )
-    return ev_agg, eng_agg
-
-
-def _norm_utc(dt: datetime | None) -> datetime | None:
-    """naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
-    if dt is None:
-        return None
-    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
-
-
 def _attach_last_active(db: Session, users: list[User]) -> None:
     """给本页用户瞬态挂 last_active_at(非 DB 列,供 AdminUserListItem from_attributes 读)。
 
@@ -144,7 +97,7 @@ def _attach_last_active(db: Session, users: list[User]) -> None:
             select(AnalyticsEvent.user_id, func.max(AnalyticsEvent.created_at))
             .where(
                 AnalyticsEvent.user_id.in_(uids),
-                AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
+                activity.active_event_condition(),
             )
             .group_by(AnalyticsEvent.user_id)
         ).all()
@@ -161,9 +114,9 @@ def _attach_last_active(db: Session, users: list[User]) -> None:
     )
     for u in users:
         candidates = [
-            _norm_utc(u.last_login_at),
-            _norm_utc(ev_map.get(u.id)),
-            _norm_utc(eng_map.get(u.id)),
+            activity.norm_utc(u.created_at),      # baseline 由 last_login_at 改为 created_at(登录不算活跃)
+            activity.norm_utc(ev_map.get(u.id)),
+            activity.norm_utc(eng_map.get(u.id)),
         ]
         u.last_active_at = max((c for c in candidates if c is not None), default=None)
 
@@ -191,16 +144,12 @@ def list_users(
     (口径见 [_last_active_expr])。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
     代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
     日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
-    # 最近活跃 = max(最近登录, 最近行为事件, 最近领券发起)。PG 用 GREATEST;SQLite 标量 max()
-    # 任一参数 NULL 即返回 NULL,故 LEFT JOIN 未命中侧 coalesce 到 last_login_at 兜底
-    # (注册即登录,该列恒非空)。派生表 1:1(按 user_id 聚合),outerjoin 不会放大行数,
-    # offset_paginate 的 count 不受影响。
-    ev_agg, eng_agg = _last_active_parts()
-    greatest = func.greatest if db.get_bind().dialect.name == "postgresql" else func.max
-    last_active = greatest(
-        User.last_login_at,
-        func.coalesce(ev_agg.c.last_at, User.last_login_at),
-        func.coalesce(eng_agg.c.last_at, User.last_login_at),
+    # 最近活跃 = max(注册时间, 最近行为事件, 最近领券发起)。baseline 由 last_login_at 改为 created_at
+    #(登录不代表在用 App;口径统一到 activity.py,含 home_view + 比价 + 领券,见 activity.ACTIVE_EVENTS)。
+    # 未命中侧 coalesce 到 created_at(恒非空基线)。派生表 1:1,outerjoin 不放大行数。
+    ev_agg, eng_agg = activity.last_active_subqueries(db)
+    last_active = activity.last_active_expr(
+        User.created_at, ev_agg, eng_agg, db.get_bind().dialect.name
     )
     stmt = (
         select(User)

app/core/config.py

@@ -178,6 +178,13 @@ class Settings(BaseSettings):
     # 进程内自动兑换 worker 的检查间隔(秒):每隔这么久醒一次,跨过北京 0 点就跑一轮。
     # 默认 600s=10min,即 0 点后最多 10 分钟内兑完(客户端文案已注明「可能存在延迟」)。
     AUTO_EXCHANGE_CHECK_INTERVAL_SEC: int = 600
+    # === 15 天不活跃清零(app.core.inactivity_reset_worker)===
+    INACTIVITY_RESET_ENABLED: bool = False           # 总闸,默认关;灰度验证后再开
+    INACTIVITY_RESET_DAYS: int = 15                  # 不活跃阈值(天),第 (N+1) 日 0 点清
+    INACTIVITY_WARN_DAYS_BEFORE: str = "7,2"         # 清零前几天各推一次;""=不推。逗号分隔
+    INACTIVITY_RESET_RUN_HOUR: int = 3               # 北京时间每日执行点(0-23)
+    INACTIVITY_NOTIFY_CHANNEL: str = "log"           # log(占位) / jpush / sms
+    INACTIVITY_RESET_CHECK_INTERVAL_SEC: int = 1800  # worker 唤醒间隔(秒)
     # 免确认收款授权(用户授权免确认模式)的授权结果回调地址,必须公网可访问 HTTPS、不带参数。
     # 发起授权 / 首单顺带授权时作为 authorization_notify_url 传给微信。一期不处理回调内容
     # (授权状态靠 query 查询兜底),但微信要求该字段非空,故启用免确认前必须配置;留空时免确认相关接口返回未配置。
@@ -198,6 +205,19 @@ class Settings(BaseSettings):
         """免确认收款授权可用 = 微信支付凭证齐全 + 授权回调地址已配。"""
         return bool(self.wxpay_configured and self.WXPAY_AUTH_NOTIFY_URL)
 
+    @property
+    def inactivity_warn_stages(self) -> list[int]:
+        """解析 INACTIVITY_WARN_DAYS_BEFORE → 降序去重的提前天数列表。
+        丢弃非数字 / <=0 / >=RESET_DAYS 的项(空串 → 空列表 = 不推)。"""
+        out: list[int] = []
+        for part in (self.INACTIVITY_WARN_DAYS_BEFORE or "").split(","):
+            part = part.strip()
+            if part.isdigit():
+                v = int(part)
+                if 0 < v < self.INACTIVITY_RESET_DAYS and v not in out:
+                    out.append(v)
+        return sorted(out, reverse=True)
+
     # ===== 穿山甲激励视频(服务端发奖回调)=====
     # 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
     # 穿山甲后台配置的"奖励校验密钥"(m-key),验签用。每个 GroMore 广告位 m-key 不同(后台各自

app/core/inactivity_reset_worker.py

@@ -0,0 +1,143 @@
+"""15 天不活跃清零的进程内每日任务。
+
+仿 daily_exchange_worker:App 启动自带,每 `INACTIVITY_RESET_CHECK_INTERVAL_SEC` 醒一次,
+跨进北京新的一天且到达 `INACTIVITY_RESET_RUN_HOUR`(默认 3 点)后跑一轮 `run_once`(预警 + 清零)。
+
+健壮性:
+- **逐用户幂等**:清完余额=0 次日不再匹配;预警按 streak 去重。启动补跑 / 多次唤醒 / 重启都安全。
+- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑。
+- **开关**:settings.INACTIVITY_RESET_ENABLED=false 时不启动(默认关,灰度验证后再开)。
+
+⚠️ 这是不可逆批量资金操作(清空金币 + 折算现金,**邀请现金不清**)。口径见
+app.repositories.inactivity / app.repositories.activity。
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+import os
+import time
+from collections.abc import Iterator
+from datetime import date, datetime
+from pathlib import Path
+
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ, cn_today
+from app.db.session import SessionLocal
+from app.integrations.notifier import get_notifier
+from app.repositories import inactivity as inactivity_repo
+
+logger = logging.getLogger("shagua.inactivity")
+_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "inactivity_reset.lock"
+
+
+def _cn_today() -> date:
+    return cn_today()
+
+
+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 _run_once_entry() -> dict:
+    """跑一轮(预警 + 清零)。独立开 Session。"""
+    notifier = get_notifier(settings.INACTIVITY_NOTIFY_CHANNEL)
+    with SessionLocal() as db:
+        return inactivity_repo.run_once(
+            db,
+            notifier=notifier,
+            reset_days=settings.INACTIVITY_RESET_DAYS,
+            warn_stages=settings.inactivity_warn_stages,
+            today=_cn_today(),
+        )
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.INACTIVITY_RESET_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("inactivity reset skipped: another worker owns lock")
+            return
+        await _run_locked_loop(interval)
+
+
+async def _run_locked_loop(interval: int) -> None:
+    logger.info(
+        "inactivity reset worker started interval=%ss run_hour=%s",
+        interval,
+        settings.INACTIVITY_RESET_RUN_HOUR,
+    )
+    # 本进程上次跑过的北京日;None=尚未跑过本进程(当天到点即补)。
+    last_run: date | None = None
+    try:
+        while True:
+            try:
+                _touch_lock()
+                today = _cn_today()
+                hour = datetime.now(CN_TZ).hour
+                if last_run != today and hour >= int(settings.INACTIVITY_RESET_RUN_HOUR):
+                    result = await asyncio.to_thread(_run_once_entry)
+                    last_run = today
+                    logger.info("inactivity reset done date=%s result=%s", today, result)
+            except SQLAlchemyError:
+                logger.exception("inactivity reset db error")
+            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
+                logger.exception("inactivity reset unexpected error")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("inactivity reset worker stopped")
+        raise
+
+
+def start_inactivity_reset_worker() -> asyncio.Task | None:
+    if not settings.INACTIVITY_RESET_ENABLED:
+        logger.info("inactivity reset disabled (INACTIVITY_RESET_ENABLED=false)")
+        return None
+    return asyncio.create_task(_run_loop(), name="inactivity-reset")
+
+
+async def stop_inactivity_reset_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/integrations/notifier.py

@@ -0,0 +1,43 @@
+"""不活跃预警通知器(可插拔)。
+
+v1 仅日志占位(LogNotifier):现状无真实推送能力(极光只用于一键登录解密 + 设备心跳告警,
+心跳 worker 也只打印),先把清零主流程 + 审计做扎实。后续实现同协议的 JPushNotifier /
+SmsNotifier 即可替换,worker/repo 不改。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Protocol
+
+logger = logging.getLogger("shagua.inactivity")
+
+
+class InactivityNotifier(Protocol):
+    channel: str
+
+    def warn(self, *, user_id: int, coin: int, cash_cents: int,
+             stage: int, days_until_reset: int) -> str:
+        """发预警(只涉及会被清的金币 + 折算现金;邀请现金不清、不预警)。
+        返回状态:'sent' / 'failed' / 'placeholder'。"""
+        ...
+
+
+class LogNotifier:
+    """占位实现:只打印,不真推。参照 heartbeat_monitor_worker「本期先不接推送」先例。"""
+
+    channel = "log"
+
+    def warn(self, *, user_id: int, coin: int, cash_cents: int,
+             stage: int, days_until_reset: int) -> str:
+        logger.warning(
+            "[inactivity-warn] user=%s coin=%s cash_cents=%s stage=T-%s days_until_reset=%s",
+            user_id, coin, cash_cents, stage, days_until_reset,
+        )
+        return "placeholder"
+
+
+def get_notifier(channel: str) -> InactivityNotifier:
+    """按配置返回通知器。未实现的通道(jpush/sms)暂回退 LogNotifier 占位。"""
+    # 后续:if channel == "jpush": return JPushNotifier()
+    #       if channel == "sms":   return SmsNotifier()
+    return LogNotifier()

app/main.py

@@ -49,6 +49,10 @@ from app.core.heartbeat_monitor_worker import (
     start_heartbeat_monitor,
     stop_heartbeat_monitor,
 )
+from app.core.inactivity_reset_worker import (
+    start_inactivity_reset_worker,
+    stop_inactivity_reset_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 (
@@ -80,12 +84,14 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     reconcile_task = start_withdraw_reconcile_worker()
     heartbeat_task = start_heartbeat_monitor()
     daily_exchange_task = start_daily_exchange_worker()
+    inactivity_task = start_inactivity_reset_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_inactivity_reset_worker(inactivity_task)
         await aclose_pricebot_client()
         logger.info("shutting down")
 

app/models/__init__.py

@@ -27,6 +27,10 @@ from app.models.coupon_state import (  # noqa: F401
     CouponSession,
 )
 from app.models.feedback import Feedback  # noqa: F401
+from app.models.inactivity import (  # noqa: F401
+    InactivityNotificationLog,
+    InactivityResetLog,
+)
 from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
 from app.models.launch_confirm_sample import LaunchConfirmSample  # noqa: F401

app/models/analytics_event.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 
 from datetime import datetime
 
-from sqlalchemy import JSON, BigInteger, DateTime, Integer, String, func
+from sqlalchemy import JSON, BigInteger, DateTime, Index, Integer, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -23,6 +23,12 @@ from app.db.base import Base
 
 class AnalyticsEvent(Base):
     __tablename__ = "analytics_event"
+    __table_args__ = (
+        # 活跃口径聚合热点(activity.active_event_condition + last_active_subqueries):
+        # 按 (event,page) 过滤 首页可见(show/home)∪比价∪领券,再 group by user_id 取
+        # max(created_at)。覆盖索引 → 该聚合走 index-only,避免高频 show 事件全表扫。
+        Index("ix_analytics_event_active", "event", "page", "user_id", "created_at"),
+    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
 

app/models/inactivity.py

@@ -0,0 +1,58 @@
+"""15 天不活跃清零相关表。
+
+- inactivity_reset_log:每次清零一行,记清零前三桶余额快照 + 原因 + 判定时活跃时间/不活跃天数,
+  供纠纷排查(需求①)。清零同时另写 2 条钱包流水(金币 + 折算现金,biz_type=inactivity_reset),
+  资金流可逐笔回溯。**邀请现金是产品红线、不清零**,invite_cash_balance_cents_before 仅为清零时
+  仍保留的邀请现金快照(便于排查、非被清金额;见 wallet.CoinAccount 注释)。
+- inactivity_notification_log:每次预警一行,记推送时余额快照 + 档位 + 通道 + 状态,
+  兼作"预警去重"依据(created_at > last_active)与"待推送"占位 outbox(v1 通道=log)。
+
+append-only,不更新。user_id 只索引、不设外键(同 analytics_event,避免删用户级联/历史留痕)。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+
+class InactivityResetLog(Base):
+    __tablename__ = "inactivity_reset_log"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
+    coin_balance_before: Mapped[int] = mapped_column(Integer, nullable=False)
+    cash_balance_cents_before: Mapped[int] = mapped_column(Integer, nullable=False)
+    invite_cash_balance_cents_before: Mapped[int] = mapped_column(Integer, nullable=False)
+    last_active_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    inactive_days: Mapped[int] = mapped_column(Integer, nullable=False)
+    reason: Mapped[str] = mapped_column(String(32), nullable=False)
+    reset_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<InactivityResetLog id={self.id} user_id={self.user_id} coin={self.coin_balance_before}>"
+
+
+class InactivityNotificationLog(Base):
+    __tablename__ = "inactivity_notification_log"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    user_id: Mapped[int] = mapped_column(Integer, index=True, nullable=False)
+    stage: Mapped[int] = mapped_column(Integer, nullable=False)      # 提前天数档(如 7 / 2)
+    inactive_days: Mapped[int] = mapped_column(Integer, nullable=False)
+    coin_balance: Mapped[int] = mapped_column(Integer, nullable=False)
+    cash_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False)
+    invite_cash_balance_cents: Mapped[int] = mapped_column(Integer, nullable=False)
+    channel: Mapped[str] = mapped_column(String(16), nullable=False)   # log / jpush / sms
+    status: Mapped[str] = mapped_column(String(16), nullable=False)    # placeholder / sent / failed
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"<InactivityNotificationLog id={self.id} user_id={self.user_id} stage={self.stage}>"

app/repositories/activity.py

@@ -0,0 +1,101 @@
+"""活跃口径唯一真源:worker(不活跃清零)与 admin(最近活跃/DAU)共用,防两处漂移。
+
+口径 = max(User.created_at, AnalyticsEvent[首页可见 show/home + 比价 + 领券], CouponPromptEngagement[claim_started])。
+**不含 last_login_at**(登录/re-login 不代表在用 App);created_at 为恒非空基线。
+清零/预警按北京自然日 0 点对齐(见 reset_cutoff)。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta, timezone
+
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session
+
+from app.core.rewards import CN_TZ, cn_today
+from app.models.analytics_event import AnalyticsEvent
+from app.models.coupon_state import CouponPromptEngagement
+
+# —— 活跃口径事件(与"用户管理"口径一致)——
+# 首页可见:前端埋点 event=show + page=home(组合判定,单个 event 名不足以区分,见
+# active_event_condition);其余为纯 event 名。
+HOME_VIEW_EVENT = "show"
+HOME_VIEW_PAGE = "home"
+COMPARE_START_EVENT = "real_compare_start"   # 发起比价(含浮窗触发)
+COUPON_START_EVENT = "real_coupon_start"     # 发起领券
+# 纯 event 名即可判定的活跃事件(首页可见是 event+page 组合、不在此列)
+ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
+ACTIVE_ENGAGE_TYPE = "claim_started"         # coupon_prompt_engagement 一键领取
+
+
+def active_event_condition():
+    """analytics_event 中算"活跃"的行为过滤:首页可见(event=show & page=home)
+    ∪ 发起比价 ∪ 发起领券。worker 子查询与 admin 展示共用,单一真源。"""
+    return or_(
+        and_(AnalyticsEvent.event == HOME_VIEW_EVENT, AnalyticsEvent.page == HOME_VIEW_PAGE),
+        AnalyticsEvent.event.in_(ACTIVE_EVENTS),
+    )
+
+
+def as_utc(value: datetime) -> datetime:
+    """任意 datetime → tz-aware UTC(无时区按 UTC 解释)。用于与 DateTime(timezone=True) 列比较,
+    比较绝对时刻、与会话时区无关(口径同 admin queries._as_utc)。"""
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def norm_utc(dt: datetime | None) -> datetime | None:
+    """naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
+    if dt is None:
+        return None
+    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
+
+
+def cn_midnight_utc(d: date) -> datetime:
+    """北京 d 日 00:00 → tz-aware UTC datetime。"""
+    return as_utc(datetime(d.year, d.month, d.day, tzinfo=CN_TZ))
+
+
+def reset_cutoff(reset_days: int, today: date | None = None) -> datetime:
+    """应清零边界(tz-aware UTC):last_active < 此值 ⟺ 距末次活跃已满 reset_days 天(北京 0 点对齐)。
+    = 北京 00:00 of (today − (reset_days − 1))。例:reset_days=15、today=1/20 → 北京 1/6 00:00。"""
+    today = today or cn_today()
+    return cn_midnight_utc(today - timedelta(days=reset_days - 1))
+
+
+def last_active_subqueries(db: Session):
+    """两个按 user_id 预聚合的派生表:最近活跃事件(见 active_event_condition)、
+    最近领券发起(claim_started)。返回 (ev_sub, eng_sub)。口径同 admin,LEFT JOIN 用。"""
+    ev_sub = (
+        select(
+            AnalyticsEvent.user_id.label("user_id"),
+            func.max(AnalyticsEvent.created_at).label("last_at"),
+        )
+        .where(AnalyticsEvent.user_id.is_not(None), active_event_condition())
+        .group_by(AnalyticsEvent.user_id)
+        .subquery()
+    )
+    eng_sub = (
+        select(
+            CouponPromptEngagement.user_id.label("user_id"),
+            func.max(CouponPromptEngagement.created_at).label("last_at"),
+        )
+        .where(
+            CouponPromptEngagement.user_id.is_not(None),
+            CouponPromptEngagement.engage_type == ACTIVE_ENGAGE_TYPE,
+        )
+        .group_by(CouponPromptEngagement.user_id)
+        .subquery()
+    )
+    return ev_sub, eng_sub
+
+
+def last_active_expr(base_col, ev_sub, eng_sub, dialect: str):
+    """max(base_col, 最近活跃事件, 最近领券) 的 SQL 表达式。PG 用 greatest、SQLite 用 max。
+    子聚合缺失(未命中)时 coalesce 到 base_col(= User.created_at,恒非空基线)。"""
+    greatest = func.greatest if dialect == "postgresql" else func.max
+    return greatest(
+        base_col,
+        func.coalesce(ev_sub.c.last_at, base_col),
+        func.coalesce(eng_sub.c.last_at, base_col),
+    )

app/repositories/inactivity.py

@@ -0,0 +1,183 @@
+"""15 天不活跃清零业务逻辑(纯同步,可单测)。worker 只是它的 asyncio 外壳。
+
+活跃口径复用 app.repositories.activity;清零走 wallet.grant_*(负数出账、写流水、不 commit)。
+逐用户独立事务,一个失败不影响其余。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date, datetime
+
+from sqlalchemy import or_, select
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+
+from app.core.rewards import CN_TZ
+from app.integrations.notifier import InactivityNotifier
+from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
+from app.models.user import User
+from app.models.wallet import CoinAccount
+from app.repositories import activity
+from app.repositories import wallet as wallet_repo
+
+logger = logging.getLogger("shagua.inactivity")
+
+RESET_BIZ_TYPE = "inactivity_reset"
+RESET_REMARK = "15天不活跃清零"
+
+# 清零候选口径:金币或折算现金有余额即入选。**邀请现金不算**——它是产品红线、不清零
+# (见 wallet.CoinAccount 注释),只有邀请现金余额的用户没有可清项,故不入选。
+_ANY_BALANCE = or_(
+    CoinAccount.coin_balance > 0,
+    CoinAccount.cash_balance_cents > 0,
+)
+
+
+def _base_query(db: Session):
+    """select(user_id, last_active, 三桶余额),join CoinAccount + 两活跃子查询。"""
+    ev_sub, eng_sub = activity.last_active_subqueries(db)
+    dialect = db.get_bind().dialect.name
+    last_active = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
+    stmt = (
+        select(
+            User.id.label("user_id"),
+            last_active.label("last_active"),
+            CoinAccount.coin_balance,
+            CoinAccount.cash_balance_cents,
+            CoinAccount.invite_cash_balance_cents,
+        )
+        .join(CoinAccount, CoinAccount.user_id == User.id)
+        .outerjoin(ev_sub, ev_sub.c.user_id == User.id)
+        .outerjoin(eng_sub, eng_sub.c.user_id == User.id)
+    )
+    return stmt, last_active
+
+
+def _cn_date(dt: datetime) -> date:
+    """datetime → 北京自然日(naive 视为 UTC)。"""
+    return activity.norm_utc(dt).astimezone(CN_TZ).date()
+
+
+def _inactive_days(last_active: datetime, today: date) -> int:
+    return (today - _cn_date(last_active)).days
+
+
+def select_inactive_users(db: Session, *, cutoff: datetime):
+    """应清零用户:last_active < cutoff 且金币/折算现金有余额(邀请现金不清、不计)。
+    返回 Row 列表(值已快照,可跨 commit)。"""
+    stmt, last_active = _base_query(db)
+    stmt = stmt.where(_ANY_BALANCE, last_active < activity.as_utc(cutoff))
+    return db.execute(stmt).all()
+
+
+def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int, reason: str) -> bool:
+    """单用户清零(独立事务、行锁)。金币 + 折算现金归零 + 写审计 + 2 条流水;**邀请现金不清**
+    (产品红线,见 wallet.CoinAccount 注释),仅作快照记入审计。返回是否真清了(有可清余额)。"""
+    acc = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
+    coin, cash, invite = acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents
+    if coin == 0 and cash == 0:  # 邀请现金不清,故不算"有可清余额"
+        return False
+    log = InactivityResetLog(
+        user_id=user_id, coin_balance_before=coin, cash_balance_cents_before=cash,
+        invite_cash_balance_cents_before=invite, last_active_at=activity.norm_utc(last_active),
+        inactive_days=inactive_days, reason=reason,
+    )
+    db.add(log)
+    db.flush()  # 拿 log.id 作 ref_id 交叉链接审计↔流水
+    ref = str(log.id)
+    if coin:
+        wallet_repo.grant_coins(db, user_id, -coin, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
+    if cash:
+        wallet_repo.grant_cash(db, user_id, -cash, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
+    # 邀请现金(invite_cash_balance_cents)刻意不动:两本账物理隔离、邀请金是产品红线。
+    db.commit()
+    return True
+
+
+def run_reset_once(db: Session, *, reset_days: int, today: date) -> dict:
+    """扫一轮清零。逐用户独立 commit,失败隔离。"""
+    stats = {"scanned": 0, "cleared": 0, "failed": 0}
+    cutoff = activity.reset_cutoff(reset_days, today)
+    reason = f"inactive_{reset_days}d"
+    rows = select_inactive_users(db, cutoff=cutoff)  # 先物化,避免边遍历边 commit
+    for row in rows:
+        stats["scanned"] += 1
+        idays = _inactive_days(row.last_active, today)
+        try:
+            if clear_user(db, user_id=row.user_id, last_active=row.last_active,
+                          inactive_days=idays, reason=reason):
+                stats["cleared"] += 1
+        except SQLAlchemyError:
+            db.rollback()
+            stats["failed"] += 1
+    return stats
+
+
+def select_warn_candidates(db: Session, *, clear_cutoff: datetime, warn_hi: datetime):
+    """预警候选:clear_cutoff <= last_active < warn_hi 且有可清余额(即已进预警窗、尚未到清零)。"""
+    stmt, last_active = _base_query(db)
+    stmt = stmt.where(
+        _ANY_BALANCE,
+        last_active >= activity.as_utc(clear_cutoff),
+        last_active < activity.as_utc(warn_hi),
+    )
+    return db.execute(stmt).all()
+
+
+def run_warn_once(db: Session, notifier: InactivityNotifier, *,
+                  reset_days: int, warn_stages: list[int], today: date) -> dict:
+    """扫一轮预警。每人取"最紧急的已到达档",按 streak 去重(notification_log.created_at > last_active)。
+    预警只涉及会被清的金币 + 折算现金;邀请现金不清、不预警(仅在 notification_log 记快照)。
+    逐用户 try/except 隔离:单用户通知器抛错 / DB 错不阻断其余,也绝不能拖累后续清零。"""
+    stats = {"warned": 0, "warn_skipped": 0, "warn_failed": 0}
+    if not warn_stages:
+        return stats
+    clear_cutoff = activity.reset_cutoff(reset_days, today)                # 到此即清零,不再预警
+    warn_hi = activity.reset_cutoff(reset_days - max(warn_stages), today)  # 最早预警档边界
+    ascending = sorted(warn_stages)                                       # 最紧急(最小 k)在前
+    for row in select_warn_candidates(db, clear_cutoff=clear_cutoff, warn_hi=warn_hi):
+        idays = _inactive_days(row.last_active, today)
+        stage = next((k for k in ascending if idays >= reset_days - k), None)
+        if stage is None:  # 防御:候选已在预警窗内、stage 必命中,此分支实际不可达
+            continue
+        try:
+            already = db.execute(
+                select(InactivityNotificationLog.id).where(
+                    InactivityNotificationLog.user_id == row.user_id,
+                    InactivityNotificationLog.stage == stage,
+                    InactivityNotificationLog.created_at > activity.as_utc(row.last_active),
+                ).limit(1)
+            ).first()
+            if already:
+                stats["warn_skipped"] += 1
+                continue
+            status = notifier.warn(
+                user_id=row.user_id, coin=row.coin_balance, cash_cents=row.cash_balance_cents,
+                stage=stage, days_until_reset=reset_days - idays,
+            )
+            db.add(InactivityNotificationLog(
+                user_id=row.user_id, stage=stage, inactive_days=idays,
+                coin_balance=row.coin_balance, cash_balance_cents=row.cash_balance_cents,
+                invite_cash_balance_cents=row.invite_cash_balance_cents,  # 快照,不参与"将清"额度
+                channel=notifier.channel, status=status,
+            ))
+            db.commit()
+            stats["warned"] += 1
+        except Exception:  # noqa: BLE001 - 单用户预警失败(通知器抛错/DB 错)隔离,不阻断其余、不拖累清零
+            db.rollback()
+            stats["warn_failed"] += 1
+    return stats
+
+
+def run_once(db: Session, *, notifier: InactivityNotifier, reset_days: int,
+             warn_stages: list[int], today: date) -> dict:
+    """一轮完整任务:先预警(阶段 A)再清零(阶段 B)。返回合并统计。
+    预警整段异常也**绝不阻塞清零**——清零是核心、不可逆资金操作,不能被通知故障拖住。"""
+    try:
+        warn = run_warn_once(db, notifier, reset_days=reset_days, warn_stages=warn_stages, today=today)
+    except Exception:  # noqa: BLE001 - 预警阶段整体失败(如候选查询失败)也要继续清零
+        logger.exception("inactivity warn phase failed; proceeding to reset")
+        db.rollback()
+        warn = {"warned": 0, "warn_skipped": 0, "warn_failed": 0, "warn_phase_error": 1}
+    reset = run_reset_once(db, reset_days=reset_days, today=today)
+    return {**warn, **reset}