"""
串联执行的供给数据流水线（可重复执行、幂等）：

1. ODPS → MySQL 全局树同步（T-1 分区）
2. ODPS → MySQL 策略需求池同步（当天 biz_dt）
3. 需求池分级评估（同上 biz_dt）
4. S/A 需求视频点位拓展判断

各子步骤内部已做去重（INSERT IGNORE、diff 同步、跳过已分级词等）；
本文件额外用进程内锁防止同一轮次并发重入，并隔离各步骤异常：前一步失败时记录
error 后继续后续步骤，最终返回失败结果而不向 APScheduler 抛异常。
"""
from __future__ import annotations

import logging
import threading
from datetime import datetime, timedelta
from typing import Any, Callable
from zoneinfo import ZoneInfo

from supply_infra.config import get_infra_settings
from supply_infra.scheduler.constants import (
    SUPPLY_PIPELINE_JOB_ID,
    SUPPLY_PIPELINE_JOB_NAME,
)
from supply_infra.scheduler.job_execution import JobExecutionRecorder, record_skipped
from supply_infra.scheduler.jobs.expand_demand_from_video_points import (
    expand_demand_from_video_points,
)
from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
from supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql import sync_global_tree_odps_to_mysql
from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
    sync_multi_demand_pool_odps_to_mysql,
)

logger = logging.getLogger(__name__)

_pipeline_lock = threading.Lock()


def _resolve_dates(biz_dt: str | None) -> tuple[str, str]:
    """返回 (biz_dt, global_tree_partition_date)，global_tree 使用 biz_dt 前一日。"""
    if biz_dt:
        resolved_biz_dt = biz_dt
    else:
        timezone = ZoneInfo(get_infra_settings().scheduler_timezone)
        resolved_biz_dt = datetime.now(timezone).strftime("%Y%m%d")
    tree_partition = (
        datetime.strptime(resolved_biz_dt, "%Y%m%d") - timedelta(days=1)
    ).strftime("%Y%m%d")
    return resolved_biz_dt, tree_partition


def _run_step(
    step_name: str,
    action: Callable[[], Any],
) -> tuple[Any, bool, str | None]:
    """执行单个流水线步骤；失败只转成结果，不允许异常越过定时任务边界。"""
    try:
        payload = action()
    except Exception as exc:
        logger.exception("Supply pipeline step failed: step=%s", step_name)
        return {"success": False, "error": str(exc)}, False, str(exc)

    if isinstance(payload, dict) and payload.get("success") is False:
        error = str(payload.get("error") or f"{step_name} returned success=False")
        logger.error("Supply pipeline step reported failure: step=%s error=%s", step_name, error)
        return payload, False, error
    return payload, True, None


def _preflight_failure_result(biz_dt: str | None, exc: Exception) -> dict[str, Any]:
    """日期/配置解析失败时也记录结果并正常返回，避免调度线程出现未捕获异常。"""
    started_at = datetime.now()
    recorder = JobExecutionRecorder(
        job_name=SUPPLY_PIPELINE_JOB_NAME,
        job_id=SUPPLY_PIPELINE_JOB_ID,
        biz_dt=str(biz_dt) if biz_dt is not None else None,
    )
    recorder.record_started()
    result: dict[str, Any] = {
        "run_id": recorder.run_id,
        "biz_dt": str(biz_dt) if biz_dt is not None else None,
        "success": False,
        "error": str(exc),
        "started_at": started_at.isoformat(),
    }
    finished_at = datetime.now()
    result["finished_at"] = finished_at.isoformat()
    result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
    recorder.record_finished(success=False, result=result, error_message=str(exc))
    return result


def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
    """
    按顺序执行全局树同步 → 需求池同步 → 需求分级 → 视频点位拓展。

    Args:
        biz_dt: 业务日 YYYYMMDD；省略则取当天。

    Returns:
        各步骤统计；若上一轮仍在执行则返回 skipped。
    """
    try:
        resolved_biz_dt, tree_partition = _resolve_dates(biz_dt)
    except Exception as exc:
        logger.exception("Supply pipeline preflight failed: biz_dt=%s", biz_dt)
        return _preflight_failure_result(biz_dt, exc)

    if not _pipeline_lock.acquire(blocking=False):
        logger.warning("Supply pipeline already running, skip this round")
        record_skipped(
            job_name=SUPPLY_PIPELINE_JOB_NAME,
            job_id=SUPPLY_PIPELINE_JOB_ID,
            biz_dt=resolved_biz_dt,
            reason="already_running",
        )
        return {
            "skipped": True,
            "reason": "already_running",
            "run_at": datetime.now().isoformat(),
        }

    started_at = datetime.now()
    recorder = JobExecutionRecorder(
        job_name=SUPPLY_PIPELINE_JOB_NAME,
        job_id=SUPPLY_PIPELINE_JOB_ID,
        biz_dt=resolved_biz_dt,
    )
    recorder.record_started()

    logger.info(
        "Supply pipeline start: biz_dt=%s global_tree_partition=%s run_id=%s",
        resolved_biz_dt,
        tree_partition,
        recorder.run_id,
    )

    result: dict[str, Any] = {
        "run_id": recorder.run_id,
        "biz_dt": resolved_biz_dt,
        "global_tree_partition": tree_partition,
        "started_at": started_at.isoformat(),
    }
    errors: list[str] = []
    step_status: dict[str, dict[str, Any]] = {}
    success = False

    try:
        steps: list[tuple[str, Callable[[], Any]]] = [
            (
                "global_tree",
                lambda: sync_global_tree_odps_to_mysql(partition_date=tree_partition),
            ),
            (
                "demand_pool",
                lambda: sync_multi_demand_pool_odps_to_mysql(
                    partition_date=resolved_biz_dt,
                ),
            ),
            (
                "grade",
                lambda: grade_demand_pool(biz_dt=resolved_biz_dt),
            ),
            (
                "expand_video_points",
                lambda: expand_demand_from_video_points(
                    biz_dt=resolved_biz_dt,
                    workers=5,
                ),
            ),
        ]
        for step_name, action in steps:
            payload, step_success, step_error = _run_step(step_name, action)
            result[step_name] = payload
            step_status[step_name] = {
                "success": step_success,
                "error": step_error,
            }
            if step_error:
                errors.append(f"{step_name}: {step_error}")

        success = all(item["success"] for item in step_status.values())
        result["success"] = success
    except Exception as exc:
        # 保护流水线编排本身；正常子步骤异常应已由 _run_step 消化。
        result["success"] = False
        errors.append(f"pipeline: {exc}")
        logger.exception(
            "Supply pipeline orchestration failed but will not escape scheduler: "
            "biz_dt=%s global_tree_partition=%s",
            resolved_biz_dt,
            tree_partition,
        )
    finally:
        finished_at = datetime.now()
        result["steps"] = step_status
        if errors:
            result["errors"] = errors
        result["finished_at"] = finished_at.isoformat()
        result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
        try:
            recorder.record_finished(
                success=success,
                result=result,
                error_message=" | ".join(errors) if errors else None,
            )
        except Exception:
            # recorder 当前已自行兜底；这里防止未来实现变化造成锁无法释放。
            logger.exception("Unexpected failure while recording supply pipeline finish")
        finally:
            _pipeline_lock.release()
        logger.info("Supply pipeline finished: %s", result)

    return result
