shaguabijia-app-server/scripts/daily_auto_exchange.py

"""0 点自动兑金币(每日定时:把每个用户金币「到分全额」兑成现金)。

背景:客户端已删手动「兑金币」入口、改文案「0 点自动兑现金（可能存在延迟）」,故服务端需在
每天 0 点把金币按汇率(100 金币 = 1 分)兑成现金。复用 exchange_in 流水类型(手动兑入口已下线,
不会与之混淆),幂等键 = 用户当天是否已有 exchange_in 流水。核心逻辑见
app/repositories/wallet.py::daily_auto_exchange。

用法:
    # 单轮(给 systemd timer / cron 用,线上每天 0 点一次)
    python -m scripts.daily_auto_exchange --once

    # 本地循环(调试用,默认每小时一轮——真到分兑换幂等,空跑无害)
    python -m scripts.daily_auto_exchange --loop --interval 3600

开关:.env 的 AUTO_EXCHANGE_ENABLED=false 时脚本直接 no-op 退出(免动 systemd timer)。
带文件锁,防上一轮没跑完下一轮又起。
"""
from __future__ import annotations

import argparse
import os
import sys
import tempfile
import time
from datetime import datetime

# Windows 控制台按 UTF-8 输出中文/¥
try:
    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
except Exception:  # noqa: BLE001
    pass

from app.core.config import settings
from app.db.session import SessionLocal
from app.repositories import wallet as wallet_repo

# 锁放系统临时目录(任何账号可写),不依赖代码目录 data/ 的写权限。需要指定位置时用环境变量覆盖。
LOCK_FILE = os.environ.get("DAILY_EXCHANGE_LOCK") or os.path.join(
    tempfile.gettempdir(), "daily_auto_exchange.lock"
)
LOCK_STALE_SEC = 30 * 60   # 锁超过 30min 视为陈旧(进程异常退出残留),自动接管


def _acquire_lock() -> bool:
    os.makedirs(os.path.dirname(LOCK_FILE) or ".", exist_ok=True)
    try:
        fd = os.open(LOCK_FILE, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        os.write(fd, f"{os.getpid()} {time.time()}".encode())
        os.close(fd)
        return True
    except FileExistsError:
        try:
            with open(LOCK_FILE) as f:
                parts = f.read().split()
            ts = float(parts[1]) if len(parts) > 1 else 0.0
            if time.time() - ts > LOCK_STALE_SEC:
                os.remove(LOCK_FILE)
                return _acquire_lock()
        except OSError:
            pass
        return False


def _release_lock() -> None:
    try:
        os.remove(LOCK_FILE)
    except OSError:
        pass


def run_once() -> None:
    if not settings.AUTO_EXCHANGE_ENABLED:
        print(f"[{datetime.now():%H:%M:%S}] AUTO_EXCHANGE_ENABLED=false,跳过(no-op)")
        return
    if not _acquire_lock():
        print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
        return
    t0 = time.time()
    try:
        with SessionLocal() as db:
            stats = wallet_repo.daily_auto_exchange(db)
        dt = time.time() - t0
        print(
            f"[{datetime.now():%H:%M:%S}] 自动兑完成 用时{dt:.1f}s "
            f"扫描{stats['scanned']} 兑换{stats['converted']} "
            f"已兑过跳过{stats['skipped_done']} 零头跳过{stats['skipped_dust']} "
            f"失败{stats['failed']} 累计兑入{stats['total_cents']}分"
        )
    finally:
        _release_lock()


def main() -> None:
    ap = argparse.ArgumentParser(description="0 点自动兑金币")
    ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
    ap.add_argument("--loop", action="store_true", help="循环跑(本地调试)")
    ap.add_argument("--interval", type=int, default=3600, help="循环间隔秒(默认 3600=1h)")
    args = ap.parse_args()

    if args.loop:
        print(f"循环模式,每 {args.interval}s 一轮(Ctrl+C 退出)")
        try:
            while True:
                run_once()
                time.sleep(args.interval)
        except KeyboardInterrupt:
            print("已退出")
    else:
        run_once()


if __name__ == "__main__":
    main()