diff --git a/app/core/config.py b/app/core/config.py index 5e88d11..bc3d1c7 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -178,8 +178,10 @@ 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 # 总闸,默认关;灰度验证后再开 + # === 15 天不活跃清零(app.core.inactivity_reset_worker,worker 常驻)=== + # ENABLED 只决定是否**真清**:false(默认)= 只记审计名单、不动钱(dry-run,灰度看名单); + # true = 真清金币 + 折算现金(邀请金不清)。看准名单后再置 true。 + 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) diff --git a/app/core/inactivity_reset_worker.py b/app/core/inactivity_reset_worker.py index 7296d59..b7e7425 100644 --- a/app/core/inactivity_reset_worker.py +++ b/app/core/inactivity_reset_worker.py @@ -6,7 +6,8 @@ 健壮性: - **逐用户幂等**:清完余额=0 次日不再匹配;预警按 streak 去重。启动补跑 / 多次唤醒 / 重启都安全。 - **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑。 -- **开关**:settings.INACTIVITY_RESET_ENABLED=false 时不启动(默认关,灰度验证后再开)。 +- **常驻 + dry-run 默认**:worker 一直跑;INACTIVITY_RESET_ENABLED=false(默认)只记审计名单、 + 不动钱(dry-run 灰度看名单),=true 才真清。 ⚠️ 这是不可逆批量资金操作(清空金币 + 折算现金,**邀请现金不清**)。口径见 app.repositories.inactivity / app.repositories.activity。 @@ -87,6 +88,7 @@ def _run_once_entry() -> dict: reset_days=settings.INACTIVITY_RESET_DAYS, warn_stages=settings.inactivity_warn_stages, today=_cn_today(), + dry_run=not settings.INACTIVITY_RESET_ENABLED, # ENABLED=false → 只记审计名单、不清 ) @@ -102,9 +104,10 @@ async def _run_loop() -> None: async def _run_locked_loop(interval: int) -> None: logger.info( - "inactivity reset worker started interval=%ss run_hour=%s", + "inactivity reset worker started interval=%ss run_hour=%s mode=%s", interval, settings.INACTIVITY_RESET_RUN_HOUR, + "clear" if settings.INACTIVITY_RESET_ENABLED else "dry-run(audit-only)", ) # 本进程上次跑过的北京日;None=尚未跑过本进程(当天到点即补)。 last_run: date | None = None @@ -129,9 +132,8 @@ async def _run_locked_loop(interval: int) -> None: 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 + # worker 常驻(不再有"完全关"档);INACTIVITY_RESET_ENABLED 只决定是否**真清**: + # false(默认)= 只记审计名单(dry-run,不动钱),true = 真清金币+现金。 return asyncio.create_task(_run_loop(), name="inactivity-reset") diff --git a/app/repositories/inactivity.py b/app/repositories/inactivity.py index 71970cc..59335de 100644 --- a/app/repositories/inactivity.py +++ b/app/repositories/inactivity.py @@ -70,13 +70,24 @@ def select_inactive_users(db: Session, *, cutoff: datetime): return db.execute(stmt).all() -def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int, reason: str) -> bool: +def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int, + reason: str, dry_run: bool = False) -> bool: """单用户清零(独立事务、行锁)。金币 + 折算现金归零 + 写审计 + 2 条流水;**邀请现金不清** - (产品红线,见 wallet.CoinAccount 注释),仅作快照记入审计。返回是否真清了(有可清余额)。""" + (产品红线,见 wallet.CoinAccount 注释),仅作快照记入审计。返回是否真处理了(有可清余额)。 + + dry_run=True:**只写审计名单、不动钱不写流水**(灰度看名单)。按 streak 去重——本 streak + 已记过(reset_at > last_active)就跳,避免 worker 每日重复记。""" 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 + if dry_run and db.execute( + select(InactivityResetLog.id).where( + InactivityResetLog.user_id == user_id, + InactivityResetLog.reset_at > activity.as_utc(last_active), + ).limit(1) + ).first(): + return False # dry-run:本 streak 已记过审计,不重复记 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), @@ -84,28 +95,29 @@ def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_day ) 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)刻意不动:两本账物理隔离、邀请金是产品红线。 + if not dry_run: # dry-run 只记审计名单,不真出账 + 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,失败隔离。""" +def run_reset_once(db: Session, *, reset_days: int, today: date, dry_run: bool = False) -> dict: + """扫一轮清零。逐用户独立 commit,失败隔离。dry_run=True 只记审计名单、不动钱(见 clear_user)。""" stats = {"scanned": 0, "cleared": 0, "failed": 0} cutoff = activity.reset_cutoff(reset_days, today) - reason = f"inactive_{reset_days}d" + reason = f"inactive_{reset_days}d" + ("_dryrun" if dry_run else "") 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): + inactive_days=idays, reason=reason, dry_run=dry_run): stats["cleared"] += 1 except SQLAlchemyError: db.rollback() @@ -170,14 +182,17 @@ def run_warn_once(db: Session, notifier: InactivityNotifier, *, def run_once(db: Session, *, notifier: InactivityNotifier, reset_days: int, - warn_stages: list[int], today: date) -> dict: + warn_stages: list[int], today: date, dry_run: bool = False) -> 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) + 预警整段异常也**绝不阻塞清零**——清零是核心、不可逆资金操作,不能被通知故障拖住。 + dry_run=True(灰度默认):只记审计名单、不清、**也不预警**(不通知一个不会发生的清零)。""" + warn = {"warned": 0, "warn_skipped": 0, "warn_failed": 0} + if not dry_run: + 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, dry_run=dry_run) return {**warn, **reset} diff --git a/docs/superpowers/specs/2026-07-16-inactivity-reset-design.md b/docs/superpowers/specs/2026-07-16-inactivity-reset-design.md index 3df388d..98de060 100644 --- a/docs/superpowers/specs/2026-07-16-inactivity-reset-design.md +++ b/docs/superpowers/specs/2026-07-16-inactivity-reset-design.md @@ -130,7 +130,7 @@ last_active = max( ## 6. 清零 worker `app/core/inactivity_reset_worker.py`（新建） -**完全仿 [`app/core/daily_exchange_worker.py`](../../../app/core/daily_exchange_worker.py)**：App 启动自带 asyncio 任务，文件锁（`data/inactivity_reset.lock`）防同机多进程并发，总闸 `INACTIVITY_RESET_ENABLED`（默认关）。 +**完全仿 [`app/core/daily_exchange_worker.py`](../../../app/core/daily_exchange_worker.py)**：App 启动自带 asyncio 任务，文件锁（`data/inactivity_reset.lock`）防同机多进程并发。**worker 常驻**；`INACTIVITY_RESET_ENABLED` 只决定是否**真清**：false（默认）= 只记审计名单、不动钱（dry-run），true = 真清。 ### 调度 @@ -194,7 +194,7 @@ class InactivityNotifier(Protocol): ## 8. 配置项（`app/core/config.py`） ``` -INACTIVITY_RESET_ENABLED = False # 总闸，默认关；灰度验证后再开 +INACTIVITY_RESET_ENABLED = False # false(默认)=只记审计名单(dry-run,不动钱);true=真清 INACTIVITY_RESET_DAYS = 15 # 不活跃阈值（天） INACTIVITY_WARN_DAYS_BEFORE = "7,2" # 清零前几天各推一次；空串=不推。逗号分隔，降序解析 INACTIVITY_RESET_RUN_HOUR = 3 # 北京时间每日执行点（0-23） @@ -225,7 +225,7 @@ INACTIVITY_RESET_CHECK_INTERVAL_SEC = 1800 # worker 唤醒间隔（可复用现 | 在途提现 | 提现申请时现金已扣入 `WithdrawOrder`，当前余额已不含在途；只清当前余额、不动提现单。提现失败退款到已清账户 = 用户的钱，正常 | | 与 `daily_auto_exchange` 并存 | 各自逐用户幂等；金币多已日结折现金，三桶全清正好覆盖 | | 时区/日界 | 统一北京（`rewards.cn_today()`/`CN_TZ`）；**清零/预警按北京自然日 0 点对齐**（末次活跃记为第 1 日 → 第 16 日 0 点清零，见 §4），非滚动 24h；流水 `created_at` 沿用北京 wall-clock naive | -| 误清防护 | 总闸默认关；先灰度只看预警/清零**名单**对不对，再开清零（§13） | +| 误清防护 | worker 常驻默认 **dry-run**（`ENABLED=false` 只记审计名单、不动钱）；看准名单再置 `true` 真清（§13） | --- @@ -233,7 +233,7 @@ INACTIVITY_RESET_CHECK_INTERVAL_SEC = 1800 # worker 唤醒间隔（可复用现 - **Android 端**（`shaguabijia-app-android`）需在**首页可见**（`onResume`/Tab 切入）时，向现有 `POST /api/v1/analytics/events` 批量上报里加一条 `event=<首页可见事件名>`（名称明天加埋点时定，暂记 `"home_view"`） 的事件，**携带登录后的 `user_id`**。 - 客户端按会话/前台去重即可（服务端只取 `max(created_at)`，多报无害）。 -- **上线顺序依赖**：`home_view` 全量覆盖前，"进首页"信号缺失，只有比价/领券能推进活跃、其余落到 `created_at` 基线（"只开首页不操作"且注册满 15 天的用户会被误清）—— 故清零总闸必须待 `home_view` 铺满后再开（§13）。 +- **上线顺序依赖**：`home_view` 全量覆盖前，"进首页"信号缺失，只有比价/领券能推进活跃、其余落到 `created_at` 基线（"只开首页不操作"且注册满 15 天的用户会被误清）—— 故**开真清（`ENABLED=true`）必须待 `home_view` 铺满后再开**（§13）；dry-run 只记名单不动钱、可先开着看。 --- @@ -250,8 +250,8 @@ INACTIVITY_RESET_CHECK_INTERVAL_SEC = 1800 # worker 唤醒间隔（可复用现 1. **后端先行**：合入共享模块 + 两表 + worker + 通知器，`INACTIVITY_RESET_ENABLED=False`；活跃口径以 `created_at` 为非空基线、**不含 last_login_at**。 2. **Android 发版**：上报 `home_view`；观察 analytics 覆盖率。 -3. **只读灰度**：临时开一个"dry-run/只出名单不清零"路径或看阶段 A 预警日志，核对预警/清零**名单**准确。 -4. **开清零**：确认无误后置 `INACTIVITY_RESET_ENABLED=True`。 +3. **dry-run 灰度（默认即是）**：`INACTIVITY_RESET_ENABLED=False` 时 worker 常驻只写审计名单（`reason=inactive_Nd_dryrun`）、不动钱、不预警；核对名单准确。 +4. **开真清**：确认无误后置 `INACTIVITY_RESET_ENABLED=True`（转为真清 + 预警）。 5. **收尾/监控**：持续观察 `home_view` 覆盖率与预警/清零名单；发现"活跃却被判不活跃"的漏报即回查埋点覆盖（口径已不含 last_login_at，登录不再兜底）。 --- @@ -263,7 +263,7 @@ INACTIVITY_RESET_CHECK_INTERVAL_SEC = 1800 # worker 唤醒间隔（可复用现 - **不活跃判定**：`last_active` 分别 `<15d / =15d / >15d` × 有/无余额 的命中矩阵。 - **清零**：三桶归零；`inactivity_reset_log` 清前值正确；三条流水 `biz_type=inactivity_reset`、`balance_after=0`、`ref_id=log.id`；`total_coin_earned` 不变。 - **预警**：命中窗口调 notifier + 写 `notification_log`；同 streak 不重推；回归后 `last_active` 前移可再次预警；漏跑补发最紧急档。 -- **worker**：总闸关不启动；文件锁互斥；逐用户失败隔离（一个抛错不影响其余，`failed` 计数）；重复跑幂等。 +- **worker**：常驻；`ENABLED=false` 走 dry-run（只记审计名单、不清、不预警）；文件锁互斥；逐用户失败隔离（一个抛错不影响其余，`failed` 计数）；重复跑幂等。 - **配置**：`INACTIVITY_WARN_DAYS_BEFORE` 解析（含空串=不推）；`RESET_DAYS`/`RUN_HOUR` 生效。 - 沿用 `tests/conftest.py`（临时 SQLite、`RATE_LIMIT_ENABLED=false`）；外部通知 monkeypatch。 diff --git a/tests/test_inactivity_reset.py b/tests/test_inactivity_reset.py index 6860d2c..33fb217 100644 --- a/tests/test_inactivity_reset.py +++ b/tests/test_inactivity_reset.py @@ -302,11 +302,35 @@ def test_run_once_warns_then_resets() -> None: db.close() -def test_worker_disabled_returns_none(monkeypatch) -> None: +def test_worker_run_once_entry_dry_run(monkeypatch) -> None: + """ENABLED=false(默认语义)→ worker 常驻但只记审计不清(dry_run = not ENABLED)。""" + from sqlalchemy import select + from app.core import inactivity_reset_worker as w from app.core.config import settings - monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", False) - assert w.start_inactivity_reset_worker() is None + from app.models.wallet import CoinAccount + + monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", False) # false = 只记审计 + monkeypatch.setattr(settings, "INACTIVITY_RESET_DAYS", 15) + monkeypatch.setattr(settings, "INACTIVITY_WARN_DAYS_BEFORE", "") + monkeypatch.setattr(w, "_cn_today", lambda: date(2026, 2, 1)) + + db = SessionLocal() + try: + uid = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=100) + db.commit() + finally: + db.close() + + w._run_once_entry() + + db = SessionLocal() + try: + assert db.get(CoinAccount, uid).coin_balance == 100 # 没清 + log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == uid)).scalar_one() + assert log.reason.endswith("dryrun") # 记了审计 + finally: + db.close() def test_worker_run_once_entry_executes(monkeypatch) -> None: @@ -338,6 +362,44 @@ def test_worker_run_once_entry_executes(monkeypatch) -> None: db.close() +def test_run_once_dry_run_records_audit_but_does_not_clear() -> None: + """dry-run:只写审计(标 dryrun)、不动钱、不预警;重复跑不重复记(streak dedup)。""" + from sqlalchemy import select + + from app.integrations.notifier import LogNotifier + from app.models.wallet import CoinAccount, CoinTransaction + from app.repositories import inactivity + + db = SessionLocal() + try: + today = date(2026, 2, 1) + old = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc), coin=100, cash=200, invite=300) + warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=50) # 预警窗 + db.commit() + + stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, + warn_stages=[7, 2], today=today, dry_run=True) + acc = db.get(CoinAccount, old) + assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (100, 200, 300) # 原封 + log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one() + assert log.coin_balance_before == 100 and log.reason.endswith("dryrun") # 审计标 dryrun + assert db.execute(select(CoinTransaction).where( + CoinTransaction.user_id == old, CoinTransaction.biz_type == "inactivity_reset")).first() is None # 无流水 + assert stats["warned"] == 0 # dry-run 不预警 + assert db.execute(select(InactivityNotificationLog).where( + InactivityNotificationLog.user_id == warn_uid)).first() is None + assert stats["cleared"] == 1 # dry-run:cleared=记了几条 + + # 再跑一次 → 不重复记(dedup),余额仍原封 + inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today, dry_run=True) + assert len(db.execute(select(InactivityResetLog).where( + InactivityResetLog.user_id == old)).scalars().all()) == 1 + assert db.get(CoinAccount, old).coin_balance == 100 + finally: + db.rollback() + db.close() + + def test_admin_list_users_last_active_ignores_login() -> None: """admin 用户列表 last_active_at 改用共享口径:登录不算活跃(baseline=created_at)、只认活跃事件。""" from app.admin.repositories import queries