# -*- coding: utf-8 -*-
"""全量刷新脚本：执行 2026-01-01 ~ 2026-02-16 的 api_full Flow。

按层逐步执行（ODS → DWD → DWS → INDEX），内嵌精细性能计时，
支持断点续跑（从指定层/任务重试），完成后执行 increment_verify 校验，
校验不一致时自动补齐。计时数据和执行统计写入 JSON 中间文件。

用法:
    cd apps/etl/connectors/feiqiu
    python -m scripts.debug.run_full_refresh [--resume-layer DWS] [--resume-task DWS_FINANCE_DAILY]
"""
from __future__ import annotations

import argparse
import json
import logging
import sys
import time
import traceback
import uuid
from dataclasses import asdict, dataclass, field
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

# ── 确保项目根目录在 sys.path ──
_FEIQIU_ROOT = Path(__file__).resolve().parents[2]
if str(_FEIQIU_ROOT) not in sys.path:
    sys.path.insert(0, str(_FEIQIU_ROOT))

from config.settings import AppConfig
from database.connection import DatabaseConnection
from database.operations import DatabaseOperations
from api.client import APIClient
from orchestration.task_registry import default_registry
from orchestration.cursor_manager import CursorManager
from orchestration.run_tracker import RunTracker
from orchestration.task_executor import TaskExecutor
from orchestration.flow_runner import FlowRunner


# ── 常量 ──────────────────────────────────────────────────────

FLOW_NAME = "api_full"
LAYERS = FlowRunner.FLOW_LAYERS[FLOW_NAME]  # ["ODS", "DWD", "DWS", "INDEX"]
WINDOW_START_STR = "2026-01-01T00:00:00"
WINDOW_END_STR = "2026-02-16T00:00:00"


# ── 数据结构 ──────────────────────────────────────────────────

@dataclass
class TaskTiming:
    """单个任务的计时与执行统计"""
    task_code: str
    layer: str
    start_time: str = ""
    end_time: str = ""
    duration_sec: float = 0.0
    status: str = ""          # SUCCESS / FAIL / ERROR / SKIP
    counts: dict = field(default_factory=dict)
    error: str | None = None
    api_calls: int = 0
    api_total_sec: float = 0.0


@dataclass
class LayerTiming:
    """单层的计时与汇总"""
    layer: str
    start_time: str = ""
    end_time: str = ""
    duration_sec: float = 0.0
    status: str = ""          # SUCCESS / PARTIAL / ERROR
    task_count: int = 0
    success_count: int = 0
    fail_count: int = 0
    skip_count: int = 0
    total_fetched: int = 0
    total_inserted: int = 0
    total_updated: int = 0
    total_errors: int = 0
    tasks: list[TaskTiming] = field(default_factory=list)


@dataclass
class RefreshReport:
    """全量刷新的完整执行报告"""
    flow: str = FLOW_NAME
    window_start: str = WINDOW_START_STR
    window_end: str = WINDOW_END_STR
    overall_start: str = ""
    overall_end: str = ""
    overall_duration_sec: float = 0.0
    overall_status: str = ""
    layers: list[LayerTiming] = field(default_factory=list)
    verification: dict = field(default_factory=dict)
    environment: dict = field(default_factory=dict)


# ── 工具函数 ──────────────────────────────────────────────────

def _setup_logging() -> logging.Logger:
    logger = logging.getLogger("full_refresh")
    logger.setLevel(logging.INFO)
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter(
            "%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S"
        ))
        logger.addHandler(handler)
    return logger


def _now_iso(tz: ZoneInfo) -> str:
    return datetime.now(tz).isoformat()


def _build_components(config: AppConfig, logger: logging.Logger):
    """构建 DB / API / TaskExecutor / FlowRunner 等组件。"""
    db_conn = DatabaseConnection(
        dsn=config["db"]["dsn"],
        session=config["db"].get("session"),
        connect_timeout=config["db"].get("connect_timeout_sec"),
    )
    api_client = APIClient(
        base_url=config["api"]["base_url"],
        token=config["api"]["token"],
        timeout=config["api"].get("timeout_sec", 20),
        retry_max=config["api"].get("retries", {}).get("max_attempts", 3),
        headers_extra=config["api"].get("headers_extra"),
    )
    db_ops = DatabaseOperations(db_conn)
    cursor_mgr = CursorManager(db_conn)
    run_tracker = RunTracker(db_conn)

    executor = TaskExecutor(
        config, db_ops, api_client,
        cursor_mgr, run_tracker, default_registry, logger,
    )
    runner = FlowRunner(
        config, executor, default_registry,
        db_conn, api_client, logger,
    )
    return db_conn, api_client, db_ops, executor, runner


def _resolve_layer_tasks(layer: str, config: AppConfig) -> list[str]:
    """解析单层的任务列表，与 FlowRunner._resolve_tasks 逻辑一致。"""
    layer_upper = layer.upper()

    if layer_upper == "ODS":
        ods_tasks = config.get("run.ods_tasks", [])
        if ods_tasks:
            return list(ods_tasks)
        registry_tasks = default_registry.get_tasks_by_layer("ODS")
        return sorted(registry_tasks) if registry_tasks else []

    elif layer_upper == "DWD":
        return ["DWD_LOAD_FROM_ODS"]

    elif layer_upper == "DWS":
        dws_tasks = config.get("run.dws_tasks", [])
        if dws_tasks:
            return list(dws_tasks)
        registry_tasks = default_registry.get_tasks_by_layer("DWS")
        return sorted(registry_tasks) if registry_tasks else []

    elif layer_upper == "INDEX":
        index_tasks = config.get("run.index_tasks", [])
        if index_tasks:
            return list(index_tasks)
        registry_tasks = default_registry.get_tasks_by_layer("INDEX")
        return sorted(registry_tasks) if registry_tasks else []

    return []



def _sanitize_for_json(obj):
    """递归处理不可序列化的值。"""
    if isinstance(obj, dict):
        return {k: _sanitize_for_json(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_sanitize_for_json(v) for v in obj]
    if isinstance(obj, datetime):
        return obj.isoformat()
    return obj


def _save_json(data, path: Path):
    """将数据序列化为 JSON 文件。"""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(_sanitize_for_json(data), ensure_ascii=False, indent=2, default=str),
        encoding="utf-8",
    )


def _load_checkpoint(path: Path) -> dict | None:
    """加载断点续跑的检查点文件。"""
    if path.exists():
        try:
            return json.loads(path.read_text(encoding="utf-8"))
        except Exception:
            return None
    return None


def _save_checkpoint(path: Path, data: dict):
    """保存断点续跑的检查点。"""
    _save_json(data, path)


# ── 单任务执行（带精细计时）─────────────────────────────────

def _execute_task_with_timing(
    task_code: str,
    layer: str,
    executor: TaskExecutor,
    config: AppConfig,
    db_conn: DatabaseConnection,
    logger: logging.Logger,
    tz: ZoneInfo,
) -> TaskTiming:
    """执行单个任务并记录精细计时。"""
    timing = TaskTiming(task_code=task_code, layer=layer)
    store_id = int(config.get("app.store_id"))
    run_uuid = f"full-refresh-{task_code.lower()}-{uuid.uuid4().hex[:8]}"

    timing.start_time = _now_iso(tz)
    t0 = time.monotonic()

    try:
        task_result = executor.run_single_task(
            task_code=task_code,
            run_uuid=run_uuid,
            store_id=store_id,
            data_source="online",
        )
        timing.duration_sec = round(time.monotonic() - t0, 3)
        timing.end_time = _now_iso(tz)

        # 解析结果
        raw_status = (task_result.get("status") or "").upper()
        counts = task_result.get("counts") or {}
        timing.counts = counts
        timing.status = raw_status if raw_status else "COMPLETE"

        # 尝试提取 API 调用统计（如果任务结果中包含）
        api_stats = task_result.get("api_stats") or {}
        timing.api_calls = api_stats.get("calls", 0)
        timing.api_total_sec = api_stats.get("total_sec", 0.0)

        logger.info(
            "  ✓ %s: %s (%.1fs) fetched=%s inserted=%s updated=%s errors=%s",
            task_code, timing.status, timing.duration_sec,
            counts.get("fetched", 0), counts.get("inserted", 0),
            counts.get("updated", 0), counts.get("errors", 0),
        )

    except Exception as exc:
        timing.duration_sec = round(time.monotonic() - t0, 3)
        timing.end_time = _now_iso(tz)
        timing.status = "ERROR"
        timing.error = str(exc)
        logger.error("  ✗ %s: 异常 (%.1fs): %s", task_code, timing.duration_sec, exc)
        # CHANGE 2026-02-16 | 任务异常后 rollback，防止 InFailedSqlTransaction 级联
        try:
            db_conn.rollback()
        except Exception:
            pass

    return timing


# ── 单层执行（带断点续跑）───────────────────────────────────

def _execute_layer(
    layer: str,
    config: AppConfig,
    executor: TaskExecutor,
    db_conn: DatabaseConnection,
    logger: logging.Logger,
    tz: ZoneInfo,
    resume_task: str | None = None,
    checkpoint_path: Path | None = None,
) -> LayerTiming:
    """执行单层所有任务，支持从指定任务恢复。"""
    layer_timing = LayerTiming(layer=layer)
    layer_timing.start_time = _now_iso(tz)
    layer_t0 = time.monotonic()

    tasks = _resolve_layer_tasks(layer, config)
    layer_timing.task_count = len(tasks)

    logger.info("━" * 70)
    logger.info("▶ 层 %s: %d 个任务", layer, len(tasks))
    if tasks:
        logger.info("  任务列表: %s", ", ".join(tasks))

    # 断点续跑：跳过 resume_task 之前的任务
    skip_until_found = False
    if resume_task:
        resume_upper = resume_task.upper()
        if resume_upper in [t.upper() for t in tasks]:
            skip_until_found = True
            logger.info("  断点续跑: 从 %s 开始", resume_upper)
        else:
            logger.warning("  断点续跑: %s 不在本层任务列表中，执行全部", resume_upper)

    for idx, task_code in enumerate(tasks, start=1):
        # 断点续跑逻辑
        if skip_until_found:
            if task_code.upper() == resume_upper:
                skip_until_found = False
                logger.info("  [%d/%d] ▶ 恢复执行: %s", idx, len(tasks), task_code)
            else:
                logger.info("  [%d/%d] ⏭ 跳过: %s (断点续跑)", idx, len(tasks), task_code)
                skipped = TaskTiming(
                    task_code=task_code, layer=layer, status="SKIPPED_RESUME",
                )
                layer_timing.tasks.append(skipped)
                layer_timing.skip_count += 1
                continue
        else:
            logger.info("  [%d/%d] %s", idx, len(tasks), task_code)

        timing = _execute_task_with_timing(
            task_code, layer, executor, config, db_conn, logger, tz,
        )
        layer_timing.tasks.append(timing)

        # 统计
        if timing.status in ("SUCCESS", "成功", "COMPLETE", "PARTIAL"):
            layer_timing.success_count += 1
        elif timing.status == "ERROR":
            layer_timing.fail_count += 1
        elif timing.status == "SKIP":
            layer_timing.skip_count += 1
        else:
            layer_timing.success_count += 1  # 未知状态视为成功

        counts = timing.counts
        layer_timing.total_fetched += counts.get("fetched", 0)
        layer_timing.total_inserted += counts.get("inserted", 0)
        layer_timing.total_updated += counts.get("updated", 0)
        layer_timing.total_errors += counts.get("errors", 0)

        # 保存检查点（每个任务完成后）
        if checkpoint_path:
            _save_checkpoint(checkpoint_path, {
                "last_completed_layer": layer,
                "last_completed_task": task_code,
                "timestamp": _now_iso(tz),
            })

        # 确保连接可用
        db_conn.ensure_open()

    layer_timing.duration_sec = round(time.monotonic() - layer_t0, 3)
    layer_timing.end_time = _now_iso(tz)

    # 层状态判定
    if layer_timing.fail_count == 0:
        layer_timing.status = "SUCCESS"
    elif layer_timing.success_count > 0:
        layer_timing.status = "PARTIAL"
    else:
        layer_timing.status = "ERROR"

    logger.info(
        "  层 %s 完成: %s (%.1fs) 成功=%d 失败=%d 跳过=%d",
        layer, layer_timing.status, layer_timing.duration_sec,
        layer_timing.success_count, layer_timing.fail_count, layer_timing.skip_count,
    )
    logger.info(
        "  汇总: fetched=%d inserted=%d updated=%d errors=%d",
        layer_timing.total_fetched, layer_timing.total_inserted,
        layer_timing.total_updated, layer_timing.total_errors,
    )

    return layer_timing



# ── 校验阶段 ──────────────────────────────────────────────────

def _run_verification(
    runner: FlowRunner,
    config: AppConfig,
    window_start: datetime,
    window_end: datetime,
    logger: logging.Logger,
    tz: ZoneInfo,
) -> dict:
    """执行 increment_verify 校验，发现不一致时自动补齐。"""
    logger.info("")
    logger.info("=" * 70)
    logger.info("▶ 开始 increment_verify 校验")
    logger.info("=" * 70)

    verify_start = _now_iso(tz)
    t0 = time.monotonic()

    try:
        # 使用 FlowRunner 的内部校验方法
        verification_summary = runner._run_verification(
            layers=LAYERS,
            window_start=window_start,
            window_end=window_end,
            window_split="month",
        )

        duration = round(time.monotonic() - t0, 3)
        verify_end = _now_iso(tz)

        result = {
            "status": verification_summary.get("status", "UNKNOWN"),
            "start_time": verify_start,
            "end_time": verify_end,
            "duration_sec": duration,
            "total_tables": verification_summary.get("total_tables", 0),
            "consistent_tables": verification_summary.get("consistent_tables", 0),
            "total_backfilled": verification_summary.get("total_backfilled", 0),
            "error_tables": verification_summary.get("error_tables", 0),
            "layers": verification_summary.get("layers", {}),
        }

        logger.info(
            "  校验完成: %s (%.1fs) 表数=%d 一致=%d 补齐=%d 错误=%d",
            result["status"], duration,
            result["total_tables"], result["consistent_tables"],
            result["total_backfilled"], result["error_tables"],
        )

        # 如果有补齐，记录详情
        if result["total_backfilled"] > 0:
            logger.info("  ℹ 已自动补齐 %d 处不一致", result["total_backfilled"])

        return result

    except Exception as exc:
        duration = round(time.monotonic() - t0, 3)
        logger.error("  ✗ 校验异常 (%.1fs): %s", duration, exc)
        return {
            "status": "ERROR",
            "start_time": verify_start,
            "end_time": _now_iso(tz),
            "duration_sec": duration,
            "error": str(exc),
            "traceback": traceback.format_exc(),
        }


# ── 主流程 ────────────────────────────────────────────────────

def run_full_refresh(
    resume_layer: str | None = None,
    resume_task: str | None = None,
    skip_verify: bool = False,
) -> RefreshReport:
    """执行全量刷新。

    Args:
        resume_layer: 从指定层开始执行（断点续跑），如 "DWS"
        resume_task: 在恢复层中从指定任务开始（断点续跑），如 "DWS_FINANCE_DAILY"
        skip_verify: 跳过校验阶段（调试用）
    Returns:
        RefreshReport 完整执行报告
    """
    logger = _setup_logging()
    logger.info("=" * 70)
    logger.info("全量刷新开始")
    logger.info("Flow: %s | 窗口: %s ~ %s", FLOW_NAME, WINDOW_START_STR, WINDOW_END_STR)
    logger.info("=" * 70)

    # 加载配置
    config = AppConfig.load()
    tz = ZoneInfo(config.get("app.timezone", "Asia/Shanghai"))
    window_start = datetime.fromisoformat(WINDOW_START_STR).replace(tzinfo=tz)
    window_end = datetime.fromisoformat(WINDOW_END_STR).replace(tzinfo=tz)

    report = RefreshReport()
    report.overall_start = _now_iso(tz)
    report.environment = {
        "store_id": config.get("app.store_id"),
        "db_name": config.get("db.name", ""),
        "api_base_url": config.get("api.base_url", ""),
        "timezone": str(tz),
    }

    logger.info("门店 ID: %s", config.get("app.store_id"))
    logger.info("数据库: %s", config.get("db.name", ""))
    logger.info("API: %s", config.get("api.base_url", ""))

    # 设置 window_override 让所有任务使用统一的全量窗口
    config.config.setdefault("run", {}).setdefault("window_override", {})
    config.config["run"]["window_override"]["start"] = window_start
    config.config["run"]["window_override"]["end"] = window_end

    # 构建组件
    db_conn, api_client, db_ops, executor, runner = _build_components(config, logger)

    # 输出目录和检查点
    output_dir = _FEIQIU_ROOT / "scripts" / "debug" / "output"
    output_dir.mkdir(parents=True, exist_ok=True)
    checkpoint_path = output_dir / "full_refresh_checkpoint.json"

    overall_t0 = time.monotonic()

    # 确定从哪一层开始
    layers_to_run = list(LAYERS)
    if resume_layer:
        resume_layer_upper = resume_layer.upper()
        layer_names_upper = [l.upper() for l in layers_to_run]
        if resume_layer_upper in layer_names_upper:
            start_idx = layer_names_upper.index(resume_layer_upper)
            skipped_layers = layers_to_run[:start_idx]
            layers_to_run = layers_to_run[start_idx:]
            if skipped_layers:
                logger.info("断点续跑: 跳过层 %s，从 %s 开始", skipped_layers, resume_layer_upper)
        else:
            logger.warning("断点续跑: 层 %s 不在 Flow 定义中，执行全部", resume_layer_upper)

    # 逐层执行
    for layer_idx, layer in enumerate(layers_to_run):
        # 仅第一个恢复层使用 resume_task
        current_resume_task = resume_task if (layer_idx == 0 and resume_layer) else None

        layer_timing = _execute_layer(
            layer=layer,
            config=config,
            executor=executor,
            db_conn=db_conn,
            logger=logger,
            tz=tz,
            resume_task=current_resume_task,
            checkpoint_path=checkpoint_path,
        )
        report.layers.append(layer_timing)

        # 层执行后检查结果
        if layer_timing.status == "ERROR":
            logger.warning("")
            logger.warning("⚠ 层 %s 全部失败，后续层可能受影响", layer)
            logger.warning("  可使用 --resume-layer %s 从此层重试", layer)

        # 每层完成后保存中间结果（防止中途崩溃丢失数据）
        _save_intermediate_report(report, output_dir, tz)

    # 校验阶段
    if not skip_verify:
        report.verification = _run_verification(
            runner, config, window_start, window_end, logger, tz,
        )
    else:
        logger.info("")
        logger.info("⏭ 跳过校验阶段 (--skip-verify)")
        report.verification = {"status": "SKIPPED"}

    # 汇总
    report.overall_duration_sec = round(time.monotonic() - overall_t0, 3)
    report.overall_end = _now_iso(tz)

    all_success = all(lt.status == "SUCCESS" for lt in report.layers)
    any_error = any(lt.status == "ERROR" for lt in report.layers)
    if all_success:
        report.overall_status = "SUCCESS"
    elif any_error:
        report.overall_status = "PARTIAL"
    else:
        report.overall_status = "PARTIAL"

    # 打印汇总
    _print_summary(report, logger)

    # 保存最终结果
    ts = datetime.now(tz).strftime("%Y%m%d_%H%M%S")
    final_path = output_dir / f"full_refresh_{ts}.json"
    _save_json(asdict(report), final_path)
    logger.info("计时数据已保存: %s", final_path)

    # 清理检查点
    if checkpoint_path.exists() and report.overall_status == "SUCCESS":
        checkpoint_path.unlink()
        logger.info("检查点已清理")

    # 清理连接
    db_conn.close()
    return report


def _save_intermediate_report(report: RefreshReport, output_dir: Path, tz: ZoneInfo):
    """保存中间结果，防止中途崩溃丢失已采集的计时数据。"""
    intermediate_path = output_dir / "full_refresh_intermediate.json"
    _save_json(asdict(report), intermediate_path)


# ── 汇总输出 ──────────────────────────────────────────────────

def _print_summary(report: RefreshReport, logger: logging.Logger):
    """打印全量刷新汇总。"""
    logger.info("")
    logger.info("=" * 70)
    logger.info("全量刷新汇总")
    logger.info("=" * 70)
    logger.info("状态: %s | 总耗时: %.1fs", report.overall_status, report.overall_duration_sec)
    logger.info("")

    # 逐层统计
    logger.info("%-8s %-10s %8s %8s %8s %8s %10s", "层", "状态", "成功", "失败", "跳过", "任务数", "耗时(s)")
    logger.info("-" * 70)
    for lt in report.layers:
        logger.info(
            "%-8s %-10s %8d %8d %8d %8d %10.1f",
            lt.layer, lt.status, lt.success_count, lt.fail_count,
            lt.skip_count, lt.task_count, lt.duration_sec,
        )

    # 记录数汇总
    logger.info("")
    logger.info("记录数汇总:")
    total_fetched = sum(lt.total_fetched for lt in report.layers)
    total_inserted = sum(lt.total_inserted for lt in report.layers)
    total_updated = sum(lt.total_updated for lt in report.layers)
    total_errors = sum(lt.total_errors for lt in report.layers)
    logger.info("  fetched=%d  inserted=%d  updated=%d  errors=%d",
                total_fetched, total_inserted, total_updated, total_errors)

    # 耗时最长的 5 个任务
    all_tasks = []
    for lt in report.layers:
        all_tasks.extend(lt.tasks)
    top5 = sorted(
        [t for t in all_tasks if t.status not in ("SKIPPED_RESUME",)],
        key=lambda t: t.duration_sec,
        reverse=True,
    )[:5]
    if top5:
        logger.info("")
        logger.info("耗时 Top 5 任务:")
        for t in top5:
            logger.info("  %-30s %8.1fs  [%s] %s", t.task_code, t.duration_sec, t.layer, t.status)

    # 失败任务
    failed = [t for t in all_tasks if t.status == "ERROR"]
    if failed:
        logger.info("")
        logger.info("失败任务 (%d 个):", len(failed))
        for t in failed:
            logger.info("  ✗ %s [%s]: %s", t.task_code, t.layer, t.error or "未知错误")

    # 校验结果
    if report.verification:
        v = report.verification
        logger.info("")
        logger.info("校验结果: %s", v.get("status", "N/A"))
        if v.get("total_tables"):
            logger.info(
                "  表数=%d 一致=%d 补齐=%d 错误=%d",
                v.get("total_tables", 0), v.get("consistent_tables", 0),
                v.get("total_backfilled", 0), v.get("error_tables", 0),
            )


# ── CLI 入口 ──────────────────────────────────────────────────

def parse_args():
    parser = argparse.ArgumentParser(
        description="全量刷新: 执行 2026-01-01 ~ 2026-02-16 的 api_full Flow",
    )
    parser.add_argument(
        "--resume-layer", type=str, default=None,
        help="断点续跑: 从指定层开始（如 DWS）",
    )
    parser.add_argument(
        "--resume-task", type=str, default=None,
        help="断点续跑: 在恢复层中从指定任务开始（如 DWS_FINANCE_DAILY）",
    )
    parser.add_argument(
        "--skip-verify", action="store_true",
        help="跳过校验阶段",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    report = run_full_refresh(
        resume_layer=args.resume_layer,
        resume_task=args.resume_task,
        skip_verify=args.skip_verify,
    )

    # 退出码
    if report.overall_status == "SUCCESS":
        sys.exit(0)
    else:
        sys.exit(1)


if __name__ == "__main__":
    main()
