WonderableAI/shaguabijia-app-server
11
0
Fork 0
Code Issues Pull Requests 13 Actions Packages Projects Releases Wiki Activity

feat(welfare): 不活跃清零 worker 常驻 + 默认 dry-run(ENABLED=false 只记审计不清)

Browse Source 
改 INACTIVITY_RESET_ENABLED 语义:worker 不再"关=不启动"、改为**常驻**;ENABLED 只控是否**真清**——
false(默认)= 只写审计名单(reason=inactive_Nd_dryrun)、不动钱不写流水、不预警(dry-run 灰度看名单);
true = 真清金币+现金 + 预警。

- clear_user(dry_run):写审计跳出账,按 streak 去重(reset_at>last_active)不重复记
- run_reset_once/run_once:透传 dry_run;dry-run 抑制预警(不通知一个不会发生的清零)
- worker:start 常驻(去掉 ENABLED 启动闸);_run_once_entry dry_run=not ENABLED;启动日志标 mode
- 测试:dry-run 只记审计/不清/不预警/dedup + worker ENABLED=false→dry-run;去掉过时 disabled 用例
- config/worker 注释 + spec §6/§13/误清防护 同步新语义

测试安全:lifespan 在测试里不跑(TestClient 非 with 形式),worker 常驻不影响用例。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
guke
2026-07-18 23:39:50 +08:00
parent 5fdf2b1ad3
commit 0cc2e9dec8
5 changed files with 119 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/core/config.py
+4 -2
View File
@@ -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)
app/core/inactivity_reset_worker.py
+7 -5
View File
@@ -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")
app/repositories/inactivity.py
+36 -21
View File
@@ -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}