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

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

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

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, engine

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 _new_run_id(now: datetime) -> str:
    return f"{now:%Y%m%d-%H%M%S}-{uuid4().hex[:8]}"


def _next_run_at(now: datetime, run_hour: int) -> datetime:
    scheduled = now.replace(hour=run_hour, minute=0, second=0, microsecond=0)
    return scheduled if now < scheduled else scheduled + timedelta(days=1)


def _trigger_for_run(worker_started_at: datetime, now: datetime, run_hour: int) -> str:
    if worker_started_at.date() == now.date() and worker_started_at.hour >= run_hour:
        return "startup_catchup"
    return "scheduled"


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,
        "api_requests": 0,
    }


def _platform_log_fields(platform: str, result: dict) -> dict:
    fields = {
        "platform": platform,
        "platform_status": result["status"],
        "duration_ms": result["duration_ms"],
    }
    for key in ("fetched", "inserted", "updated", "pages", "api_requests", "windows"):
        if key in result:
            fields[key] = result[key]
    return fields


def _run_platform(
    *,
    platform: str,
    query_time_type: int,
    common: dict,
    reconcile: Callable[[], dict],
) -> tuple[dict, str | None]:
    started = time.perf_counter()
    logger.info(
        "CPS auto reconcile platform started run_id=%s platform=%s",
        common["run_id"],
        platform,
        extra={
            **common,
            "event": "cps_reconcile.platform_started",
            "platform": platform,
            "query_time_type": query_time_type,
        },
    )
    try:
        raw_result = reconcile()
    except Exception as exc:  # noqa: BLE001 - 单平台失败不能阻断另一平台
        result = {
            "status": "failed",
            "duration_ms": round((time.perf_counter() - started) * 1000, 3),
            "error_type": type(exc).__name__,
            "error_summary": str(exc)[:500],
        }
        logger.exception(
            "CPS auto reconcile platform failed run_id=%s platform=%s error_type=%s",
            common["run_id"],
            platform,
            type(exc).__name__,
            extra={
                **common,
                "event": "cps_reconcile.platform_failed",
                **_platform_log_fields(platform, result),
                "error_type": type(exc).__name__,
                "error_summary": str(exc)[:500],
            },
        )
        return result, str(exc)

    result = {
        **raw_result,
        "status": "success",
        "duration_ms": round((time.perf_counter() - started) * 1000, 3),
    }
    logger.info(
        "CPS auto reconcile platform completed run_id=%s platform=%s fetched=%s inserted=%s updated=%s",
        common["run_id"],
        platform,
        result.get("fetched", 0),
        result.get("inserted", 0),
        result.get("updated", 0),
        extra={
            **common,
            "event": "cps_reconcile.platform_completed",
            **_platform_log_fields(platform, result),
        },
    )
    return result, None


def _skip_platform(platform: str, common: dict) -> dict:
    result = {
        **_empty_result(),
        "status": "skipped",
        "skipped": "not_configured",
        "duration_ms": 0.0,
    }
    logger.warning(
        "CPS auto reconcile platform skipped run_id=%s platform=%s reason=not_configured",
        common["run_id"],
        platform,
        extra={
            **common,
            "event": "cps_reconcile.platform_skipped",
            **_platform_log_fields(platform, result),
            "skip_reason": "not_configured",
        },
    )
    return result


def _reconcile_once(
    now: datetime | None = None,
    *,
    run_id: str | None = None,
    trigger: str = "scheduled",
) -> dict:
    """独立拉取美团和京东；单平台异常只记日志，不影响另一平台。"""
    end = (now or _cn_now()).astimezone(CN_TZ)
    run_id = run_id or _new_run_id(end)
    lookback_days = max(1, int(settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS))
    start = end - timedelta(days=lookback_days)
    task_started = time.perf_counter()
    common = {
        "run_id": run_id,
        "trigger": trigger,
        "scheduled_date": end.date().isoformat(),
        "app_env": settings.APP_ENV,
        "db_dialect": engine.dialect.name,
        "hostname": socket.gethostname(),
        "pid": os.getpid(),
        "window_start": start.isoformat(),
        "window_end": end.isoformat(),
        "lookback_days": lookback_days,
    }
    result = {
        "run_id": run_id,
        "trigger": trigger,
        "window_start": start.isoformat(),
        "window_end": end.isoformat(),
        "meituan": None,
        "jd": None,
        "errors": {},
    }
    logger.info(
        "CPS auto reconcile started run_id=%s trigger=%s window=%s..%s",
        run_id,
        trigger,
        result["window_start"],
        result["window_end"],
        extra={**common, "event": "cps_reconcile.started"},
    )

    if settings.mt_cps_configured:
        def reconcile_meituan() -> dict:
            with SessionLocal() as db:
                return cps_repo.reconcile_orders(
                    db,
                    start_time=int(start.timestamp()),
                    end_time=int(end.timestamp()),
                    query_time_type=2,
                    audit_context=common,
                )

        result["meituan"], error = _run_platform(
            platform="meituan",
            query_time_type=2,
            common=common,
            reconcile=reconcile_meituan,
        )
        if error is not None:
            result["errors"]["meituan"] = error
    else:
        result["meituan"] = _skip_platform("meituan", common)

    _touch_lock()
    if settings.jd_union_configured:
        def reconcile_jd() -> dict:
            with SessionLocal() as db:
                return cps_repo.reconcile_jd_orders(
                    db,
                    start_time=start,
                    end_time=end,
                    query_time_type=3,
                    audit_context=common,
                )

        result["jd"], error = _run_platform(
            platform="jd",
            query_time_type=3,
            common=common,
            reconcile=reconcile_jd,
        )
        if error is not None:
            result["errors"]["jd"] = error
    else:
        result["jd"] = _skip_platform("jd", common)

    successes = sum(
        platform_result.get("status") == "success"
        for platform_result in (result["meituan"], result["jd"])
    )
    if result["errors"]:
        task_status = "partial_success" if successes else "failed"
    else:
        task_status = "success"
    completed_at = _cn_now()
    duration_ms = round((time.perf_counter() - task_started) * 1000, 3)
    next_run_at = _next_run_at(
        completed_at,
        min(max(int(settings.CPS_AUTO_RECONCILE_RUN_HOUR), 0), 23),
    ).isoformat()
    result.update({
        "status": task_status,
        "duration_ms": duration_ms,
        "manual_retry_required": bool(result["errors"]),
        "next_run_at": next_run_at,
    })
    completion_fields = {
        **common,
        "event": "cps_reconcile.completed",
        "task_status": task_status,
        "duration_ms": duration_ms,
        "manual_retry_required": result["manual_retry_required"],
        "failed_platforms": sorted(result["errors"]),
        "next_run_at": next_run_at,
        "meituan_result": result["meituan"],
        "jd_result": result["jd"],
    }
    log_method = logger.info if task_status == "success" else logger.warning
    log_method(
        "CPS auto reconcile completed run_id=%s status=%s duration_ms=%s failed_platforms=%s next_run_at=%s",
        run_id,
        task_status,
        duration_ms,
        ",".join(sorted(result["errors"])) or "-",
        next_run_at,
        extra=completion_fields,
    )

    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 worker skipped: another worker owns lock",
                extra={
                    "event": "cps_reconcile.worker_skipped",
                    "skip_reason": "lock_not_acquired",
                    "pid": os.getpid(),
                },
            )
            return
        await _run_locked_loop(interval, run_hour)


async def _run_locked_loop(interval: int, run_hour: int) -> None:
    worker_started_at = _cn_now()
    logger.info(
        "CPS auto reconcile worker started run_hour=%s interval=%ss lookback_days=%s",
        run_hour,
        interval,
        settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
        extra={
            "event": "cps_reconcile.worker_started",
            "pid": os.getpid(),
            "hostname": socket.gethostname(),
            "app_env": settings.APP_ENV,
            "db_dialect": engine.dialect.name,
            "run_hour": run_hour,
            "interval_sec": interval,
            "lookback_days": settings.CPS_AUTO_RECONCILE_LOOKBACK_DAYS,
            "next_run_at": _next_run_at(worker_started_at, run_hour).isoformat(),
        },
    )
    last_run: date | None = None
    try:
        while True:
            run_id: str | None = None
            try:
                _touch_lock()
                now = _cn_now()
                if _should_run(last_run, now, run_hour):
                    run_id = _new_run_id(now)
                    trigger = _trigger_for_run(worker_started_at, now, run_hour)
                    await asyncio.to_thread(
                        _reconcile_once,
                        now,
                        run_id=run_id,
                        trigger=trigger,
                    )
                    last_run = now.date()
            except Exception:  # noqa: BLE001 - 后台任务不能因单次异常退出
                logger.exception(
                    "CPS auto reconcile unexpected worker error",
                    extra={
                        "event": "cps_reconcile.unexpected_failed",
                        "run_id": run_id or "unassigned",
                        "pid": os.getpid(),
                    },
                )
            await asyncio.sleep(interval)
    except asyncio.CancelledError:
        logger.info(
            "CPS auto reconcile worker stopped",
            extra={"event": "cps_reconcile.worker_stopped", "pid": os.getpid()},
        )
        raise


def start_cps_reconcile_worker() -> asyncio.Task | None:
    if not settings.CPS_AUTO_RECONCILE_ENABLED:
        logger.info(
            "CPS auto reconcile disabled",
            extra={
                "event": "cps_reconcile.worker_skipped",
                "skip_reason": "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",
            extra={
                "event": "cps_reconcile.worker_skipped",
                "skip_reason": "all_platform_credentials_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