"""
性能分析脚本 — 读取全量刷新阶段采集的计时 JSON，统计耗时、识别瓶颈、生成优化报告。

用法:
    cd apps/etl/connectors/feiqiu
    python -m scripts.debug.analyze_performance [--input <json>] [--output <md>] [--skip-sql]

功能:
    1. 层级耗时统计：各层总耗时、平均耗时、任务数
    2. 任务耗时排名：Top 5 瓶颈任务，含 fetched/inserted 等指标
    3. API 调用分析：响应时间、分页效率（每页记录数 vs 请求次数）
    4. SQL 查询分析：连接数据库执行 EXPLAIN ANALYZE 分析关键查询
    5. 优化建议：基于分析结果给出具体优化建议
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# 路径常量
# ---------------------------------------------------------------------------
_SCRIPT_DIR = Path(__file__).resolve().parent
_FEIQIU_ROOT = _SCRIPT_DIR.parent.parent          # apps/etl/connectors/feiqiu
_OUTPUT_DIR = _SCRIPT_DIR / "output"
_etl_report_root = os.environ.get("ETL_REPORT_ROOT")
if not _etl_report_root:
    raise KeyError("环境变量 ETL_REPORT_ROOT 未定义。请在根 .env 中配置。")
_REPORTS_DIR = Path(_etl_report_root)


# ---------------------------------------------------------------------------
# 数据结构
# ---------------------------------------------------------------------------
@dataclass
class TaskTiming:
    """单个任务的计时数据。"""
    task_code: str
    layer: str
    duration_sec: float
    status: str
    counts: dict[str, int]
    error: str | None
    api_calls: int
    api_total_sec: float

    @property
    def fetched(self) -> int:
        return self.counts.get("fetched", 0)

    @property
    def inserted(self) -> int:
        return self.counts.get("inserted", 0)

    @property
    def updated(self) -> int:
        return self.counts.get("updated", 0)

    @property
    def skipped(self) -> int:
        return self.counts.get("skipped", 0)

    @property
    def throughput(self) -> float:
        """每秒处理记录数（fetched / duration）。"""
        if self.duration_sec <= 0:
            return 0.0
        return self.fetched / self.duration_sec


@dataclass
class LayerTiming:
    """单层的汇总计时。"""
    layer: str
    duration_sec: float
    status: str
    task_count: int
    success_count: int
    fail_count: int
    skip_count: int
    total_fetched: int
    total_inserted: int
    total_updated: int
    total_errors: int
    tasks: list[TaskTiming]


@dataclass
class VerificationSummary:
    """校验阶段摘要。"""
    status: str
    duration_sec: float
    total_tables: int
    consistent_tables: int
    total_backfilled: int
    error_tables: int
    layers: dict[str, Any]


@dataclass
class PerformanceData:
    """完整的性能数据。"""
    flow: str
    window_start: str
    window_end: str
    overall_duration_sec: float
    overall_status: str
    layers: list[LayerTiming]
    verification: VerificationSummary | None


@dataclass
class SQLAnalysisResult:
    """SQL EXPLAIN ANALYZE 分析结果。"""
    query_name: str
    table_name: str
    plan_summary: str
    total_cost: float
    actual_time_ms: float
    rows_processed: int
    seq_scans: list[str]
    missing_indexes: list[str]
    recommendations: list[str]


@dataclass
class PerformanceReport:
    """性能分析报告数据。"""
    data: PerformanceData
    bottleneck_tasks: list[TaskTiming]
    layer_stats: list[dict[str, Any]]
    api_analysis: list[dict[str, Any]]
    sql_analysis: list[SQLAnalysisResult] = field(default_factory=list)
    recommendations: list[str] = field(default_factory=list)


# ---------------------------------------------------------------------------
# 日志
# ---------------------------------------------------------------------------
def _setup_logging() -> logging.Logger:
    logger = logging.getLogger("analyze_performance")
    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 load_timing_data(json_path: Path, logger: logging.Logger) -> PerformanceData:
    """从 JSON 文件加载计时数据。"""
    logger.info("加载计时数据: %s", json_path)
    raw = json.loads(json_path.read_text(encoding="utf-8"))

    layers: list[LayerTiming] = []
    for layer_raw in raw.get("layers", []):
        tasks = [
            TaskTiming(
                task_code=t["task_code"],
                layer=t.get("layer", layer_raw["layer"]),
                duration_sec=t.get("duration_sec", 0),
                status=t.get("status", "UNKNOWN"),
                counts=t.get("counts", {}),
                error=t.get("error"),
                api_calls=t.get("api_calls", 0),
                api_total_sec=t.get("api_total_sec", 0.0),
            )
            for t in layer_raw.get("tasks", [])
        ]
        layers.append(LayerTiming(
            layer=layer_raw["layer"],
            duration_sec=layer_raw.get("duration_sec", 0),
            status=layer_raw.get("status", "UNKNOWN"),
            task_count=layer_raw.get("task_count", len(tasks)),
            success_count=layer_raw.get("success_count", 0),
            fail_count=layer_raw.get("fail_count", 0),
            skip_count=layer_raw.get("skip_count", 0),
            total_fetched=layer_raw.get("total_fetched", 0),
            total_inserted=layer_raw.get("total_inserted", 0),
            total_updated=layer_raw.get("total_updated", 0),
            total_errors=layer_raw.get("total_errors", 0),
            tasks=tasks,
        ))

    verification = None
    if "verification" in raw:
        v = raw["verification"]
        verification = VerificationSummary(
            status=v.get("status", "UNKNOWN"),
            duration_sec=v.get("duration_sec", 0),
            total_tables=v.get("total_tables", 0),
            consistent_tables=v.get("consistent_tables", 0),
            total_backfilled=v.get("total_backfilled", 0),
            error_tables=v.get("error_tables", 0),
            layers=v.get("layers", {}),
        )

    return PerformanceData(
        flow=raw.get("flow", ""),
        window_start=raw.get("window_start", ""),
        window_end=raw.get("window_end", ""),
        overall_duration_sec=raw.get("overall_duration_sec", 0),
        overall_status=raw.get("overall_status", "UNKNOWN"),
        layers=layers,
        verification=verification,
    )


# ---------------------------------------------------------------------------
# 分析函数
# ---------------------------------------------------------------------------
def analyze_layer_stats(data: PerformanceData) -> list[dict[str, Any]]:
    """统计各层耗时、任务数、吞吐量。"""
    stats = []
    for layer in data.layers:
        executed = [t for t in layer.tasks if t.status != "SKIP"]
        durations = [t.duration_sec for t in executed] if executed else [0]
        avg_dur = sum(durations) / len(durations) if durations else 0
        total_fetched = sum(t.fetched for t in executed)
        throughput = total_fetched / layer.duration_sec if layer.duration_sec > 0 else 0

        stats.append({
            "layer": layer.layer,
            "duration_sec": layer.duration_sec,
            "pct_of_total": (layer.duration_sec / data.overall_duration_sec * 100
                             if data.overall_duration_sec > 0 else 0),
            "task_count": layer.task_count,
            "executed_count": len(executed),
            "success_count": layer.success_count,
            "fail_count": layer.fail_count,
            "skip_count": layer.skip_count,
            "avg_task_sec": round(avg_dur, 2),
            "max_task_sec": round(max(durations), 2),
            "min_task_sec": round(min(durations), 2),
            "total_fetched": total_fetched,
            "total_inserted": layer.total_inserted,
            "total_updated": layer.total_updated,
            "throughput_per_sec": round(throughput, 1),
            "status": layer.status,
        })
    return stats


def find_bottleneck_tasks(data: PerformanceData, top_n: int = 5) -> list[TaskTiming]:
    """识别耗时最长的前 N 个任务。"""
    all_tasks: list[TaskTiming] = []
    for layer in data.layers:
        all_tasks.extend(layer.tasks)
    # 按耗时降序排列，排除 SKIP 状态
    active = [t for t in all_tasks if t.status != "SKIP"]
    active.sort(key=lambda t: t.duration_sec, reverse=True)
    return active[:top_n]


def analyze_api_calls(data: PerformanceData) -> list[dict[str, Any]]:
    """分析 API 调用的响应时间和分页效率。"""
    results = []
    for layer in data.layers:
        for task in layer.tasks:
            if task.status == "SKIP":
                continue
            fetched = task.fetched
            # 根据默认 API_PAGE_SIZE=200 估算分页次数
            page_size = 200
            estimated_pages = max(1, (fetched + page_size - 1) // page_size) if fetched > 0 else 0

            # 计算 DB 处理时间（总耗时 - API 耗时）
            db_time = max(0, task.duration_sec - task.api_total_sec)

            # 每条记录的处理耗时
            per_record_ms = (task.duration_sec / fetched * 1000) if fetched > 0 else 0

            results.append({
                "task_code": task.task_code,
                "layer": task.layer,
                "fetched": fetched,
                "api_calls": task.api_calls,
                "api_total_sec": task.api_total_sec,
                "estimated_pages": estimated_pages,
                "avg_page_time_ms": (task.api_total_sec / estimated_pages * 1000
                                     if estimated_pages > 0 and task.api_total_sec > 0 else 0),
                "records_per_page": (fetched / estimated_pages
                                     if estimated_pages > 0 else 0),
                "db_time_sec": round(db_time, 2),
                "per_record_ms": round(per_record_ms, 2),
                "total_sec": task.duration_sec,
                "status": task.status,
            })
    return results


def analyze_sql_queries(
    dsn: str,
    logger: logging.Logger,
) -> list[SQLAnalysisResult]:
    """连接数据库执行 EXPLAIN ANALYZE 分析关键查询。"""
    try:
        import psycopg2  # noqa: F811
    except ImportError:
        logger.warning("psycopg2 未安装，跳过 SQL 分析")
        return []

    # 关键查询列表：ODS INSERT、DWD MERGE、DWS 汇总
    queries = [
        {
            "name": "ODS 批量 INSERT（payment_transactions）",
            "table": "ods.payment_transactions",
            "sql": """
                EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
                SELECT * FROM ods.payment_transactions
                WHERE fetched_at >= NOW() - INTERVAL '7 days'
                LIMIT 100
            """,
        },
        {
            "name": "ODS 批量 INSERT（platform_coupon_redemption_records）",
            "table": "ods.platform_coupon_redemption_records",
            "sql": """
                EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
                SELECT * FROM ods.platform_coupon_redemption_records
                WHERE fetched_at >= NOW() - INTERVAL '7 days'
                LIMIT 100
            """,
        },
        {
            "name": "ODS content_hash 去重查询",
            "table": "ods.member_balance_changes",
            "sql": """
                EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
                SELECT id, content_hash FROM ods.member_balance_changes
                WHERE fetched_at >= NOW() - INTERVAL '7 days'
            """,
        },
        {
            "name": "DWD SCD2 合并（dim_table）",
            "table": "dwd.dim_table",
            "sql": """
                EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
                SELECT * FROM dwd.dim_table
                WHERE scd2_is_current = 1
            """,
        },
        {
            "name": "DWS 订单汇总查询",
            "table": "dws.dws_order_summary",
            "sql": """
                EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
                SELECT * FROM dws.dws_order_summary
                WHERE order_date >= CURRENT_DATE - INTERVAL '30 days'
                LIMIT 100
            """,
        },
    ]

    results: list[SQLAnalysisResult] = []
    conn = None
    try:
        conn = psycopg2.connect(dsn, connect_timeout=10)
        conn.autocommit = True
        cur = conn.cursor()

        for q in queries:
            try:
                cur.execute(q["sql"])
                rows = cur.fetchall()
                plan_text = "\n".join(r[0] for r in rows)

                # 解析执行计划
                result = _parse_explain_plan(q["name"], q["table"], plan_text)
                results.append(result)
                logger.info("  ✓ %s: %.1fms", q["name"], result.actual_time_ms)
            except Exception as e:
                logger.warning("  ✗ %s: %s", q["name"], e)
                results.append(SQLAnalysisResult(
                    query_name=q["name"],
                    table_name=q["table"],
                    plan_summary=f"执行失败: {e}",
                    total_cost=0,
                    actual_time_ms=0,
                    rows_processed=0,
                    seq_scans=[],
                    missing_indexes=[],
                    recommendations=[f"查询执行失败，需检查表是否存在: {e}"],
                ))
    except Exception as e:
        logger.error("数据库连接失败: %s", e)
    finally:
        if conn:
            conn.close()

    return results


def _parse_explain_plan(
    query_name: str,
    table_name: str,
    plan_text: str,
) -> SQLAnalysisResult:
    """解析 EXPLAIN ANALYZE 输出，提取关键指标。"""
    import re

    seq_scans: list[str] = []
    missing_indexes: list[str] = []
    recommendations: list[str] = []
    total_cost = 0.0
    actual_time_ms = 0.0
    rows_processed = 0

    for line in plan_text.split("\n"):
        # 提取总耗时
        m = re.search(r"actual time=([\d.]+)\.\.([\d.]+)", line)
        if m:
            actual_time_ms = max(actual_time_ms, float(m.group(2)))

        # 提取 cost
        m = re.search(r"cost=([\d.]+)\.\.([\d.]+)", line)
        if m:
            total_cost = max(total_cost, float(m.group(2)))

        # 提取行数
        m = re.search(r"rows=(\d+)", line)
        if m:
            rows_processed = max(rows_processed, int(m.group(1)))

        # 检测 Seq Scan
        if "Seq Scan" in line:
            m_tbl = re.search(r"Seq Scan on (\S+)", line)
            tbl = m_tbl.group(1) if m_tbl else "unknown"
            seq_scans.append(tbl)

    # 基于分析结果生成建议
    if seq_scans:
        for tbl in seq_scans:
            missing_indexes.append(tbl)
            recommendations.append(f"表 {tbl} 存在全表扫描，建议添加索引")

    if actual_time_ms > 100:
        recommendations.append(f"查询耗时 {actual_time_ms:.1f}ms，考虑优化查询或添加索引")

    # 截取前 10 行作为摘要
    summary_lines = plan_text.strip().split("\n")[:10]
    plan_summary = "\n".join(summary_lines)

    return SQLAnalysisResult(
        query_name=query_name,
        table_name=table_name,
        plan_summary=plan_summary,
        total_cost=total_cost,
        actual_time_ms=actual_time_ms,
        rows_processed=rows_processed,
        seq_scans=seq_scans,
        missing_indexes=missing_indexes,
        recommendations=recommendations,
    )



def generate_recommendations(
    report: PerformanceReport,
    logger: logging.Logger,
) -> list[str]:
    """基于分析结果生成优化建议。"""
    recs: list[str] = []

    # 1. 基于瓶颈任务的建议
    for task in report.bottleneck_tasks:
        if task.duration_sec > 100:
            skip_ratio = task.skipped / task.fetched * 100 if task.fetched > 0 else 0
            if skip_ratio > 90:
                recs.append(
                    f"**{task.task_code}**（耗时 {task.duration_sec:.1f}s）："
                    f"跳过率 {skip_ratio:.0f}%，建议优化 content_hash 去重逻辑，"
                    f"在数据库端用索引加速 hash 比对，或在 API 端增加增量过滤参数减少无效拉取"
                )
            elif task.fetched > 10000:
                recs.append(
                    f"**{task.task_code}**（耗时 {task.duration_sec:.1f}s）："
                    f"拉取 {task.fetched:,} 条记录，建议增大 API_PAGE_SIZE 或启用并行分页"
                )
            else:
                recs.append(
                    f"**{task.task_code}**（耗时 {task.duration_sec:.1f}s）："
                    f"建议分析具体耗时分布（API vs DB），针对性优化"
                )

    # 2. 基于层级统计的建议
    for stat in report.layer_stats:
        if stat["pct_of_total"] > 80:
            recs.append(
                f"**{stat['layer']} 层**占总耗时 {stat['pct_of_total']:.1f}%，"
                f"是主要瓶颈层，建议优先优化该层任务"
            )
        if stat["skip_count"] > stat["task_count"] * 0.5:
            recs.append(
                f"**{stat['layer']} 层**有 {stat['skip_count']}/{stat['task_count']} "
                f"个任务被跳过，建议检查跳过条件是否合理"
            )

    # 3. 基于 API 分析的建议
    high_per_record = [a for a in report.api_analysis
                       if a["per_record_ms"] > 5 and a["fetched"] > 1000]
    if high_per_record:
        recs.append(
            "以下任务每条记录处理耗时较高（>5ms），建议优化批量写入逻辑：" +
            "、".join(f"{a['task_code']}({a['per_record_ms']:.1f}ms/条)"
                      for a in high_per_record[:5])
        )

    # 4. 基于 SQL 分析的建议
    for sql_r in report.sql_analysis:
        recs.extend(sql_r.recommendations)

    # 5. 通用建议
    if report.data.overall_duration_sec > 600:
        recs.append(
            f"全量刷新总耗时 {report.data.overall_duration_sec:.0f}s（{report.data.overall_duration_sec/60:.1f}分钟），"
            "建议考虑以下通用优化策略："
        )
        recs.append("  - ODS 层任务间无依赖，可并行执行以大幅缩短总耗时")
        recs.append("  - 对高跳过率任务，在 API 请求中增加时间过滤参数减少无效数据传输")
        recs.append("  - 对大表 INSERT，使用 COPY 协议替代逐行 INSERT 提升写入性能")
        recs.append("  - 考虑在 content_hash 列上建立索引加速去重判断")

    return recs


# ---------------------------------------------------------------------------
# 报告生成
# ---------------------------------------------------------------------------
def generate_report(report: PerformanceReport) -> str:
    """生成 Markdown 格式的性能分析报告。"""
    lines: list[str] = []
    _w = lines.append

    _w("# ETL 性能分析报告")
    _w("")
    _w(f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    _w(f"> 数据来源: 全量刷新计时数据（{report.data.flow}）")
    _w(f"> 时间窗口: {report.data.window_start} ~ {report.data.window_end}")
    _w(f"> 总耗时: {report.data.overall_duration_sec:.1f}s "
       f"({report.data.overall_duration_sec/60:.1f}分钟)")
    _w(f"> 状态: {report.data.overall_status}")
    _w("")

    # ── 目录 ──
    _w("## 目录")
    _w("")
    _w("1. [执行概览](#1-执行概览)")
    _w("2. [层级耗时统计](#2-层级耗时统计)")
    _w("3. [性能瓶颈 Top 5](#3-性能瓶颈-top-5)")
    _w("4. [任务耗时明细](#4-任务耗时明细)")
    _w("5. [API 调用分析](#5-api-调用分析)")
    _w("6. [SQL 查询分析](#6-sql-查询分析)")
    _w("7. [校验阶段分析](#7-校验阶段分析)")
    _w("8. [优化建议](#8-优化建议)")
    _w("")

    # ── 1. 执行概览 ──
    _w("## 1. 执行概览")
    _w("")
    _w("| 指标 | 值 |")
    _w("|------|-----|")
    _w(f"| Flow | `{report.data.flow}` |")
    _w(f"| 时间窗口 | {report.data.window_start} ~ {report.data.window_end} |")
    _w(f"| 总耗时 | {report.data.overall_duration_sec:.1f}s "
       f"({report.data.overall_duration_sec/60:.1f}分钟) |")
    _w(f"| 状态 | {report.data.overall_status} |")
    total_tasks = sum(s["task_count"] for s in report.layer_stats)
    total_success = sum(s["success_count"] for s in report.layer_stats)
    total_fail = sum(s["fail_count"] for s in report.layer_stats)
    total_skip = sum(s["skip_count"] for s in report.layer_stats)
    _w(f"| 总任务数 | {total_tasks} |")
    _w(f"| 成功/失败/跳过 | {total_success}/{total_fail}/{total_skip} |")
    total_fetched = sum(s["total_fetched"] for s in report.layer_stats)
    total_inserted = sum(s["total_inserted"] for s in report.layer_stats)
    _w(f"| 总拉取记录 | {total_fetched:,} |")
    _w(f"| 总写入记录 | {total_inserted:,} |")
    if report.data.verification:
        v = report.data.verification
        _w(f"| 校验耗时 | {v.duration_sec:.1f}s |")
        _w(f"| 校验表数 | {v.total_tables}（一致 {v.consistent_tables}，"
           f"补齐 {v.total_backfilled}，错误 {v.error_tables}）|")
    _w("")

    # ── 2. 层级耗时统计 ──
    _w("## 2. 层级耗时统计")
    _w("")
    _w("| 层 | 耗时(s) | 占比 | 任务数 | 执行数 | 成功 | 失败 | 跳过 | "
       "平均(s) | 最大(s) | 拉取 | 写入 | 吞吐(条/s) |")
    _w("|-----|---------|------|--------|--------|------|------|------|"
       "---------|---------|------|------|------------|")
    for s in report.layer_stats:
        _w(f"| {s['layer']} | {s['duration_sec']:.1f} | {s['pct_of_total']:.1f}% | "
           f"{s['task_count']} | {s['executed_count']} | {s['success_count']} | "
           f"{s['fail_count']} | {s['skip_count']} | {s['avg_task_sec']} | "
           f"{s['max_task_sec']} | {s['total_fetched']:,} | {s['total_inserted']:,} | "
           f"{s['throughput_per_sec']} |")
    _w("")

    # 耗时分布可视化（文本柱状图）
    _w("### 耗时分布")
    _w("")
    _w("```")
    max_dur = max(s["duration_sec"] for s in report.layer_stats) if report.layer_stats else 1
    for s in report.layer_stats:
        bar_len = int(s["duration_sec"] / max_dur * 40) if max_dur > 0 else 0
        bar = "█" * bar_len
        _w(f"  {s['layer']:>5} │{bar} {s['duration_sec']:.1f}s ({s['pct_of_total']:.1f}%)")
    _w("```")
    _w("")

    # ── 3. 性能瓶颈 Top 5 ──
    _w("## 3. 性能瓶颈 Top 5")
    _w("")
    _w("| 排名 | 任务 | 层 | 耗时(s) | 状态 | 拉取 | 写入 | 更新 | 跳过 | "
       "吞吐(条/s) | 每条耗时(ms) |")
    _w("|------|------|-----|---------|------|------|------|------|------|"
       "------------|-------------|")
    for i, t in enumerate(report.bottleneck_tasks, 1):
        per_rec = (t.duration_sec / t.fetched * 1000) if t.fetched > 0 else 0
        _w(f"| {i} | `{t.task_code}` | {t.layer} | {t.duration_sec:.1f} | "
           f"{t.status} | {t.fetched:,} | {t.inserted:,} | {t.updated:,} | "
           f"{t.skipped:,} | {t.throughput:.1f} | {per_rec:.2f} |")
    _w("")

    # 瓶颈分析
    _w("### 瓶颈分析")
    _w("")
    for i, t in enumerate(report.bottleneck_tasks, 1):
        _w(f"**{i}. {t.task_code}**（{t.duration_sec:.1f}s）")
        if t.fetched > 0:
            skip_ratio = t.skipped / t.fetched * 100
            _w(f"- 拉取 {t.fetched:,} 条，跳过 {t.skipped:,} 条（跳过率 {skip_ratio:.0f}%）")
            _w(f"- 实际写入 {t.inserted:,} 条，写入率 {t.inserted/t.fetched*100:.1f}%")
            _w(f"- 每条记录处理耗时 {t.duration_sec/t.fetched*1000:.2f}ms")
            if skip_ratio > 90:
                _w(f"- ⚠️ 跳过率极高，大量时间花在 content_hash 比对上")
        if t.error:
            _w(f"- ❌ 错误: {t.error}")
        _w("")

    # ── 4. 任务耗时明细 ──
    _w("## 4. 任务耗时明细")
    _w("")
    for layer in report.data.layers:
        _w(f"### {layer.layer} 层")
        _w("")
        _w("| 任务 | 耗时(s) | 状态 | 拉取 | 写入 | 跳过 | 错误 |")
        _w("|------|---------|------|------|------|------|------|")
        sorted_tasks = sorted(layer.tasks, key=lambda t: t.duration_sec, reverse=True)
        for t in sorted_tasks:
            _w(f"| `{t.task_code}` | {t.duration_sec:.1f} | {t.status} | "
               f"{t.fetched:,} | {t.inserted:,} | {t.skipped:,} | "
               f"{t.counts.get('errors', 0)} |")
        _w("")

    # ── 5. API 调用分析 ──
    _w("## 5. API 调用分析")
    _w("")
    # 只展示有实际数据拉取的任务
    api_with_data = [a for a in report.api_analysis if a["fetched"] > 0]
    if api_with_data:
        api_with_data.sort(key=lambda a: a["total_sec"], reverse=True)
        _w("| 任务 | 拉取 | 估算页数 | 总耗时(s) | DB耗时(s) | 每条(ms) |")
        _w("|------|------|----------|-----------|-----------|----------|")
        for a in api_with_data:
            _w(f"| `{a['task_code']}` | {a['fetched']:,} | {a['estimated_pages']} | "
               f"{a['total_sec']:.1f} | {a['db_time_sec']} | {a['per_record_ms']} |")
        _w("")

        # 分页效率分析
        _w("### 分页效率分析")
        _w("")
        total_records = sum(a["fetched"] for a in api_with_data)
        total_pages = sum(a["estimated_pages"] for a in api_with_data)
        avg_per_page = total_records / total_pages if total_pages > 0 else 0
        _w(f"- 总拉取记录: {total_records:,}")
        _w(f"- 估算总页数: {total_pages:,}")
        _w(f"- 平均每页记录数: {avg_per_page:.1f}")
        _w(f"- 当前 API_PAGE_SIZE: 200")
        _w("")
        if avg_per_page < 150:
            _w("> ⚠️ 实际每页记录数低于 PAGE_SIZE，部分端点可能返回不满页的数据")
            _w("")
    else:
        _w("本次运行中 API 调用计时数据为 0（可能未单独采集 API 耗时）。")
        _w("")
        _w("> 注意：当前计时数据中 `api_calls` 和 `api_total_sec` 均为 0，"
           "说明全量刷新脚本未单独采集 API 调用耗时。")
        _w("> 建议在后续版本中为 API 调用添加独立计时，以便区分 API 等待时间和 DB 写入时间。")
        _w("")

    # ── 6. SQL 查询分析 ──
    _w("## 6. SQL 查询分析")
    _w("")
    if report.sql_analysis:
        for sql_r in report.sql_analysis:
            _w(f"### {sql_r.query_name}")
            _w("")
            _w(f"- 表: `{sql_r.table_name}`")
            _w(f"- 实际耗时: {sql_r.actual_time_ms:.1f}ms")
            _w(f"- 预估成本: {sql_r.total_cost:.1f}")
            _w(f"- 处理行数: {sql_r.rows_processed:,}")
            if sql_r.seq_scans:
                _w(f"- ⚠️ 全表扫描: {', '.join(sql_r.seq_scans)}")
            if sql_r.missing_indexes:
                _w(f"- 🔍 建议添加索引: {', '.join(sql_r.missing_indexes)}")
            _w("")
            _w("```")
            _w(sql_r.plan_summary)
            _w("```")
            _w("")
    else:
        _w("未执行 SQL 分析（使用 `--skip-sql` 跳过或数据库连接失败）。")
        _w("")

    # ── 7. 校验阶段分析 ──
    _w("## 7. 校验阶段分析")
    _w("")
    if report.data.verification:
        v = report.data.verification
        _w(f"- 状态: {v.status}")
        _w(f"- 耗时: {v.duration_sec:.1f}s")
        _w(f"- 校验表数: {v.total_tables}")
        _w(f"- 一致表数: {v.consistent_tables}")
        _w(f"- 补齐记录: {v.total_backfilled}")
        _w(f"- 错误表数: {v.error_tables}")
        _w("")

        for layer_name, layer_v in v.layers.items():
            _w(f"### {layer_name} 层校验")
            _w("")
            _w(f"- 状态: {layer_v.get('status', 'N/A')}")
            _w(f"- 表数: {layer_v.get('total_tables', 0)}")
            _w(f"- 一致: {layer_v.get('consistent_tables', 0)}")
            _w(f"- 不一致: {layer_v.get('inconsistent_tables', 0)}")
            _w(f"- 源记录: {layer_v.get('total_source_count', 0):,}")
            _w(f"- 目标记录: {layer_v.get('total_target_count', 0):,}")
            _w(f"- 补齐: {layer_v.get('total_backfilled', 0)}")
            _w(f"- 耗时: {layer_v.get('elapsed_seconds', 0):.1f}s")
            _w("")
    else:
        _w("无校验数据。")
        _w("")

    # ── 8. 优化建议 ──
    _w("## 8. 优化建议")
    _w("")
    if report.recommendations:
        for i, rec in enumerate(report.recommendations, 1):
            _w(f"{i}. {rec}")
        _w("")
    else:
        _w("暂无优化建议。")
        _w("")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# 主流程
# ---------------------------------------------------------------------------
def find_latest_timing_file(output_dir: Path) -> Path | None:
    """查找最新的计时 JSON 文件（排除 intermediate/checkpoint 中间文件）。"""
    exclude = {"intermediate", "checkpoint"}
    json_files = sorted(
        [f for f in output_dir.glob("full_refresh_*.json")
         if not any(kw in f.name for kw in exclude)],
        reverse=True,
    )
    return json_files[0] if json_files else None


def run_analysis(
    json_path: Path,
    logger: logging.Logger,
    skip_sql: bool = False,
) -> PerformanceReport:
    """执行完整的性能分析。"""
    # 1. 加载数据
    data = load_timing_data(json_path, logger)
    logger.info("数据加载完成: %s, 总耗时 %.1fs, %d 层",
                data.flow, data.overall_duration_sec, len(data.layers))

    # 2. 层级统计
    logger.info("分析层级耗时...")
    layer_stats = analyze_layer_stats(data)
    for s in layer_stats:
        logger.info("  %s: %.1fs (%.1f%%), %d 任务",
                     s["layer"], s["duration_sec"], s["pct_of_total"], s["task_count"])

    # 3. 瓶颈识别
    logger.info("识别性能瓶颈...")
    bottlenecks = find_bottleneck_tasks(data, top_n=5)
    for i, t in enumerate(bottlenecks, 1):
        logger.info("  Top %d: %s (%.1fs, %s)", i, t.task_code, t.duration_sec, t.layer)

    # 4. API 分析
    logger.info("分析 API 调用...")
    api_analysis = analyze_api_calls(data)

    # 5. SQL 分析
    sql_analysis: list[SQLAnalysisResult] = []
    if not skip_sql:
        logger.info("分析 SQL 查询执行计划...")
        dsn = _load_dsn()
        if dsn:
            sql_analysis = analyze_sql_queries(dsn, logger)
            logger.info("SQL 分析完成: %d 个查询", len(sql_analysis))
        else:
            logger.warning("未找到数据库 DSN，跳过 SQL 分析")
    else:
        logger.info("跳过 SQL 分析（--skip-sql）")

    # 6. 构建报告
    report = PerformanceReport(
        data=data,
        bottleneck_tasks=bottlenecks,
        layer_stats=layer_stats,
        api_analysis=api_analysis,
        sql_analysis=sql_analysis,
    )

    # 7. 生成建议
    logger.info("生成优化建议...")
    report.recommendations = generate_recommendations(report, logger)

    return report


def _load_dsn() -> str | None:
    """从 .env 加载数据库 DSN。"""
    env_path = _FEIQIU_ROOT / ".env"
    if not env_path.exists():
        return None
    try:
        from dotenv import dotenv_values
        values = dotenv_values(env_path)
        return values.get("PG_DSN")
    except ImportError:
        # 手动解析
        for line in env_path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if line.startswith("PG_DSN="):
                return line.split("=", 1)[1].strip()
    return None


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="ETL 性能分析")
    parser.add_argument(
        "--input", "-i",
        help="计时 JSON 文件路径（默认自动查找最新文件）",
    )
    parser.add_argument(
        "--output", "-o",
        help="报告输出路径（默认 docs/reports/performance_report_YYYYMMDD.md）",
    )
    parser.add_argument(
        "--skip-sql",
        action="store_true",
        help="跳过 SQL EXPLAIN ANALYZE 分析",
    )
    return parser.parse_args()


def main():
    logger = _setup_logging()
    args = parse_args()

    # 确定输入文件
    if args.input:
        json_path = Path(args.input)
    else:
        json_path = find_latest_timing_file(_OUTPUT_DIR)
        if not json_path:
            logger.error("未找到计时 JSON 文件，请指定 --input 参数")
            sys.exit(1)

    if not json_path.exists():
        logger.error("文件不存在: %s", json_path)
        sys.exit(1)

    logger.info("═══ ETL 性能分析 ═══")

    # 执行分析
    report = run_analysis(json_path, logger, skip_sql=args.skip_sql)

    # 生成报告
    md_content = generate_report(report)

    # 确定输出路径
    _REPORTS_DIR.mkdir(parents=True, exist_ok=True)
    if args.output:
        output_path = Path(args.output)
    else:
        date_str = datetime.now().strftime("%Y%m%d")
        output_path = _REPORTS_DIR / f"performance_report_{date_str}.md"

    output_path.write_text(md_content, encoding="utf-8")
    logger.info("报告已保存: %s", output_path)

    # 打印摘要
    logger.info("")
    logger.info("═══ 分析摘要 ═══")
    logger.info("  总耗时: %.1fs (%.1f分钟)",
                report.data.overall_duration_sec,
                report.data.overall_duration_sec / 60)
    logger.info("  层数: %d", len(report.layer_stats))
    logger.info("  瓶颈任务: %s",
                ", ".join(f"{t.task_code}({t.duration_sec:.0f}s)"
                          for t in report.bottleneck_tasks))
    logger.info("  SQL 分析: %d 个查询", len(report.sql_analysis))
    logger.info("  优化建议: %d 条", len(report.recommendations))


if __name__ == "__main__":
    main()
