diff --git a/app/repositories/inactivity.py b/app/repositories/inactivity.py new file mode 100644 index 0000000..928fe84 --- /dev/null +++ b/app/repositories/inactivity.py @@ -0,0 +1,107 @@ +"""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.models.inactivity import 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 diff --git a/tests/test_inactivity_reset.py b/tests/test_inactivity_reset.py index b3f328a..053ef96 100644 --- a/tests/test_inactivity_reset.py +++ b/tests/test_inactivity_reset.py @@ -127,3 +127,52 @@ def test_log_notifier_returns_placeholder(caplog) -> None: assert status == "placeholder" # 未实现通道回退 LogNotifier(占位) assert get_notifier("jpush").channel == "log" + + +def test_run_reset_clears_three_buckets_and_writes_audit_and_flows() -> None: + from sqlalchemy import select, update + from app.models.wallet import CoinAccount, CoinTransaction, CashTransaction, InviteCashTransaction + from app.repositories import inactivity + + db = SessionLocal() + try: + # 清掉前序测试遗留的余额(session-scoped DB,前测 commit 后 rollback 是 no-op) + db.execute(update(CoinAccount).values(coin_balance=0, cash_balance_cents=0, + invite_cash_balance_cents=0)) + db.commit() + + 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()