shaguabijia-app-server/scripts/reset_signin.py

"""清空指定用户的签到记录,方便反复测试"未签到→自动弹窗→签到领金币"。

默认**只删 signin_record**(签到历史),不动金币/现金——这样能保留你已注入/兑换的余额,
只把"今天是否已签到 / 连续天数"重置掉。删后当天即变成"未签到",再进福利页会重新自动弹签到弹窗。

⚠️ 注意:只删签到记录的话,之前签到已发的金币仍留在余额和 coin_transaction 里(biz_type='signin')。
如果想把签到发的金币也一并退掉(余额减回、删 signin 流水),加 --with-coins。

用法(在 shaguabijia-app-server 目录下,用本项目的 python 跑):
    python scripts/reset_signin.py              # 只清 user_id=2 的签到记录
    python scripts/reset_signin.py 5            # 清 user_id=5
    python scripts/reset_signin.py --all        # 清所有用户的签到记录
    python scripts/reset_signin.py --with-coins # 连签到发的金币一起退回(余额减、删 signin 流水)
    python scripts/reset_signin.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"


def snapshot(cur: sqlite3.Cursor, where: str, params: tuple) -> dict:
    """返回签到记录数 + 签到金币流水数/总额 + 账户金币余额,用于打印 before/after。"""
    signin_rows = cur.execute(
        f"select count(*) from signin_record where {where}", params
    ).fetchone()[0]
    signin_coin_rows, signin_coin_sum = cur.execute(
        f"select count(*), coalesce(sum(amount), 0) from coin_transaction "
        f"where biz_type='signin' and {where}",
        params,
    ).fetchone()
    accounts = cur.execute(
        f"select user_id, coin_balance, total_coin_earned from coin_account where {where}",
        params,
    ).fetchall()
    return {
        "signin_rows": signin_rows,
        "signin_coin_rows": signin_coin_rows,
        "signin_coin_sum": signin_coin_sum,
        "accounts": accounts,
    }


def print_snapshot(label: str, snap: dict) -> None:
    print(f"--- {label} ---")
    print(f"  signin_record: {snap['signin_rows']}")
    print(f"  coin_transaction(signin): {snap['signin_coin_rows']} 条, 共 {snap['signin_coin_sum']} 金币")
    if snap["accounts"]:
        for uid, coin, earned in snap["accounts"]:
            print(f"  coin_account user={uid}: coin={coin} 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("--with-coins", action="store_true",
                        help="同时退回签到发的金币(余额减、删 signin 流水)")
    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}  with_coins: {args.with_coins}")
    print_snapshot("before", snapshot(cur, where, params))

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

    if args.with_coins:
        # 先按用户把签到金币总额退回(余额、累计收益都减),再删 signin 流水
        rows = cur.execute(
            f"select user_id, coalesce(sum(amount), 0) from coin_transaction "
            f"where biz_type='signin' and {where} group by user_id",
            params,
        ).fetchall()
        for uid, total in rows:
            cur.execute(
                "update coin_account set coin_balance = coin_balance - ?, "
                "total_coin_earned = max(0, total_coin_earned - ?) where user_id = ?",
                (total, total, uid),
            )
        cur.execute(f"delete from coin_transaction where biz_type='signin' and {where}", params)

    cur.execute(f"delete from signin_record where {where}", params)
    conn.commit()

    print_snapshot("after", snapshot(cur, where, params))
    conn.close()
    print("签到记录已清空。提醒:App 内存里的状态不会自动同步,重启 App 或进收益明细页刷新;"
          "重进福利页(当天未签到)会重新自动弹签到弹窗。")


if __name__ == "__main__":
    main()