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

"""美团、京东 CPS 订单每日自动对账任务。

每天北京时间 `CPS_AUTO_RECONCILE_RUN_HOUR`（默认 05:00）后执行一次，按更新时间
回拉最近若干天订单并复用 admin CPS 仓储层的幂等 upsert。手动对账接口保持独立、不受影响。
"""
from __future__ import annotations

import asyncio
import contextlib
import logging
import os
import time
from collections.abc import Iterator
from datetime import date, datetime, timedelta
from pathlib import Path

from app.admin.repositories import cps as cps_repo
from app.core.config import settings
from app.core.rewards import CN_TZ
from app.db.session import SessionLocal

logger = logging.getLogger("shagua.cps_reconcile")
_LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "cps_reconcile.lock"


def _cn_now() -> datetime:
    return datetime.now(CN_TZ)


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]:
    """同机多进程保护：同一时间只允许一个 CPS 自动对账 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 _empty_result() -> dict:
    return {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0}


def _reconcile_once(now: datetime | None = None) -> dict:
    """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
    end = (now or _cn_now()).astimezone(CN_TZ)
    lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
    start = end - timedelta(days=lookback_days)
    result = {
        "window_start": start.isoformat(),
        "window_end": end.isoformat(),
        "meituan": None,
        "jd": None,
        "errors": {},
    }

    if settings.mt_cps_configured:
        try:
            with SessionLocal() as db:
                result["meituan"] = cps_repo.reconcile_orders(
                    db,
                    start_time=int(start.timestamp()),
                    end_time=int(end.timestamp()),
                    query_time_type=2,
                )
        except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
            result["errors"]["meituan"] = str(exc)
            logger.exception("CPS auto reconcile failed platform=meituan")
    else:
        result["meituan"] = {**_empty_result(), "skipped": "not_configured"}

    _touch_lock()
    if settings.jd_union_configured:
        try:
            with SessionLocal() as db:
                result["jd"] = cps_repo.reconcile_jd_orders(
                    db,
                    start_time=start,
                    end_time=end,
                    query_time_type=3,
                )
        except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
            result["errors"]["jd"] = str(exc)
            logger.exception("CPS auto reconcile failed platform=jd")
    else:
        result["jd"] = {**_empty_result(), "skipped": "not_configured"}

    return result


def _should_run(last_run: date | None, now: datetime, run_hour: int) -> bool:
    return last_run != now.date() and now.hour >= run_hour


async def _run_loop() -> None:
    interval = max(30, int(settings.CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC))
    run_hour = min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23)
    lock_stale_after = max(interval * 3, 1800)
    with _single_instance_lock(lock_stale_after) as lock_acquired:
        if not lock_acquired:
            logger.warning("CPS auto reconcile skipped: another worker owns lock")
            return
        await _run_locked_loop(interval, run_hour)


async def _run_locked_loop(interval: int, run_hour: int) -> None:
    logger.info(
        "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
        run_hour,
        interval,
        settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
    )
    last_run: date | None = None
    try:
        while True:
            try:
                _touch_lock()
                now = _cn_now()
                if _should_run(last_run, now, run_hour):
                    result = await asyncio.to_thread(_reconcile_once, now)
                    last_run = now.date()
                    logger.info("CPS auto reconcile done date=%s result=%s", last_run, result)
            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
                logger.exception("CPS auto reconcile unexpected error")
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        logger.info("CPS auto reconcile worker stopped")
        raise


def start_cps_reconcile_worker() -> asyncio.Task | None:
    if not settings.CPS_AUTO_RECONCILE_ENABLED:
        logger.info("CPS auto reconcile disabled")
        return None
    if not settings.mt_cps_configured and not settings.jd_union_configured:
        logger.warning("CPS auto reconcile not started: Meituan and JD credentials are missing")
        return None
    return asyncio.create_task(_run_loop(), name="cps-auto-reconcile")


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