shaguabijia-app-server/scripts/reset_welfare.py

"""一键回滚某个用户的福利数据(金币/现金/签到/一次性任务/看广告发奖),方便反复测试。

清掉:coin_transaction / cash_transaction / signin_record / user_task / ad_reward_record,
并把 coin_account 的金币余额、累计收益、现金余额全部归零。
**不动** savings_record(那是 profile 省钱卡的 demo 演示数据,与领奖测试无关)。

用法(在 shaguabijia-app-server 目录下,用本项目的 python 跑):
    python scripts/reset_welfare.py            # 默认重置 user_id=2(手机登录用户)
    python scripts/reset_welfare.py 5           # 重置 user_id=5
    python scripts/reset_welfare.py --all       # 重置所有用户
    python scripts/reset_welfare.py --dry-run    # 只看现状,不改

DB 默认取 <项目根>/data/app.db,可用 --db 指定。
"""
from __future__ import annotations

import argparse
import sqlite3
import sys
from pathlib import Path

# Windows 控制台默认 GBK,强制 UTF-8 否则中文输出乱码
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

# 本脚本在 scripts/ 下,项目根是上一级
DEFAULT_DB = Path(__file__).resolve().parent.parent / "data" / "app.db"

# 受影响的表:每个用户的钱包/行为数据
WALLET_TABLES = ("coin_transaction", "cash_transaction", "signin_record", "user_task", "ad_reward_record")


def snapshot(cur: sqlite3.Cursor, where: str, params: tuple) -> dict:
    """返回当前各表行数 + 账户余额,用于打印 before/after。"""
    counts = {
        t: cur.execute(f"select count(*) from {t} where {where}", params).fetchone()[0]
        for t in WALLET_TABLES
    }
    accounts = cur.execute(
        f"select user_id, coin_balance, cash_balance_cents, total_coin_earned "
        f"from coin_account where {where}",
        params,
    ).fetchall()
    return {"counts": counts, "accounts": accounts}


def print_snapshot(label: str, snap: dict) -> None:
    print(f"--- {label} ---")
    for t, n in snap["counts"].items():
        print(f"  {t}: {n}")
    if snap["accounts"]:
        for uid, coin, cash, earned in snap["accounts"]:
            print(f"  coin_account user={uid}: coin={coin} cash_cents={cash} earned={earned}")
    else:
        print("  coin_account: (无)")


def main() -> None:
    parser = argparse.ArgumentParser(description="回滚用户福利数据")
    parser.add_argument("user_id", nargs="?", type=int, default=2,
                        help="要重置的 user_id(默认 2)")
    parser.add_argument("--all", action="store_true", help="重置所有用户")
    parser.add_argument("--dry-run", action="store_true", help="只看现状,不修改")
    parser.add_argument("--db", default=str(DEFAULT_DB), help="SQLite 路径")
    args = parser.parse_args()

    if args.all:
        where, params, scope = "1=1", (), "ALL users"
    else:
        where, params, scope = "user_id=?", (args.user_id,), f"user_id={args.user_id}"

    db_path = Path(args.db)
    if not db_path.exists():
        raise SystemExit(f"DB 不存在: {db_path}")

    conn = sqlite3.connect(str(db_path))
    cur = conn.cursor()
    print(f"DB: {db_path}  scope: {scope}")
    print_snapshot("before", snapshot(cur, where, params))

    if args.dry_run:
        print("(dry-run,未修改)")
        conn.close()
        return

    for t in WALLET_TABLES:
        cur.execute(f"delete from {t} where {where}", params)
    cur.execute(
        f"update coin_account set coin_balance=0, total_coin_earned=0, "
        f"cash_balance_cents=0 where {where}",
        params,
    )
    conn.commit()

    print_snapshot("after", snapshot(cur, where, params))
    conn.close()
    print("回滚完成。提醒:App 内存里的共享余额态不会自动同步,重启 App 或进收益明细页刷新。")


if __name__ == "__main__":
    main()