shaguabijia-app-server/app/core/withdraw_reconcile_worker.py

"""提现 pending 单自动对账后台任务。"""
from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from pathlib import Path

from sqlalchemy.exc import SQLAlchemyError

from app.core.config import settings
from app.db.session import SessionLocal
from app.integrations.wxpay import WxPayNotConfiguredError
from app.repositories import app_config
from app.repositories import wallet as wallet_repo

_AUTO_RECONCILE_KEY = "withdraw_auto_reconcile_enabled"

logger = logging.getLogger("shagua.withdraw_reconcile")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.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 _reconcile_once(older_than_minutes: int) -> dict | None:
    """读运营开关(app_config):关着返回 None(本轮跳过),开着才真扫单。

    env WITHDRAW_AUTO_RECONCILE_ENABLED 是部署级总闸(决定 worker 起不起);
    这里的 DB 开关是运营级日常开关,后台一改下一轮即生效、跨进程一致、无需重启。
    """
    with SessionLocal() as db:
        if not app_config.get_value(db, _AUTO_RECONCILE_KEY):
            return None
        return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)


async def _run_loop() -> None:
    interval = max(30, int(settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC))
    older_than = max(1, int(settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES))
    lock_stale_after = max(interval * 3, 600)
    with _single_instance_lock(lock_stale_after) as lock_acquired:
        if not lock_acquired:
            logger.warning("withdraw auto reconcile skipped: another worker owns lock")
            return

        await _run_locked_loop(interval, older_than)


async def _run_locked_loop(interval: int, older_than: int) -> None:
    logger.info(
        "withdraw auto reconcile worker started interval=%ss older_than=%sm "
        "(runtime on/off via app_config '%s')",
        interval,
        older_than,
        _AUTO_RECONCILE_KEY,
    )
    try:
        while True:
            try:
                _touch_lock()
                result = await asyncio.to_thread(_reconcile_once, older_than)
                if result is not None and (result["checked"] or result["resolved"]):
                    logger.info("withdraw auto reconcile result=%s", result)
            except WxPayNotConfiguredError:
                logger.warning("withdraw auto reconcile skipped: wxpay not configured")
            except SQLAlchemyError:
                logger.exception("withdraw auto reconcile db error")
            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
                logger.exception("withdraw auto reconcile unexpected error")
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        logger.info("withdraw auto reconcile stopped")
        raise


def start_withdraw_reconcile_worker() -> asyncio.Task | None:
    if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
        # 部署级总闸关:worker 进程不启动(运营后台开关此时无效,需先在 env 打开)。
        logger.info("withdraw auto reconcile worker not started (env master switch off)")
        return None
    if not settings.wxpay_configured:
        logger.warning("withdraw auto reconcile enabled but wxpay not configured")
    return asyncio.create_task(_run_loop(), name="withdraw-auto-reconcile")


async def stop_withdraw_reconcile_worker(task: asyncio.Task | None) -> None:
    if task is None:
        return
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task