zhuzihao
8a2f72d366
Co-authored-by: zzhyyyyy <2685922758@qq.com> Reviewed-on: #63 Co-authored-by: zhuzihao <zhuzihao@wonderable.ai> Co-committed-by: zhuzihao <zhuzihao@wonderable.ai>
126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
"""0 点金币→现金自动兑换的进程内定时任务。
|
|
|
|
替代原 systemd timer(deploy/daily-exchange.*):App 启动即自带,每
|
|
`AUTO_EXCHANGE_CHECK_INTERVAL_SEC` 醒一次,跨进北京新的一天(0 点后)就跑一轮
|
|
`wallet.daily_auto_exchange`。
|
|
|
|
健壮性:
|
|
- **逐用户幂等**:当天已有 exchange_in 流水的用户跳过(见 wallet._has_exchange_in_on),
|
|
故启动补跑 / 多次唤醒 / 进程重启都安全,不会重复兑。
|
|
- **当天首跑即补**:进程起来时若当天还没兑过,立即兑一轮(等价原 timer 的 Persistent 补跑)。
|
|
- **同机多进程互斥**:文件锁保证多 worker 只有一个实际跑(防跨进程并发导致 TOCTOU 双兑)。
|
|
- **开关**:settings.AUTO_EXCHANGE_ENABLED=false 时不启动(与脚本/原 timer 同一开关)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections.abc import Iterator
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
|
|
from app.core import rewards
|
|
from app.core.config import settings
|
|
from app.db.session import SessionLocal
|
|
from app.repositories import wallet as wallet_repo
|
|
|
|
logger = logging.getLogger("shagua.daily_exchange")
|
|
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "daily_exchange.lock"
|
|
|
|
|
|
def _touch_lock() -> None:
|
|
with contextlib.suppress(FileNotFoundError):
|
|
os.utime(_LOCK_PATH, None)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
|
|
"""同机多进程保护:同一时间只允许一个自动兑换 worker 运行。"""
|
|
_LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
fd: int | None = None
|
|
try:
|
|
try:
|
|
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
|
except FileExistsError:
|
|
try:
|
|
age = time.time() - _LOCK_PATH.stat().st_mtime
|
|
except FileNotFoundError:
|
|
age = stale_after_sec + 1
|
|
if age > stale_after_sec:
|
|
with contextlib.suppress(FileNotFoundError):
|
|
_LOCK_PATH.unlink()
|
|
try:
|
|
fd = os.open(str(_LOCK_PATH), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
|
except FileExistsError:
|
|
fd = None
|
|
|
|
if fd is None:
|
|
yield False
|
|
return
|
|
|
|
os.write(fd, f"pid={os.getpid()} started_at={int(time.time())}\n".encode("ascii"))
|
|
yield True
|
|
finally:
|
|
if fd is not None:
|
|
os.close(fd)
|
|
with contextlib.suppress(FileNotFoundError):
|
|
_LOCK_PATH.unlink()
|
|
|
|
|
|
def _exchange_once() -> dict:
|
|
with SessionLocal() as db:
|
|
return wallet_repo.daily_auto_exchange(db)
|
|
|
|
|
|
async def _run_loop() -> None:
|
|
interval = max(60, int(settings.AUTO_EXCHANGE_CHECK_INTERVAL_SEC))
|
|
lock_stale_after = max(interval * 3, 1800)
|
|
with _single_instance_lock(lock_stale_after) as lock_acquired:
|
|
if not lock_acquired:
|
|
logger.warning("daily auto-exchange skipped: another worker owns lock")
|
|
return
|
|
await _run_locked_loop(interval)
|
|
|
|
|
|
async def _run_locked_loop(interval: int) -> None:
|
|
logger.info("daily auto-exchange worker started interval=%ss", interval)
|
|
# 本进程上次跑过的北京日;None=尚未跑过本进程(启动即补当天)。
|
|
last_run: date | None = None
|
|
try:
|
|
while True:
|
|
try:
|
|
_touch_lock()
|
|
today = rewards.cn_today()
|
|
if last_run != today:
|
|
result = await asyncio.to_thread(_exchange_once)
|
|
last_run = today
|
|
logger.info("daily auto-exchange done date=%s result=%s", today, result)
|
|
except SQLAlchemyError:
|
|
logger.exception("daily auto-exchange db error")
|
|
except Exception: # noqa: BLE001 - 后台任务不能因单次异常退出
|
|
logger.exception("daily auto-exchange unexpected error")
|
|
await asyncio.sleep(interval)
|
|
except asyncio.CancelledError:
|
|
logger.info("daily auto-exchange worker stopped")
|
|
raise
|
|
|
|
|
|
def start_daily_exchange_worker() -> asyncio.Task | None:
|
|
if not settings.AUTO_EXCHANGE_ENABLED:
|
|
logger.info("daily auto-exchange disabled (AUTO_EXCHANGE_ENABLED=false)")
|
|
return None
|
|
return asyncio.create_task(_run_loop(), name="daily-auto-exchange")
|
|
|
|
|
|
async def stop_daily_exchange_worker(task: asyncio.Task | None) -> None:
|
|
if task is None:
|
|
return
|
|
task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await task
|