from __future__ import annotations

import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen


DEFAULT_CLINIC_BASE = "https://clinic.pingcap.com/clinic/api/v1"
REPORT_SOURCE_FORMAT = "clinic-health-check-api-extract/v1"
REPORT_STATUS_RANK = {"partial": 1, "success": 2}


def text(value: Any) -> str:
    return "" if value is None else str(value)


def links(values: Iterable[tuple[str, Any]]) -> list[dict[str, str]]:
    result: list[dict[str, str]] = []
    seen: set[str] = set()
    for label, raw_url in values:
        url = text(raw_url).strip()
        if not url or url in seen:
            continue
        seen.add(url)
        result.append({"text": label, "href": url})
    return result


def mapped_links(value: Any) -> list[dict[str, str]]:
    if not isinstance(value, dict):
        return []
    return links((key.replace("_", " ").title(), url) for key, url in value.items())


def cell(value: Any, cell_links: Iterable[dict[str, str]] = ()) -> dict[str, Any]:
    return {"text": text(value), "links": list(cell_links)}


def item_content(detail: dict[str, Any], name: str) -> dict[str, Any]:
    for item in detail.get("items") or []:
        if item.get("name") == name and isinstance(item.get("content"), dict):
            return item["content"]
    return {}


def section_prefix(content: dict[str, Any]) -> str:
    summary = content.get("summary")
    summary_text = (
        json.dumps(summary, ensure_ascii=False, sort_keys=True)
        if isinstance(summary, (dict, list))
        else text(summary)
    )
    return "\n\n".join(
        value
        for value in (
            text(content.get("name")),
            text(content.get("status")),
            text(content.get("description")),
            summary_text,
        )
        if value
    )


def extreme_slow_query_section(detail: dict[str, Any]) -> dict[str, Any]:
    content = item_content(detail, "slow_query_hugeResourceUsage")
    headers = [
        "Digest",
        "Sample Query",
        "Hit Records",
        "Avg Query Time (s)",
        "Max Query Time (s)",
        "Hit Time Records",
        "Avg Max Mem",
        "Max Max Mem",
        "Hit Mem Records",
        "Samples",
    ]
    rows: list[list[dict[str, Any]]] = []
    for result in content.get("results") or []:
        detail_links = links(
            ("Sample", sample.get("link"))
            for sample in result.get("digest_detail") or []
            if isinstance(sample, dict)
        )
        rows.append(
            [
                cell(result.get("digest"), detail_links),
                cell(result.get("sample_query")),
                cell(result.get("hit_records")),
                cell(result.get("avg_query_time_s")),
                cell(result.get("max_query_time_s")),
                cell(result.get("hit_query_time_records")),
                cell(result.get("avg_max_mem_b")),
                cell(result.get("max_max_mem_b")),
                cell(result.get("hit_mem_records")),
                cell("View Samples", detail_links),
            ]
        )
    return {"prefix": section_prefix(content), "table": {"headers": headers, "rows": rows}}


def table_full_scan_section(detail: dict[str, Any]) -> dict[str, Any]:
    content = item_content(detail, "tablefullscan_slow_query")
    headers = [
        "Digest",
        "Sample Query",
        "Total Scan Rows",
        "Slow Count",
        "Max Scanned Rows",
        "P50 Scanned Rows",
        "P95 Scanned Rows",
        "P50 Latency (s)",
        "P95 Latency (s)",
        "Total Latency (s)",
        "Actions",
    ]
    rows: list[list[dict[str, Any]]] = []
    for result in content.get("results") or []:
        sample_links = links(
            ("Sample", sample.get("clinic_detail_url"))
            for sample in result.get("samples") or []
            if isinstance(sample, dict)
        )
        rows.append(
            [
                cell(result.get("digest"), sample_links),
                cell(result.get("query")),
                cell(result.get("total_scanned_rows")),
                cell(result.get("slow_count")),
                cell(result.get("max_scanned_rows")),
                cell(result.get("p50_scanned_rows")),
                cell(result.get("p95_scanned_rows")),
                cell(result.get("p50_latency_s")),
                cell(result.get("p95_latency_s")),
                cell(result.get("total_latency_s")),
                cell("View Samples", sample_links),
            ]
        )
    return {"prefix": section_prefix(content), "table": {"headers": headers, "rows": rows}}


def top_digest_section(detail: dict[str, Any]) -> dict[str, Any]:
    content = item_content(detail, "slowlog_top_digest")
    dimensions = {
        "Count": "top_by_count",
        "Total Latency": "top_by_total_latency",
        "SUM Total Keys": "top_by_total_keys",
        "SUM Process Keys": "top_by_process_keys",
        "SUM Request Count": "top_by_request_count",
    }
    headers = [
        "Rank",
        "Digest",
        "Sample Query",
        "Slow Count",
        "Total Latency (s)",
        "Total Keys",
        "Process Keys",
        "Request Count",
        "SUM RU",
        "Max Latency (s)",
    ]
    raw_results = content.get("results") or {}
    tabs: dict[str, dict[str, Any]] = {}
    for label, result_key in dimensions.items():
        rows: list[list[dict[str, Any]]] = []
        for result in raw_results.get(result_key) or []:
            rows.append(
                [
                    cell(result.get("rank")),
                    cell(result.get("digest")),
                    cell(result.get("query")),
                    cell(result.get("slow_count")),
                    cell(result.get("total_latency_s")),
                    cell(result.get("total_keys")),
                    cell(result.get("process_keys")),
                    cell(result.get("request_count")),
                    cell(result.get("sum_ru")),
                    cell(result.get("max_latency_s")),
                ]
            )
        tabs[label] = {"headers": headers, "rows": rows}
    return {"prefix": section_prefix(content), "tabs": tabs}


def utc_time(epoch: Any) -> str:
    if not isinstance(epoch, (int, float)):
        return ""
    return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat().replace("+00:00", "Z")


def hotspot_section(detail: dict[str, Any], item_name: str) -> dict[str, Any]:
    content = item_content(detail, item_name)
    is_cpu = item_name == "sql_tikv_high_cpu"
    max_header = "Max CPU" if is_cpu else "Max Duration"
    summary_headers = [
        "SQL Digest",
        "Affected",
        "Max Severity",
        "TopSQL Share",
        max_header,
        "Slow Log",
    ]
    summary_rows: list[list[dict[str, Any]]] = []
    for summary in content.get("digest_summary") or []:
        representative_links = mapped_links(summary.get("representative_links"))
        max_value = (
            summary.get("max_cpu_percent")
            if is_cpu
            else summary.get("max_duration_ms")
        )
        summary_rows.append(
            [
                cell(summary.get("digest"), representative_links),
                cell(
                    f"{summary.get('affected_windows', 0)} windows\n"
                    f"{summary.get('affected_instances', 0)} instances"
                ),
                cell(summary.get("max_severity")),
                cell(
                    f"avg {summary.get('avg_topsql_share_percent', 0)}%\n"
                    f"max {summary.get('max_topsql_share_percent', 0)}%"
                ),
                cell(max_value),
                cell(f"Slow Logs: {summary.get('slow_log_count', 0)}", representative_links),
            ]
        )

    window_headers = [
        "Severity",
        "TiKV Instance",
        "Window",
        "CPU Usage" if is_cpu else "Duration",
        "Top SQL",
        "Slow Log",
    ]
    window_rows: list[list[dict[str, Any]]] = []
    for result in content.get("results") or []:
        anomaly = result.get("node_anomaly") or {}
        candidates = [
            candidate
            for candidate in result.get("sql_candidates") or []
            if isinstance(candidate, dict)
        ]
        candidate_links = [
            link
            for candidate in candidates
            for link in mapped_links(candidate.get("links"))
        ]
        metric_links = mapped_links(anomaly.get("links"))
        if is_cpu:
            signal = (
                f"max {anomaly.get('max_cpu_percent', 0)}%\n"
                f"avg {anomaly.get('avg_cpu_percent', 0)}%"
            )
        else:
            signal = text(
                anomaly.get("max_duration_ms")
                or anomaly.get("anomaly_duration_seconds")
            )
        digests = "\n".join(text(candidate.get("digest")) for candidate in candidates)
        slow_count = sum(
            int((candidate.get("slow_log") or {}).get("count") or 0)
            for candidate in candidates
        )
        window_rows.append(
            [
                cell(result.get("node_severity")),
                cell(anomaly.get("instance") or anomaly.get("source_instance"), metric_links),
                cell(
                    f"{utc_time(anomaly.get('window_start'))} ~\n"
                    f"{utc_time(anomaly.get('window_end'))}"
                ),
                cell(signal),
                cell(digests, candidate_links),
                cell(f"Slow Logs: {slow_count}", candidate_links),
            ]
        )

    return {
        "prefix": section_prefix(content),
        "tables": [
            {"kind": "digest_summary", "headers": summary_headers, "rows": summary_rows},
            {"kind": "windows", "headers": window_headers, "rows": window_rows},
        ],
    }


def build_sections(detail: dict[str, Any]) -> dict[str, Any]:
    return {
        "extreme_slow_query": extreme_slow_query_section(detail),
        "table_full_scan": table_full_scan_section(detail),
        "slow_log_top_digest": top_digest_section(detail),
        "sql_tikv_high_duration": hotspot_section(detail, "sql_tikv_high_duration"),
        "sql_tikv_high_cpu": hotspot_section(detail, "sql_tikv_high_cpu"),
    }


def request_json(url: str, api_key: str) -> dict[str, Any]:
    request = Request(
        url,
        headers={"Accept": "application/json", "Authorization": f"Bearer {api_key}"},
    )
    try:
        with urlopen(request, timeout=180) as response:
            return json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"Clinic returned {exc.code} for {url}: {body}") from exc
    except URLError as exc:
        raise RuntimeError(f"cannot reach Clinic at {url}: {exc}") from exc


def list_results(base_url: str, org_id: str, cluster_id: str, api_key: str) -> list[dict[str, Any]]:
    page = 1
    limit = 100
    results: list[dict[str, Any]] = []
    while True:
        query = urlencode({"page": page, "limit": limit})
        payload = request_json(
            f"{base_url.rstrip('/')}/compute-engine/{org_id}/{cluster_id}"
            f"/healthcheck/result?{query}",
            api_key,
        )
        page_results = payload.get("results") or []
        results.extend(page_results)
        if len(results) >= int(payload.get("total") or 0) or len(page_results) < limit:
            return results
        page += 1


def select_latest_daily_results(
    results: list[dict[str, Any]], latest_days: int
) -> list[dict[str, Any]]:
    """Choose one strongest report for each of the latest distinct observation days."""
    by_day: dict[str, dict[str, Any]] = {}
    for result in results:
        status = text(result.get("status")).lower()
        observation_day = text(result.get("timeRangeAStart"))[:10]
        if status not in REPORT_STATUS_RANK or len(observation_day) != 10:
            continue
        current = by_day.get(observation_day)
        score = (
            REPORT_STATUS_RANK[status],
            text(result.get("createdAt")),
            text(result.get("uuid")),
        )
        current_score = (
            REPORT_STATUS_RANK.get(text(current.get("status")).lower(), 0),
            text(current.get("createdAt")),
            text(current.get("uuid")),
        ) if current else None
        if current_score is None or score > current_score:
            by_day[observation_day] = result

    selected_days = sorted(by_day, reverse=True)[:latest_days]
    return [by_day[day] for day in sorted(selected_days)]


def report_url(org_id: str, cluster_id: str, result: dict[str, Any]) -> str:
    query = urlencode(
        {
            "from": result["timeRangeAStart"],
            "to": result["timeRangeAEnd"],
        }
    )
    return (
        f"https://clinic.pingcap.com/portal/#/orgs/{org_id}/clusters/{cluster_id}"
        f"/health-check/{result['uuid']}?{query}"
    )


def extract_reports(
    base_url: str,
    org_id: str,
    cluster_id: str,
    api_key: str,
    after_created_at: str,
    latest_days: int | None = None,
) -> list[dict[str, Any]]:
    results = list_results(base_url, org_id, cluster_id, api_key)
    if latest_days is not None:
        candidates = select_latest_daily_results(results, latest_days)
    else:
        candidates = [
            result
            for result in results
            if text(result.get("status")).lower() in REPORT_STATUS_RANK
            and text(result.get("createdAt")) > after_created_at
        ]
    reports: list[dict[str, Any]] = []
    for result in sorted(candidates, key=lambda value: value["createdAt"]):
        result_id = result["uuid"]
        detail = request_json(
            f"{base_url.rstrip('/')}/compute-engine/{org_id}/{cluster_id}"
            f"/healthcheck/result/{result_id}",
            api_key,
        )
        if detail.get("clusterId") != cluster_id:
            raise ValueError(f"report {result_id} belongs to a different cluster")
        start = detail.get("timeRangeAStart") or result.get("timeRangeAStart")
        end = detail.get("timeRangeAEnd") or result.get("timeRangeAEnd")
        created_at = detail.get("createdAt") or result.get("createdAt")
        normalized = {
            "id": result_id,
            "date": text(start)[:10],
            "from": start,
            "to": end,
            "created_at": created_at,
            "status": detail.get("status") or result.get("status"),
            "source_url": report_url(org_id, cluster_id, result),
            "sections": build_sections(detail),
        }
        reports.append(normalized)
    return reports


def latest_created_at(reference: dict[str, Any]) -> str:
    return max(
        (text(report.get("created_at") or report.get("to")) for report in reference.get("reports") or []),
        default="",
    )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Fetch new Clinic Health Check reports as a lean SQL signal extract."
    )
    parser.add_argument("--org-id", required=True)
    parser.add_argument("--cluster-id", required=True)
    parser.add_argument("--provenance-from", type=Path)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--after-created-at")
    parser.add_argument("--latest-days", type=int)
    parser.add_argument("--dbaas-org-id")
    parser.add_argument("--org-name")
    parser.add_argument("--cluster-name")
    parser.add_argument("--project-id")
    parser.add_argument("--provider")
    parser.add_argument("--region")
    parser.add_argument("--clinic-base", default=DEFAULT_CLINIC_BASE)
    args = parser.parse_args()

    api_key = os.environ.get("CLINIC_API_KEY", "").strip()
    if not api_key:
        raise ValueError("CLINIC_API_KEY is required")

    if args.latest_days is not None and not 1 <= args.latest_days <= 366:
        raise ValueError("--latest-days must be between 1 and 366")
    if args.latest_days is not None and args.after_created_at:
        raise ValueError("--latest-days and --after-created-at cannot be used together")

    reference = (
        json.loads(args.provenance_from.read_text(encoding="utf-8"))
        if args.provenance_from
        else {"provenance": {}, "reports": []}
    )
    after_created_at = args.after_created_at or latest_created_at(reference)
    reports = extract_reports(
        args.clinic_base,
        args.org_id,
        args.cluster_id,
        api_key,
        after_created_at,
        args.latest_days,
    )
    if not reports:
        print(json.dumps({"reports": 0, "after_created_at": after_created_at}, indent=2))
        return

    provenance = dict(reference.get("provenance") or {})
    supplied_metadata = {
        "dbaas_org_id": args.dbaas_org_id,
        "org_name": args.org_name,
        "cluster_name": args.cluster_name,
        "project_id": args.project_id,
        "provider": args.provider,
        "region": args.region,
    }
    provenance.update(
        {key: value for key, value in supplied_metadata.items() if value is not None}
    )
    provenance.update(
        {
            "source": "Clinic Health Check combined report API",
            "source_format": REPORT_SOURCE_FORMAT,
            "portal_org_id": args.org_id,
            "cluster_id": args.cluster_id,
            "report_count": len(reports),
            "date_start": reports[0]["date"],
            "date_end": reports[-1]["date"],
            "report_created_at_start": reports[0]["created_at"],
            "report_created_at_end": reports[-1]["created_at"],
            "after_created_at": after_created_at,
            "selection_mode": "latest_distinct_observation_days" if args.latest_days else "created_after",
            "latest_days": args.latest_days,
            "note": (
                "Fetched from authenticated Clinic combined report APIs. "
                "Only SQL check summaries and evidence links are retained; "
                "no synthetic rows are included. When latest_days is set, one "
                "highest-quality report is retained per distinct observation day."
            ),
        }
    )
    payload = {"provenance": provenance, "reports": reports}
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print(
        json.dumps(
            {
                "output": str(args.output),
                "reports": len(reports),
                "created_at_start": reports[0]["created_at"],
                "created_at_end": reports[-1]["created_at"],
                "report_ids": [report["id"] for report in reports],
            },
            ensure_ascii=False,
            indent=2,
        )
    )


if __name__ == "__main__":
    try:
        main()
    except (RuntimeError, ValueError, KeyError) as exc:
        print(str(exc), file=sys.stderr)
        raise SystemExit(1) from exc