From 23e4671312e1cd14a06e26829e26134bf8241fcd Mon Sep 17 00:00:00 2001 From: guke Date: Thu, 16 Jul 2026 18:16:47 +0800 Subject: [PATCH] =?UTF-8?q?docs(welfare):=2015=E5=A4=A9=E4=B8=8D=E6=B4=BB?= =?UTF-8?q?=E8=B7=83=E6=B8=85=E9=9B=B6=20=E5=AE=9E=E7=8E=B0=E8=AE=A1?= =?UTF-8?q?=E5=88=92(11=20=E4=BB=BB=E5=8A=A1,=20TDD=20=E5=88=86=E8=A7=A3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从已定稿 spec 拆出可执行计划:模型/迁移/活跃口径共享模块/配置/通知器/清零+预警业务逻辑/ 每日 worker/admin 重构/表字典。每任务含失败测试→实现→通过→提交的 bite-sized 步骤,含完整代码。 活跃口径 = created_at 基线 + home_view/比价/领券(不含 last_login_at);home_view 事件名以 HOME_VIEW_EVENT 常量占位(前端明天定名);总闸默认关。 Co-Authored-By: Claude Opus 4.8 --- .../plans/2026-07-16-inactivity-reset.md | 1394 +++++++++++++++++ 1 file changed, 1394 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-16-inactivity-reset.md diff --git a/docs/superpowers/plans/2026-07-16-inactivity-reset.md b/docs/superpowers/plans/2026-07-16-inactivity-reset.md new file mode 100644 index 0000000..648ab49 --- /dev/null +++ b/docs/superpowers/plans/2026-07-16-inactivity-reset.md @@ -0,0 +1,1394 @@ +# 15 天不活跃清零(金币/现金) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 连续 15 天不活跃(北京自然日、第 16 日 0 点对齐)的用户,自动清零金币+折算现金+邀请现金,清零前按可配置节奏预警,全程留审计。 + +**Architecture:** 活跃口径抽成共享模块 `app/repositories/activity.py`(worker 与 admin 共用,单一真源);业务逻辑在 `app/repositories/inactivity.py`(纯同步、可单测);进程内每日 asyncio worker `app/core/inactivity_reset_worker.py`(仿 `daily_exchange_worker`,文件锁 + 北京日守卫 + 总闸)。预警走可插拔 `InactivityNotifier`,v1 为日志占位。审计双写:`inactivity_reset_log` 专表 + 钱包流水 `biz_type=inactivity_reset`。 + +**Tech Stack:** FastAPI · SQLAlchemy 2.0 (`Mapped`) · Alembic (`render_as_batch`) · pydantic-settings · pytest + TestClient/SQLite。 + +**Spec:** `docs/superpowers/specs/2026-07-16-inactivity-reset-design.md` + +**活跃口径(最终):** `last_active = max(User.created_at, AnalyticsEvent[home_view/real_compare_start/real_coupon_start], CouponPromptEngagement[claim_started])`。**不含 last_login_at**;`created_at` 为恒非空基线。`home_view` 事件名前端明天敲定,后端以常量 `HOME_VIEW_EVENT` 占位(暂 `"home_view"`)。 + +**清零边界:** 末次活跃日记为「第 1 日」→ 第 16 日 0 点(北京)清零 = `北京 00:00 of (last_active_date + RESET_DAYS)`。等价 `应清零 ⟺ last_active < reset_cutoff = 北京 00:00 of (cn_today() − (RESET_DAYS−1))`。 + +--- + +## File Structure + +**新增** +- `app/models/inactivity.py` — `InactivityResetLog` + `InactivityNotificationLog`(两个小关联模型同文件,仿 `wallet.py` 多模型同文件) +- `app/repositories/activity.py` — 活跃口径唯一真源(常量 + tz 助手 + cutoff + 子查询 + `last_active_expr`) +- `app/integrations/notifier.py` — `InactivityNotifier` 协议 + `LogNotifier` + `get_notifier` +- `app/repositories/inactivity.py` — 选取/清零/预警业务逻辑(纯同步) +- `app/core/inactivity_reset_worker.py` — 每日 worker(asyncio + 文件锁 + 守卫 + 启停) +- `alembic/versions/_add_inactivity_tables.py` — 建两表迁移 +- `tests/test_inactivity_reset.py` — 单测/集成 + +**改动** +- `app/models/__init__.py` — 注册两模型 +- `app/core/config.py` — `INACTIVITY_*` 配置 + `inactivity_warn_stages` 属性 +- `app/main.py` — lifespan 接线 start/stop worker +- `app/admin/repositories/queries.py`、`app/admin/repositories/stats.py` — 改用 `activity.py`(移除本地重复口径) + +--- + +## Task 1: 两张新表模型 + 注册 + +**Files:** +- Create: `app/models/inactivity.py` +- Modify: `app/models/__init__.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +Create `tests/test_inactivity_reset.py`: + +```python +"""15 天不活跃清零:模型 / 活跃口径 / 清零 / 预警 / 配置 / worker。""" +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone + +from sqlalchemy import select + +from app.db.session import SessionLocal +from app.models.inactivity import InactivityNotificationLog, InactivityResetLog + + +def test_reset_and_notification_models_persist() -> None: + db = SessionLocal() + try: + db.add(InactivityResetLog( + user_id=1, coin_balance_before=10, cash_balance_cents_before=20, + invite_cash_balance_cents_before=30, + last_active_at=datetime(2026, 1, 1, tzinfo=timezone.utc), + inactive_days=15, reason="inactive_15d", + )) + db.add(InactivityNotificationLog( + user_id=1, stage=7, inactive_days=8, coin_balance=10, + cash_balance_cents=20, invite_cash_balance_cents=30, + channel="log", status="placeholder", + )) + db.commit() + r = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == 1)).scalar_one() + assert r.reason == "inactive_15d" and r.reset_at is not None + n = db.execute(select(InactivityNotificationLog).where(InactivityNotificationLog.user_id == 1)).scalar_one() + assert n.stage == 7 and n.created_at is not None + finally: + db.rollback() + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_reset_and_notification_models_persist -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.models.inactivity'` + +- [ ] **Step 3: Create the models** + +Create `app/models/inactivity.py`: + +```python +"""15 天不活跃清零相关表。 + +- inactivity_reset_log:每次清零一行,记清零前三桶余额 + 原因 + 判定时活跃时间/不活跃天数, + 供纠纷排查(需求①)。清零同时另写 3 条钱包流水(biz_type=inactivity_reset),资金流可逐笔回溯。 +- 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"" + + +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"" +``` + +- [ ] **Step 4: Register in models/__init__.py** + +In `app/models/__init__.py`, add after the `from app.models.invite import ...` line (keep alphabetical-ish grouping near other domain models): + +```python +from app.models.inactivity import ( # noqa: F401 + InactivityNotificationLog, + InactivityResetLog, +) +``` + +- [ ] **Step 5: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py::test_reset_and_notification_models_persist -q` +Expected: PASS (conftest builds all tables via `Base.metadata.create_all`, so the new tables exist in the SQLite test DB.) + +- [ ] **Step 6: Commit** + +```bash +git add app/models/inactivity.py app/models/__init__.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): inactivity_reset_log + inactivity_notification_log 模型" +``` + +--- + +## Task 2: Alembic 迁移建两表 + +**Files:** +- Create: `alembic/versions/_add_inactivity_tables.py` (由 autogenerate 生成文件名/revision) + +- [ ] **Step 1: Confirm single head** + +Run: `alembic heads` +Expected: 恰好一个 head(单行)。若多个 head,先 `alembic merge -m "merge heads" ` 再继续。 + +- [ ] **Step 2: Autogenerate the migration** + +Run: `alembic revision --autogenerate -m "add inactivity tables"` +Expected: 在 `alembic/versions/` 生成一个新文件,`down_revision` 自动指向当前 head。 + +- [ ] **Step 3: Replace the migration body** + +打开生成的文件,**只保留新两表**(如 autogenerate 顺带检出其它表的历史索引漂移,删掉那些无关 op,仿 `1699fc2c069f_add_analytics_event_table.py:50-51` 的做法)。`upgrade`/`downgrade` 改成: + +```python +def upgrade() -> None: + op.create_table( + "inactivity_reset_log", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("coin_balance_before", sa.Integer(), nullable=False), + sa.Column("cash_balance_cents_before", sa.Integer(), nullable=False), + sa.Column("invite_cash_balance_cents_before", sa.Integer(), nullable=False), + sa.Column("last_active_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("inactive_days", sa.Integer(), nullable=False), + sa.Column("reason", sa.String(length=32), nullable=False), + sa.Column("reset_at", sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("inactivity_reset_log", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_inactivity_reset_log_user_id"), ["user_id"], unique=False) + batch_op.create_index(batch_op.f("ix_inactivity_reset_log_reset_at"), ["reset_at"], unique=False) + + op.create_table( + "inactivity_notification_log", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.Integer(), nullable=False), + sa.Column("stage", sa.Integer(), nullable=False), + sa.Column("inactive_days", sa.Integer(), nullable=False), + sa.Column("coin_balance", sa.Integer(), nullable=False), + sa.Column("cash_balance_cents", sa.Integer(), nullable=False), + sa.Column("invite_cash_balance_cents", sa.Integer(), nullable=False), + sa.Column("channel", sa.String(length=16), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), + server_default=sa.text("(CURRENT_TIMESTAMP)"), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + with op.batch_alter_table("inactivity_notification_log", schema=None) as batch_op: + batch_op.create_index(batch_op.f("ix_inactivity_notification_log_user_id"), ["user_id"], unique=False) + batch_op.create_index(batch_op.f("ix_inactivity_notification_log_created_at"), ["created_at"], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table("inactivity_notification_log", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_inactivity_notification_log_created_at")) + batch_op.drop_index(batch_op.f("ix_inactivity_notification_log_user_id")) + op.drop_table("inactivity_notification_log") + with op.batch_alter_table("inactivity_reset_log", schema=None) as batch_op: + batch_op.drop_index(batch_op.f("ix_inactivity_reset_log_reset_at")) + batch_op.drop_index(batch_op.f("ix_inactivity_reset_log_user_id")) + op.drop_table("inactivity_reset_log") +``` + +确保文件顶部保留自动生成的 `import sqlalchemy as sa` / `from alembic import op` / `revision` / `down_revision`。 + +- [ ] **Step 4: Apply and verify round-trips** + +Run: `alembic upgrade head && alembic downgrade -1 && alembic upgrade head` +Expected: 三步都无错;`upgrade` 建表、`downgrade` 删表、再 `upgrade` 重建。 + +- [ ] **Step 5: Commit** + +```bash +git add alembic/versions/ +git commit -m "feat(welfare): 迁移新增 inactivity_reset_log / inactivity_notification_log 两表" +``` + +--- + +## Task 3: 活跃口径共享模块 — 常量 + tz 助手 + cutoff(纯函数) + +**Files:** +- Create: `app/repositories/activity.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +追加到 `tests/test_inactivity_reset.py`: + +```python +from app.repositories import activity + + +def test_reset_cutoff_is_cn_midnight_of_today_minus_days_minus_1() -> None: + # RESET_DAYS=15, today=1/20 → cutoff = 北京 00:00 of 1/6 = 1/5 16:00 UTC + cutoff = activity.reset_cutoff(15, today=date(2026, 1, 20)) + assert cutoff == datetime(2026, 1, 5, 16, 0, tzinfo=timezone.utc) + + +def test_active_event_constants() -> None: + assert activity.HOME_VIEW_EVENT in activity.ACTIVE_EVENTS + assert "real_compare_start" in activity.ACTIVE_EVENTS + assert "real_coupon_start" in activity.ACTIVE_EVENTS + assert activity.ACTIVE_ENGAGE_TYPE == "claim_started" + + +def test_as_utc_normalizes() -> None: + assert activity.as_utc(datetime(2026, 1, 1)) == datetime(2026, 1, 1, tzinfo=timezone.utc) + cn = datetime(2026, 1, 1, tzinfo=activity.CN_TZ) # 北京 0 点 = 前一天 16:00 UTC + assert activity.as_utc(cn) == datetime(2025, 12, 31, 16, 0, tzinfo=timezone.utc) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py -k "cutoff or constants or as_utc" -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.repositories.activity'` + +- [ ] **Step 3: Create the module (constants + helpers only for now)** + +Create `app/repositories/activity.py`: + +```python +"""活跃口径唯一真源:worker(不活跃清零)与 admin(最近活跃/DAU)共用,防两处漂移。 + +口径 = max(User.created_at, AnalyticsEvent[ACTIVE_EVENTS], 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 func, 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 + +# —— 活跃事件名(与"用户管理"口径一致)—— +# home_view:进首页(前端埋点,名称前端明天敲定,此处占位;定名后仅改这一常量)。 +HOME_VIEW_EVENT = "home_view" +COMPARE_START_EVENT = "real_compare_start" # 发起比价(含浮窗触发) +COUPON_START_EVENT = "real_coupon_start" # 发起领券 +ACTIVE_EVENTS = (HOME_VIEW_EVENT, COMPARE_START_EVENT, COUPON_START_EVENT) +ACTIVE_ENGAGE_TYPE = "claim_started" # coupon_prompt_engagement 一键领取 + + +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)) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py -k "cutoff or constants or as_utc" -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/repositories/activity.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 活跃口径共享模块-常量与北京0点cutoff助手" +``` + +--- + +## Task 4: 活跃口径共享模块 — 子查询 + last_active_expr(DB) + +**Files:** +- Modify: `app/repositories/activity.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +追加测试助手 + 用例到 `tests/test_inactivity_reset.py`: + +```python +from app.core.rewards import CN_TZ +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 import wallet as wallet_repo + +_PHONE_SEQ = [0] + + +def _new_user(db, *, created_at, coin=0, cash=0, invite=0) -> int: + """直接建一个 User + CoinAccount,created_at 可控。返回 user_id。""" + _PHONE_SEQ[0] += 1 + u = User(phone=f"139{_PHONE_SEQ[0]:08d}", created_at=created_at, + last_login_at=created_at, status="active") + db.add(u) + db.flush() + acc = wallet_repo.get_or_create_account(db, u.id, commit=False) + acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents = coin, cash, invite + acc.total_coin_earned = coin + db.flush() + return u.id + + +def _add_event(db, user_id, event, when: datetime) -> None: + db.add(AnalyticsEvent(event=event, device_id="d", user_id=user_id, client_ts=0, created_at=when)) + + +def _add_engage(db, user_id, when: datetime, engage_type="claim_started") -> None: + db.add(CouponPromptEngagement(device_id=f"dev{user_id}", package="p", user_id=user_id, + engage_date=when.date(), engage_type=engage_type, created_at=when)) + + +def test_last_active_expr_takes_max_of_baseline_and_events() -> None: + from sqlalchemy import select + db = SessionLocal() + try: + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + uid = _new_user(db, created_at=base, coin=5) + _add_event(db, uid, "real_compare_start", datetime(2026, 1, 10, tzinfo=timezone.utc)) + db.commit() + ev_sub, eng_sub = activity.last_active_subqueries(db) + dialect = db.get_bind().dialect.name + expr = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect) + stmt = (select(expr).select_from(User) + .outerjoin(ev_sub, ev_sub.c.user_id == User.id) + .outerjoin(eng_sub, eng_sub.c.user_id == User.id) + .where(User.id == uid)) + got = activity.norm_utc(db.execute(stmt).scalar_one()) + assert got == datetime(2026, 1, 10, tzinfo=timezone.utc) # 事件 > 基线 + finally: + db.rollback() + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_last_active_expr_takes_max_of_baseline_and_events -q` +Expected: FAIL — `AttributeError: module 'app.repositories.activity' has no attribute 'last_active_subqueries'` + +- [ ] **Step 3: Add subqueries + expr to activity.py** + +在 `app/repositories/activity.py` 末尾追加: + +```python +def last_active_subqueries(db: Session): + """两个按 user_id 预聚合的派生表:最近活跃事件(ACTIVE_EVENTS)、最近领券发起(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), AnalyticsEvent.event.in_(ACTIVE_EVENTS)) + .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), + ) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py::test_last_active_expr_takes_max_of_baseline_and_events -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/repositories/activity.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 活跃口径共享模块-子查询与 last_active_expr" +``` + +--- + +## Task 5: 配置项 + +**Files:** +- Modify: `app/core/config.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +追加: + +```python +def test_inactivity_warn_stages_parsing() -> None: + from app.core.config import Settings + s = Settings(INACTIVITY_WARN_DAYS_BEFORE="7,2", INACTIVITY_RESET_DAYS=15) + assert s.inactivity_warn_stages == [7, 2] # 降序去重 + s2 = Settings(INACTIVITY_WARN_DAYS_BEFORE="", INACTIVITY_RESET_DAYS=15) + assert s2.inactivity_warn_stages == [] # 空=不推 + s3 = Settings(INACTIVITY_WARN_DAYS_BEFORE="2,20,7,2", INACTIVITY_RESET_DAYS=15) + assert s3.inactivity_warn_stages == [7, 2] # 去重 + 丢弃 >=RESET_DAYS(20) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_inactivity_warn_stages_parsing -q` +Expected: FAIL — `TypeError: ... unexpected keyword argument 'INACTIVITY_WARN_DAYS_BEFORE'` + +- [ ] **Step 3: Add settings** + +在 `app/core/config.py` 的 `Settings` 类里,紧接 `AUTO_EXCHANGE_CHECK_INTERVAL_SEC: int = 600`(约 line 171)之后加: + +```python + # === 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 唤醒间隔(秒) +``` + +并在类内(与其它 `@property` 放一起,如 `wxpay_configured` 附近)加解析属性: + +```python + @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) +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py::test_inactivity_warn_stages_parsing -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/core/config.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): INACTIVITY_* 配置项 + warn stages 解析" +``` + +--- + +## Task 6: 可插拔通知器 + +**Files:** +- Create: `app/integrations/notifier.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +```python +def test_log_notifier_returns_placeholder(caplog) -> None: + from app.integrations.notifier import LogNotifier, get_notifier + n = get_notifier("log") + assert isinstance(n, LogNotifier) and n.channel == "log" + status = n.warn(user_id=1, coin=10, cash_cents=20, invite_cash_cents=30, stage=7, days_until_reset=8) + assert status == "placeholder" + # 未实现通道回退 LogNotifier(占位) + assert get_notifier("jpush").channel == "log" +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_log_notifier_returns_placeholder -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.integrations.notifier'` + +- [ ] **Step 3: Create the notifier** + +Create `app/integrations/notifier.py`: + +```python +"""不活跃预警通知器(可插拔)。 + +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, invite_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, invite_cash_cents: int, + stage: int, days_until_reset: int) -> str: + logger.warning( + "[inactivity-warn] user=%s coin=%s cash_cents=%s invite_cash_cents=%s " + "stage=T-%s days_until_reset=%s", + user_id, coin, cash_cents, invite_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() +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py::test_log_notifier_returns_placeholder -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/integrations/notifier.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 可插拔不活跃预警通知器 + LogNotifier 占位" +``` + +--- + +## Task 7: 业务逻辑 — 选取 + 清零 + +**Files:** +- Create: `app/repositories/inactivity.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +```python +def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None: + from sqlalchemy import select + from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction + from app.repositories import inactivity + + db = SessionLocal() + try: + today = date(2026, 2, 1) + # 末次活跃 = created_at 基线 = 1/10(距 today 22 天 → 应清) + old = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc), + coin=100, cash=200, invite=300) + # 活跃用户:昨天有 home_view → 不清 + fresh = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc), coin=50) + _add_event(db, fresh, "home_view", datetime(2026, 1, 31, tzinfo=timezone.utc)) + db.commit() + + stats = inactivity.run_reset_once(db, reset_days=15, today=today) + assert stats["cleared"] == 1 and stats["failed"] == 0 + + acc = db.get(CoinAccount, old) + assert (acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents) == (0, 0, 0) + assert acc.total_coin_earned == 100 # 历史累计不动 + + log = db.execute(select(InactivityResetLog).where(InactivityResetLog.user_id == old)).scalar_one() + assert (log.coin_balance_before, log.cash_balance_cents_before, + log.invite_cash_balance_cents_before) == (100, 200, 300) + assert log.inactive_days == 22 and log.reason == "inactive_15d" + + ct = db.execute(select(CoinTransaction).where( + CoinTransaction.user_id == old, CoinTransaction.biz_type == "inactivity_reset")).scalar_one() + assert ct.amount == -100 and ct.balance_after == 0 and ct.ref_id == str(log.id) + assert db.execute(select(CashTransaction).where( + CashTransaction.user_id == old, CashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -200 + assert db.execute(select(InviteCashTransaction).where( + InviteCashTransaction.user_id == old, InviteCashTransaction.biz_type == "inactivity_reset")).scalar_one().amount_cents == -300 + + # 活跃用户不动;再跑一次幂等(已清零 → 不再匹配) + assert db.get(CoinAccount, fresh).coin_balance == 50 + assert inactivity.run_reset_once(db, reset_days=15, today=today)["cleared"] == 0 + finally: + db.rollback() + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_run_reset_clears_three_buckets_and_writes_audit_and_flows -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.repositories.inactivity'` + +- [ ] **Step 3: Create the repository (selection + clear + run_reset_once)** + +Create `app/repositories/inactivity.py`: + +```python +"""15 天不活跃清零业务逻辑(纯同步,可单测)。worker 只是它的 asyncio 外壳。 + +活跃口径复用 app.repositories.activity;清零走 wallet.grant_*(负数出账、写流水、不 commit)。 +逐用户独立事务,一个失败不影响其余。 +""" +from __future__ import annotations + +from datetime import date, datetime, timezone + +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 + +RESET_BIZ_TYPE = "inactivity_reset" +RESET_REMARK = "15天不活跃清零" + +_ANY_BALANCE = or_( + CoinAccount.coin_balance > 0, + CoinAccount.cash_balance_cents > 0, + CoinAccount.invite_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: + """单用户清零(独立事务、行锁)。三桶归零 + 写审计 + 3 条流水。返回是否真清了(有余额)。""" + 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 and invite == 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) + if invite: + wallet_repo.grant_invite_cash(db, user_id, -invite, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK) + 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 +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py::test_run_reset_clears_three_buckets_and_writes_audit_and_flows -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/repositories/inactivity.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 不活跃清零-选取与逐用户清零(三桶归零+审计+流水)" +``` + +--- + +## Task 8: 业务逻辑 — 预警 + run_once 组合 + +**Files:** +- Modify: `app/repositories/inactivity.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +```python +def test_run_warn_picks_stage_and_dedups_within_streak() -> None: + from app.integrations.notifier import LogNotifier + from app.repositories import inactivity + + db = SessionLocal() + try: + today = date(2026, 2, 1) + # 末次活跃 1/22(距 today 10 天)→ 档 7 命中(idays>=8),档 2 未到(需>=13) + uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=100) + db.commit() + + stats = inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today) + assert stats["warned"] == 1 + from sqlalchemy import select + rows = db.execute(select(InactivityNotificationLog).where( + InactivityNotificationLog.user_id == uid)).scalars().all() + assert len(rows) == 1 and rows[0].stage == 7 and rows[0].status == "placeholder" + assert rows[0].inactive_days == 10 and rows[0].coin_balance == 100 + + # 同一 streak 再跑 → 不重推 + assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0 + + # 无余额用户不预警 + _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=0) + db.commit() + assert inactivity.run_warn_once(db, LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today)["warned"] == 0 + finally: + db.rollback() + db.close() + + +def test_run_once_warns_then_resets() -> None: + from app.integrations.notifier import LogNotifier + from app.repositories import inactivity + db = SessionLocal() + try: + today = date(2026, 2, 1) + warn_uid = _new_user(db, created_at=datetime(2026, 1, 22, tzinfo=timezone.utc), coin=10) # 10天→预警 + clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10) # 27天→清零 + db.commit() + stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, warn_stages=[7, 2], today=today) + assert stats["warned"] == 1 and stats["cleared"] == 1 + from app.models.wallet import CoinAccount + assert db.get(CoinAccount, clear_uid).coin_balance == 0 + assert db.get(CoinAccount, warn_uid).coin_balance == 10 # 预警不动钱 + finally: + db.rollback() + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py -k "run_warn or run_once" -q` +Expected: FAIL — `AttributeError: module 'app.repositories.inactivity' has no attribute 'run_warn_once'` + +- [ ] **Step 3: Add warn + run_once to inactivity.py** + +在 `app/repositories/inactivity.py` 末尾追加: + +```python +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)。""" + stats = {"warned": 0, "warn_skipped": 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: + continue + 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, + invite_cash_cents=row.invite_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 + return stats + + +def run_once(db: Session, *, notifier: InactivityNotifier, reset_days: int, + warn_stages: list[int], today: date) -> dict: + """一轮完整任务:先预警(阶段 A)再清零(阶段 B)。返回合并统计。""" + warn = run_warn_once(db, notifier, reset_days=reset_days, warn_stages=warn_stages, today=today) + reset = run_reset_once(db, reset_days=reset_days, today=today) + return {**warn, **reset} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py -k "run_warn or run_once" -q` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add app/repositories/inactivity.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 不活跃预警(分档+streak去重)+ run_once 组合" +``` + +--- + +## Task 9: 进程内每日 worker + lifespan 接线 + +**Files:** +- Create: `app/core/inactivity_reset_worker.py` +- Modify: `app/main.py` +- Test: `tests/test_inactivity_reset.py` + +- [ ] **Step 1: Write the failing test** + +```python +def test_worker_disabled_returns_none(monkeypatch) -> None: + 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 + + +def test_worker_run_once_entry_executes(monkeypatch) -> None: + """_run_once_entry 用真实 SessionLocal 跑一轮,总闸开时能清掉一个不活跃用户。""" + from app.core import inactivity_reset_worker as w + from app.core.config import settings + from app.models.wallet import CoinAccount + + monkeypatch.setattr(settings, "INACTIVITY_RESET_ENABLED", True) + 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() + + stats = w._run_once_entry() + assert stats["cleared"] >= 1 + + db = SessionLocal() + try: + assert db.get(CoinAccount, uid).coin_balance == 0 + finally: + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py -k "worker" -q` +Expected: FAIL — `ModuleNotFoundError: No module named 'app.core.inactivity_reset_worker'` + +- [ ] **Step 3: Create the worker** + +Create `app/core/inactivity_reset_worker.py`(结构、文件锁完全仿 `app/core/daily_exchange_worker.py`): + +```python +"""15 天不活跃清零的进程内每日任务。 + +仿 daily_exchange_worker:App 启动自带,每 INACTIVITY_RESET_CHECK_INTERVAL_SEC 醒一次, +跨进北京新的一天且到达 INACTIVITY_RESET_RUN_HOUR 后跑一轮(预警 + 清零)。 +- 逐用户幂等:清完余额=0 次日不再匹配;预警 streak 去重。启动补跑 / 多次唤醒 / 重启都安全。 +- 同机多进程互斥:文件锁。 +- 开关:INACTIVITY_RESET_ENABLED=false 时不启动。 +""" +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]: + _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", interval) + 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 +``` + +- [ ] **Step 4: Wire into lifespan** + +在 `app/main.py`: + +(a) 导入(紧接 `from app.core.daily_exchange_worker import (...)` 之后,约 line 47): + +```python +from app.core.inactivity_reset_worker import ( + start_inactivity_reset_worker, + stop_inactivity_reset_worker, +) +``` + +(b) 启动(在 `daily_exchange_task = start_daily_exchange_worker()` 之后,约 line 82): + +```python + inactivity_task = start_inactivity_reset_worker() +``` + +(c) 停止(在 `await stop_daily_exchange_worker(daily_exchange_task)` 之后,约 line 88): + +```python + await stop_inactivity_reset_worker(inactivity_task) +``` + +- [ ] **Step 5: Run test to verify it passes** + +Run: `pytest tests/test_inactivity_reset.py -k "worker" -q` +Expected: PASS + +- [ ] **Step 6: Commit** + +```bash +git add app/core/inactivity_reset_worker.py app/main.py tests/test_inactivity_reset.py +git commit -m "feat(welfare): 不活跃清零每日 worker + lifespan 接线(总闸默认关)" +``` + +--- + +## Task 10: 重构 admin 改用共享口径(R4:单一真源) + +**Files:** +- Modify: `app/admin/repositories/stats.py` +- Modify: `app/admin/repositories/queries.py` +- Test: 现有 `tests/test_admin_read.py` 回归 + 追加一条 + +**背景:** 现 admin 口径散落且**含 last_login_at**。改造后 = 复用 `activity.py`(含 `home_view`、不含 last_login_at、以 created_at 为基线)。这会改变 admin「最近活跃/DAU」口径(登录不再计活跃、纳入 home_view),属预期变化。 + +- [ ] **Step 1: Write the failing test** + +追加到 `tests/test_inactivity_reset.py`(验证共享口径 = admin 会用到的排序列口径): + +```python +def test_admin_last_active_uses_shared_expr_without_login() -> None: + """admin 用户列表的 last_active 计算与共享口径一致:登录不推进、home_view 推进。""" + from sqlalchemy import select + from app.admin.repositories.queries import _last_active_parts # 改造后仍在,内部委托 activity + db = SessionLocal() + try: + # 仅有旧 created_at + 很新的 last_login_at,无任何活跃事件 → last_active 应=created_at(不看登录) + uid = _new_user(db, created_at=datetime(2026, 1, 1, tzinfo=timezone.utc)) + u = db.get(User, uid) + u.last_login_at = datetime(2026, 6, 1, tzinfo=timezone.utc) # 登录很新 + db.commit() + ev_sub, eng_sub = _last_active_parts() + dialect = db.get_bind().dialect.name + expr = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect) + stmt = (select(expr).select_from(User) + .outerjoin(ev_sub, ev_sub.c.user_id == User.id) + .outerjoin(eng_sub, eng_sub.c.user_id == User.id) + .where(User.id == uid)) + assert activity.norm_utc(db.execute(stmt).scalar_one()) == datetime(2026, 1, 1, tzinfo=timezone.utc) + finally: + db.rollback() + db.close() +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `pytest tests/test_inactivity_reset.py::test_admin_last_active_uses_shared_expr_without_login -q` +Expected: FAIL — 现 `_last_active_parts()` 无参、且 `_last_active_expr`/`list_users` 仍以 `last_login_at` 为基线,断言不成立(得到 2026-06-01)。 + +- [ ] **Step 3: Refactor stats.py** + +在 `app/admin/repositories/stats.py`,把本地事件常量改为复用共享(约 line 51-52): + +```python +from app.repositories.activity import COMPARE_START_EVENT, COUPON_START_EVENT # noqa: F401 +``` + +删除原来的 `COMPARE_START_EVENT = "real_compare_start"` / `COUPON_START_EVENT = "real_coupon_start"` 两行赋值(其它文件从 stats 导入它们的地方不受影响,因为已 re-export)。`_period_active_user_ids`(约 line 130)保持不变——它是"区间活跃"(含 last_login_at 落在区间),与"最近活跃排序列"是两套用途,本次不动其登录口径;仅统一事件名常量来源。 + +- [ ] **Step 4: Refactor queries.py** + +在 `app/admin/repositories/queries.py`: + +(a) 顶部改为从 activity 导入常量/助手,并让 `_last_active_parts` / `_norm_utc` / `_as_utc` 委托共享实现: + +```python +from app.repositories.activity import ( + ACTIVE_EVENTS as _ACTIVE_EVENTS, # noqa: F401 (= home_view + 比价 + 领券) + as_utc as _as_utc, + last_active_expr as _shared_last_active_expr, + last_active_subqueries, + norm_utc as _norm_utc, +) +``` + +删除本地重复定义:`_ACTIVE_EVENTS = (...)`(约 line 38)、`_norm_utc`(约 line 127-131)、`_as_utc`(约 line 625-634)。若 `_as_utc_naive`(约 line 1012)仍被引用,保留它并让其调用导入来的 `_as_utc`。 + +(b) `_last_active_parts()` 改为直接委托: + +```python +def _last_active_parts(): + """(保留函数名以兼容调用点)最近活跃两个聚合子查询,委托共享口径。""" + return last_active_subqueries(_session_of_current_call) # 见下:改签名 +``` + +> 说明:共享 `last_active_subqueries(db)` 需要 `db`。现 `_last_active_parts()` 无参(用全局 `select(...)` 构造,不需 session)。共享实现同样只用 `select(...).subquery()`、不真执行,可把 `db` 参数忽略/设可选。**因此把共享函数签名放宽**:在 `app/repositories/activity.py` 把 + +```python +def last_active_subqueries(db: Session): +``` + +改为 + +```python +def last_active_subqueries(db: Session | None = None): +``` + +(函数体不使用 db,只构造子查询;`db` 仅为语义占位)。然后 queries.py: + +```python +def _last_active_parts(): + return last_active_subqueries() +``` + +(c) `list_users` 里构造 `last_active`(约 line 199-204)改用共享 expr、基线换 `User.created_at`: + +```python + ev_agg, eng_agg = _last_active_parts() + last_active = _shared_last_active_expr( + User.created_at, ev_agg, eng_agg, db.get_bind().dialect.name + ) +``` + +删除原先 `greatest = func.greatest if ... else func.max` 与手写 `greatest(User.last_login_at, coalesce(..., User.last_login_at), ...)` 那几行。 + +(d) `_attach_last_active`(约 line 134-168)把基线由 `last_login_at` 换 `created_at`,事件集用共享常量: + +```python + for u in users: + candidates = [ + _norm_utc(u.created_at), + _norm_utc(ev_map.get(u.id)), + _norm_utc(eng_map.get(u.id)), + ] + u.last_active_at = max((c for c in candidates if c is not None), default=None) +``` + +并把该函数内两处 `AnalyticsEvent.event.in_(_ACTIVE_EVENTS)` 保持(现 `_ACTIVE_EVENTS` 已是导入来的三事件集,含 home_view)。 + +- [ ] **Step 5: Run new test + full admin regression** + +Run: `pytest tests/test_inactivity_reset.py::test_admin_last_active_uses_shared_expr_without_login tests/test_admin_read.py -q` +Expected: 新用例 PASS。若 `test_admin_read.py` 有断言依赖旧口径(把 last_login_at 当活跃、或未含 home_view),按新口径**更新其预期值**(这是设计明确的口径变化,见 spec §12);非活跃口径断言应保持通过。 + +- [ ] **Step 6: Run the whole suite** + +Run: `pytest -q` +Expected: 全绿。重点看 `test_admin_*`、`test_analytics_*`。任何红都要判断是"口径预期变化需更新断言"还是"真回归需修实现"。 + +- [ ] **Step 7: Commit** + +```bash +git add app/admin/repositories/queries.py app/admin/repositories/stats.py app/repositories/activity.py tests/ +git commit -m "refactor(admin): 最近活跃口径改用共享 activity 模块(移除 last_login_at,纳入 home_view)" +``` + +--- + +## Task 11: 表字典文档(轻量) + +**Files:** +- Create: `docs/database/inactivity_reset_log.md` +- Create: `docs/database/inactivity_notification_log.md` + +- [ ] **Step 1: Write the docs** + +仿 `docs/database/` 现有条目风格(字段表 + 用途一句话)。`inactivity_reset_log`:记录每次清零的清零前三桶余额 + 原因 + 判定时活跃时间/不活跃天数,供纠纷排查;与钱包流水 `biz_type=inactivity_reset`(ref_id=本表 id)交叉对账。`inactivity_notification_log`:记录每次预警的余额快照 + 档位 + 通道 + 状态,兼预警去重依据与占位 outbox。 + +- [ ] **Step 2: Commit** + +```bash +git add docs/database/inactivity_reset_log.md docs/database/inactivity_notification_log.md +git commit -m "docs(database): 新增 inactivity 两表字典条目" +``` + +--- + +## Self-Review + +**1. Spec coverage:** +- R1 清零(15天/北京0点边界) → Task 3(reset_cutoff)+ Task 7(run_reset)+ Task 9(worker)✓ +- R2 审计(原因+清零前余额)→ Task 1(inactivity_reset_log)+ Task 7(写审计+流水 ref 交叉)✓ +- R3 预警(xx金币xx现金)→ Task 1(notification_log)+ Task 6(notifier)+ Task 8(run_warn,快照余额)✓ +- R4 活跃口径与用户管理一致 → Task 3/4(activity 共享)+ Task 10(admin 重构)✓ +- R5 完全可配置 → Task 5(天数/次数/执行点/通道)✓ +- 活跃口径 = created_at 基线 + home_view/比价/领券,不含 last_login_at → Task 3/4/10 ✓ +- 幂等/重新活跃/streak 去重 → Task 7(清完=0)/Task 8(created_at>last_active 去重)✓ +- 触发=进程内每日 worker + 总闸默认关 → Task 9 ✓ + +**2. Placeholder scan:** 无 TBD/TODO 泛化步骤;`home_view` 事件名以 `HOME_VIEW_EVENT` 常量显式占位(前端明天定名后仅改该常量,已在 Task 3 注明)。migration 的 revision/down_revision 由 autogenerate 生成(非占位)。 + +**3. Type consistency:** 跨任务一致 —— `activity.reset_cutoff/as_utc/norm_utc/last_active_subqueries/last_active_expr/ACTIVE_EVENTS/ACTIVE_ENGAGE_TYPE/HOME_VIEW_EVENT`;`inactivity.run_once/run_reset_once/run_warn_once/clear_user/select_inactive_users/select_warn_candidates/RESET_BIZ_TYPE`;`notifier.get_notifier/InactivityNotifier.warn(...)->str/channel`;模型 `InactivityResetLog/InactivityNotificationLog` 字段名与 Task 1 定义一致;`wallet.grant_coins/grant_cash/grant_invite_cash(db, uid, amount, *, biz_type, ref_id, remark)` 与仓库实际签名一致;`get_or_create_account(db, uid, commit=False, lock=True)` 一致。 + +**已知跨仓依赖:** `home_view` 埋点由 Android 端明天新增(spec §11);后端先以常量占位、总闸默认关,`home_view` 覆盖稳定后再开总闸(spec §13)。