From 40c99afd9281abfce406eda3e71a4cf484509d5a Mon Sep 17 00:00:00 2001 From: guke Date: Sat, 18 Jul 2026 17:32:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(welfare):=20=E5=AE=8C=E6=88=90=E4=B8=8D?= =?UTF-8?q?=E6=B4=BB=E8=B7=83=E6=B8=85=E9=9B=B6-=E9=82=80=E8=AF=B7?= =?UTF-8?q?=E9=87=91=E4=B8=8D=E6=B8=85=20+=20=E9=A2=84=E8=AD=A6=E7=BC=96?= =?UTF-8?q?=E6=8E=92=20+=20worker=20+=20admin=E5=8F=A3=E5=BE=84=E7=BB=9F?= =?UTF-8?q?=E4=B8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 清零改两桶(金币 + 折算现金);邀请奖励金物理隔离、不清(产品红线),仅快照入审计 - 预警编排 select_warn_candidates / run_warn_once / run_once(分档 + streak 去重); notifier.warn 去掉 invite 参数(邀请金不清、不预警) - 预警逐用户 try/except 隔离 + run_once 保证预警故障绝不阻塞清零(接真推送前必备) - 每日 worker inactivity_reset_worker(文件锁 + 北京日守卫 + RUN_HOUR 门槛 + 总闸默认关) + main.py lifespan 接线 - admin 最近活跃口径改用共享 activity(移除 last_login_at、纳入 home_view 占位) Co-Authored-By: Claude Opus 4.8 (1M context) --- app/admin/repositories/queries.py | 73 ++-------- app/core/inactivity_reset_worker.py | 143 +++++++++++++++++++ app/integrations/notifier.py | 12 +- app/main.py | 6 + app/models/inactivity.py | 6 +- app/repositories/inactivity.py | 92 +++++++++++-- tests/test_inactivity_reset.py | 205 +++++++++++++++++++++++++++- 7 files changed, 452 insertions(+), 85 deletions(-) create mode 100644 app/core/inactivity_reset_worker.py diff --git a/app/admin/repositories/queries.py b/app/admin/repositories/queries.py index 501dba9..0426b3e 100644 --- a/app/admin/repositories/queries.py +++ b/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), + AnalyticsEvent.event.in_(activity.ACTIVE_EVENTS), ) .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) diff --git a/app/core/inactivity_reset_worker.py b/app/core/inactivity_reset_worker.py new file mode 100644 index 0000000..7296d59 --- /dev/null +++ b/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 diff --git a/app/integrations/notifier.py b/app/integrations/notifier.py index e1b10fe..5731d12 100644 --- a/app/integrations/notifier.py +++ b/app/integrations/notifier.py @@ -15,9 +15,10 @@ 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, + def warn(self, *, user_id: int, coin: int, cash_cents: int, stage: int, days_until_reset: int) -> str: - """发预警,返回状态:'sent' / 'failed' / 'placeholder'。""" + """发预警(只涉及会被清的金币 + 折算现金;邀请现金不清、不预警)。 + 返回状态:'sent' / 'failed' / 'placeholder'。""" ... @@ -26,12 +27,11 @@ class LogNotifier: channel = "log" - def warn(self, *, user_id: int, coin: int, cash_cents: int, invite_cash_cents: int, + 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 invite_cash_cents=%s " - "stage=T-%s days_until_reset=%s", - user_id, coin, cash_cents, invite_cash_cents, stage, days_until_reset, + "[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" diff --git a/app/main.py b/app/main.py index 3b3b6ff..f99611e 100644 --- a/app/main.py +++ b/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") diff --git a/app/models/inactivity.py b/app/models/inactivity.py index 343a785..e67fa27 100644 --- a/app/models/inactivity.py +++ b/app/models/inactivity.py @@ -1,7 +1,9 @@ """15 天不活跃清零相关表。 -- inactivity_reset_log:每次清零一行,记清零前三桶余额 + 原因 + 判定时活跃时间/不活跃天数, - 供纠纷排查(需求①)。清零同时另写 3 条钱包流水(biz_type=inactivity_reset),资金流可逐笔回溯。 +- 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)。 diff --git a/app/repositories/inactivity.py b/app/repositories/inactivity.py index 928fe84..71970cc 100644 --- a/app/repositories/inactivity.py +++ b/app/repositories/inactivity.py @@ -5,26 +5,31 @@ """ from __future__ import annotations -from datetime import date, datetime, timezone +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.models.inactivity import InactivityResetLog +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, - CoinAccount.invite_cash_balance_cents > 0, ) @@ -58,17 +63,19 @@ def _inactive_days(last_active: datetime, today: date) -> int: def select_inactive_users(db: Session, *, cutoff: datetime): - """应清零用户:last_active < cutoff 且三桶有余额。返回 Row 列表(值已快照,可跨 commit)。""" + """应清零用户: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 条流水。返回是否真清了(有余额)。""" + """单用户清零(独立事务、行锁)。金币 + 折算现金归零 + 写审计 + 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 and invite == 0: + if coin == 0 and cash == 0: # 邀请现金不清,故不算"有可清余额" return False log = InactivityResetLog( user_id=user_id, coin_balance_before=coin, cash_balance_cents_before=cash, @@ -82,8 +89,7 @@ def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_day 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) + # 邀请现金(invite_cash_balance_cents)刻意不动:两本账物理隔离、邀请金是产品红线。 db.commit() return True @@ -105,3 +111,73 @@ def run_reset_once(db: Session, *, reset_days: int, today: date) -> dict: 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} diff --git a/tests/test_inactivity_reset.py b/tests/test_inactivity_reset.py index 5499ae1..d25a4b2 100644 --- a/tests/test_inactivity_reset.py +++ b/tests/test_inactivity_reset.py @@ -85,9 +85,11 @@ def _isolate_inactivity_state(): 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, + # 199 前缀 + 递增序号:共享测试库跨文件累积用户,别的文件用固定手机号(如 test_admin_write + # 的 13900000001..),这里用没人用的 199 段避免撞 user.phone / username 的 UNIQUE。 + u = User(phone=f"199{_PHONE_SEQ[0]:08d}", created_at=created_at, last_login_at=created_at, status="active", - username=f"2{_PHONE_SEQ[0]:010d}") + username=f"inact{_PHONE_SEQ[0]}") db.add(u) db.flush() acc = wallet_repo.get_or_create_account(db, u.id, commit=False) @@ -142,13 +144,13 @@ 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) + status = n.warn(user_id=1, coin=10, cash_cents=20, stage=7, days_until_reset=8) assert status == "placeholder" # 未实现通道回退 LogNotifier(占位) assert get_notifier("jpush").channel == "log" -def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None: +def test_run_reset_clears_coin_and_cash_but_preserves_invite_cash() -> None: from sqlalchemy import select from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction from app.repositories import inactivity @@ -168,10 +170,13 @@ def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None: 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) + # 金币 + 折算现金清零;邀请现金是产品红线,原封不动(见 wallet.CoinAccount 注释) + assert (acc.coin_balance, acc.cash_balance_cents) == (0, 0) + assert acc.invite_cash_balance_cents == 300 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" @@ -181,12 +186,198 @@ def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None: 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 + InviteCashTransaction.user_id == old, + InviteCashTransaction.biz_type == "inactivity_reset")).first() is None - # 活跃用户不动;再跑一次幂等(已清零 → 不再匹配) + # 活跃用户不动;再跑一次幂等(coin+cash 已 0、邀请现金不算候选 → 不再匹配) 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() + + +def test_user_with_only_invite_cash_is_not_cleared() -> None: + """只有邀请现金余额的久不活跃用户:邀请现金是产品红线,不清 → 根本不该被选中。""" + from app.models.wallet import CoinAccount + from app.repositories import inactivity + + db = SessionLocal() + try: + today = date(2026, 2, 1) + uid = _new_user(db, created_at=datetime(2026, 1, 10, tzinfo=timezone.utc), + coin=0, cash=0, invite=500) + db.commit() + stats = inactivity.run_reset_once(db, reset_days=15, today=today) + assert stats["cleared"] == 0 + assert db.get(CoinAccount, uid).invite_cash_balance_cents == 500 # 原封不动 + finally: + db.rollback() + db.close() + + +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.models.wallet import CoinAccount + 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 + assert db.get(CoinAccount, clear_uid).coin_balance == 0 + assert db.get(CoinAccount, warn_uid).coin_balance == 10 # 预警不动钱 + finally: + db.rollback() + db.close() + + +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() + + +def test_admin_list_users_last_active_ignores_login() -> None: + """admin 用户列表 last_active_at 改用共享口径:登录不算活跃(baseline=created_at)、只认活跃事件。""" + from app.admin.repositories import queries + + db = SessionLocal() + try: + created = datetime(2026, 1, 1, tzinfo=timezone.utc) + uid = _new_user(db, created_at=created) + u = db.get(User, uid) + u.last_login_at = datetime(2026, 6, 1, tzinfo=timezone.utc) # 登录很新、但无任何活跃事件 + db.commit() + phone = db.get(User, uid).phone + users, _cursor, _total = queries.list_users(db, phone=phone) + item = next(x for x in users if x.id == uid) + assert activity.norm_utc(item.last_active_at) == created # 登录不算 → last_active=created_at + finally: + db.rollback() + db.close() + + +def test_run_warn_isolates_notifier_failure_and_does_not_block_reset() -> None: + """单用户通知器抛错:预警计 warn_failed、不外抛,且清零(reset)照常执行。""" + from app.models.wallet import CoinAccount + from app.repositories import inactivity + + class BoomNotifier: + channel = "log" + + def warn(self, *, user_id, coin, cash_cents, stage, days_until_reset) -> str: + raise RuntimeError("push service down") + + 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=BoomNotifier(), reset_days=15, + warn_stages=[7, 2], today=today) + assert stats["warned"] == 0 and stats["warn_failed"] >= 1 # 预警失败被隔离 + assert stats["cleared"] == 1 # 关键:清零没被阻塞 + assert db.get(CoinAccount, clear_uid).coin_balance == 0 + assert db.get(CoinAccount, warn_uid).coin_balance == 10 # 预警用户不动钱 + # 预警失败已回滚,不留半条 notification_log + from sqlalchemy import select + assert db.execute(select(InactivityNotificationLog).where( + InactivityNotificationLog.user_id == warn_uid)).first() is None + finally: + db.rollback() + db.close() + + +def test_run_once_reset_runs_even_if_warn_phase_throws(monkeypatch) -> None: + """预警整段异常(如候选查询失败)也绝不阻塞清零。""" + from app.integrations.notifier import LogNotifier + from app.models.wallet import CoinAccount + from app.repositories import inactivity + + def boom(*a, **k): + raise RuntimeError("warn phase blew up") + + monkeypatch.setattr(inactivity, "run_warn_once", boom) + + db = SessionLocal() + try: + today = date(2026, 2, 1) + clear_uid = _new_user(db, created_at=datetime(2026, 1, 5, tzinfo=timezone.utc), coin=10) + db.commit() + stats = inactivity.run_once(db, notifier=LogNotifier(), reset_days=15, + warn_stages=[7, 2], today=today) + assert stats["cleared"] == 1 + assert db.get(CoinAccount, clear_uid).coin_balance == 0 + finally: + db.rollback() + db.close()