shaguabijia-app-server/scripts/reset_guide_video.py

"""重置指定账号的「领券引导视频」已播次数,让领券等候浮层重新放引导视频,方便反复看这支片。

原理:浮层放不放引导视频只由两件事决定(见 app/repositories/guide_video.py `start_play`):
  1. 运营配置 app_config(key=coupon_guide_video):enabled / video_url / max_plays;
  2. 该账号 guide_video_play 的**行数**(开播即计次)—— 行数 >= max_plays 就不再下发,
     `/api/v1/guide-video/start` 返 should_play=false,客户端照旧放广告。
所以「想再看一遍」= 删掉这个账号的 guide_video_play 行。配置本脚本**只读不改**(要换片子 /
开关去运营后台「配置 - 领券引导视频」;次数 3 / 金币 120 是服务端默认值,后台不开放调整),
这样不会把别的账号的线上行为一起动了。

默认连**当时发的金币一起退**(每次 reward_coin,现配置 120):这些流水 biz_type='guide_video',
不退的话每重置一轮余额就白涨一轮,收益明细里还会堆出一串「新手引导视频奖励」。想留着用 --keep-coins。
例外:金币已被兑换成现金、余额兜不住时**整笔跳过不退** —— coin_balance 必须恒等于流水总和,
硬退会退成负数,夹到 0 又会吃掉别处赚的金币,两种做法都会让账对不上(同 reset_signin_today.py)。

用法(在项目根、已 pip install -e . 的环境里跑):
  python scripts/reset_guide_video.py                  # 默认测试号 11111111111
  python scripts/reset_guide_video.py 13800138000      # 指定手机号
  python scripts/reset_guide_video.py --user-id 5      # 直接指定 user_id
  python scripts/reset_guide_video.py --dry-run        # 预览(照常执行再回滚),不落库
  python scripts/reset_guide_video.py --keep-coins     # 只清次数,已发金币不退

走 SessionLocal 连 DATABASE_URL(SQLite / Postgres 都行),因此**只允许 APP_ENV=dev 改库**
(--dry-run 只读,任何环境都能跑)。
"""
from __future__ import annotations

import argparse
import sys
from pathlib import Path

from sqlalchemy import select

from app.core.config import settings
from app.db.session import SessionLocal, engine
from app.models.guide_video import GuideVideoPlay
from app.models.user import User
from app.models.wallet import CoinAccount, CoinTransaction
from app.repositories import guide_video as crud_guide

# Windows 控制台默认 GBK,强制 UTF-8 否则中文输出乱码。stderr 也要设:
# SystemExit(如"用户不存在")的中文提示走的是 stderr。
for _stream in (sys.stdout, sys.stderr):
    if hasattr(_stream, "reconfigure"):
        _stream.reconfigure(encoding="utf-8")

# dev 下 engine 是 echo=True(APP_DEBUG),几十行 SQL 会把前后对比刷没。echo 走 SQLAlchemy 自己的
# InstanceLogger,不吃 logging.setLevel,只能改 engine.echo。
engine.echo = False

DEFAULT_PHONE = "11111111111"


def resolve_user(db, phone: str, user_id: int | None) -> User:
    if user_id is not None:
        user = db.get(User, user_id)
        if user is None:
            raise SystemExit(f"user_id={user_id} 不存在")
        return user
    user = db.execute(select(User).where(User.phone == phone)).scalar_one_or_none()
    if user is None:
        raise SystemExit(f"手机号 {phone} 没有对应用户(注意 phone 才是登录账号,username 是展示 ID)")
    return user


def load_plays(db, user_id: int) -> list[GuideVideoPlay]:
    return list(db.execute(
        select(GuideVideoPlay)
        .where(GuideVideoPlay.user_id == user_id)
        .order_by(GuideVideoPlay.id)
    ).scalars().all())


def print_config(db) -> dict:
    """打印运营配置 + 片子是否真在盘上(配了地址但文件丢了,客户端会黑屏/加载失败)。"""
    cfg = crud_guide.get_config(db)
    video_url = (cfg.get("video_url") or "").strip()
    print("--- 运营配置(app_config: coupon_guide_video,本脚本不改) ---")
    print(f"  enabled={cfg['enabled']}  max_plays={cfg['max_plays']}  reward_coin={cfg['reward_coin']}")
    print(f"  video_url={video_url or '(未配片)'}")
    if video_url.startswith(settings.MEDIA_URL_PREFIX + "/"):
        rel = video_url[len(settings.MEDIA_URL_PREFIX) + 1:]
        f = Path(settings.MEDIA_ROOT) / rel
        if f.is_file():
            print(f"  文件: {f} ({f.stat().st_size / 1024 / 1024:.1f} MB) ✓")
        else:
            print(f"  ⚠️ 文件不存在: {f} —— 客户端会加载失败,请去运营后台重新上传")
    return cfg


def print_state(db, user: User, cfg: dict, label: str) -> None:
    plays = load_plays(db, user.id)
    max_plays = int(cfg.get("max_plays") or 0)
    used = len(plays)
    print(f"--- {label} ---")
    print(f"  guide_video_play: {used} 行(已用次数,开播即计),上限 {max_plays}")
    for p in plays:
        print(f"    #{p.id} 第{p.seq}次 {p.status} +{p.coin}金币 "
              f"completed={p.completed} 开播于 {p.started_at} token={p.play_token[:12]}…")

    # 用仓储层原样复算一遍"下次会不会放",而不是脚本里自己判规则 —— 这就是客户端会拿到的结果
    enabled = bool(cfg.get("enabled"))
    has_video = bool((cfg.get("video_url") or "").strip())
    will_play = enabled and has_video and max_plays > 0 and used < max_plays
    reason = (
        "会放引导视频" if will_play
        else "开关关着" if not enabled
        else "没配片" if not has_video
        else "max_plays=0" if max_plays <= 0
        else f"次数已用完({used}/{max_plays})"
    )
    print(f"  [下次点一键领取] should_play={will_play} —— {reason}")

    acc = db.get(CoinAccount, user.id)
    if acc is None:
        print("  coin_account: (无)")
    else:
        print(f"  coin_account: coin={acc.coin_balance} earned={acc.total_coin_earned}")


def refund_plays(db, user_id: int, tokens: list[str]) -> None:
    """退回这些播放发的金币:删流水 + 扣余额。余额兜不住的整笔跳过(见模块 docstring)。"""
    if not tokens:
        return
    rows = list(db.execute(
        select(CoinTransaction).where(
            CoinTransaction.user_id == user_id,
            CoinTransaction.biz_type == crud_guide.BIZ_TYPE,
            CoinTransaction.ref_id.in_(tokens),
        )
    ).scalars().all())
    if not rows:
        print("  没有可退的金币流水(可能是 reward_coin=0,或本来就没发过)")
        return
    acc = db.get(CoinAccount, user_id)
    if acc is None:
        return

    # 从最近一笔往回退,退到余额兜不住为止 —— 保证"最新发的那笔"一定被退掉。
    rows.sort(key=lambda r: r.id, reverse=True)
    refundable: list[CoinTransaction] = []
    total = 0
    for r in rows:
        if total + r.amount > acc.coin_balance:
            break
        refundable.append(r)
        total += r.amount

    for r in refundable:
        db.delete(r)
    if total:
        acc.coin_balance -= total
        acc.total_coin_earned = max(0, acc.total_coin_earned - total)
        print(f"  已退回 {total} 金币({len(refundable)}/{len(rows)} 笔引导视频奖励)")

    stuck = len(rows) - len(refundable)
    if stuck:
        print(f"  ⚠️ 还有 {stuck} 笔退不掉(金币已被兑换/花掉,余额 {acc.coin_balance} 兜不住),原样保留 —— "
              "硬退会让余额和流水总和对不上。收益明细里会多留几条,不影响再看视频。")


def main() -> None:
    parser = argparse.ArgumentParser(description="重置账号的领券引导视频次数,让浮层重新放引导视频")
    parser.add_argument("phone", nargs="?", default=DEFAULT_PHONE,
                        help=f"手机号(默认 {DEFAULT_PHONE})")
    parser.add_argument("--user-id", type=int, default=None, help="直接按 user_id 定位,优先于 phone")
    parser.add_argument("--keep-coins", action="store_true",
                        help="不退已发金币(余额会越测越高,收益明细堆重复流水)")
    parser.add_argument("--dry-run", action="store_true", help="预览,最后回滚不落库")
    args = parser.parse_args()

    if not args.dry_run and settings.APP_ENV != "dev":
        raise SystemExit(f"APP_ENV={settings.APP_ENV},拒绝改库(只有 dev 能改;--dry-run 可任意环境)")

    db = SessionLocal()
    try:
        user = resolve_user(db, args.phone, args.user_id)
        print(f"DB: {settings.DATABASE_URL}  APP_ENV: {settings.APP_ENV}")
        print(f"用户: id={user.id} phone={user.phone} username={user.username}  keep_coins: {args.keep_coins}")
        cfg = print_config(db)
        print_state(db, user, cfg, "before")

        plays = load_plays(db, user.id)
        if not plays:
            print("该账号没有任何播放记录,次数本来就是满的,无需重置。")
            db.rollback()
            return

        tokens = [p.play_token for p in plays]
        for p in plays:
            db.delete(p)
        db.flush()

        if args.keep_coins:
            print("(--keep-coins:保留已发金币,流水和余额不动)")
        else:
            refund_plays(db, user.id, tokens)
            # 注:更早流水的 balance_after 是当时的快照,不回改 —— 收益明细里历史行的余额列
            # 会与现余额对不上,dev 测试库无妨。

        print_state(db, user, cfg, "after")

        if args.dry_run:
            db.rollback()
            print(f"(dry-run:以上 after 为预览,已回滚,库没动;真跑会删 {len(plays)} 条播放记录)")
            return
        db.commit()
        print(f"完成:删掉 {len(plays)} 条播放记录,{user.phone} 下次点「一键自动领取」浮层会重新放引导视频。")
        print("提醒:客户端是在浮层建窗时调 /guide-video/start 的,不用重装、不用重登,直接再点一次一键领取即可;"
              "但「广告显示」开关关掉时整块浮层不建窗(连引导视频也不放),测之前先确认它是开的。")
    finally:
        db.close()


if __name__ == "__main__":
    main()