from __future__ import annotations

import argparse
import json
import math
import os
import re
import urllib.error
import urllib.parse
import urllib.request
from datetime import timedelta
from datetime import datetime
from datetime import timezone
from typing import Any


DEFAULT_BASE_URL = "https://clinic.pingcap.com"
DEFAULT_TIMEOUT_S = 60
CPU_QUERY = 'sum by (pod, instance) (rate(process_cpu_seconds_total{component="tikv"}[1m]))'
QUOTA_QUERY = "max by (pod, instance) (tikv_server_cpu_cores_quota)"
NODE_PATTERN = re.compile(r"\bdb(?:-[A-Za-z0-9]+)?-tikv-[A-Za-z0-9-]+\b")
WINDOW_START_HOUR_UTC = 3
WINDOW_DURATION_MINUTES = 30
WINDOW_COUNT = 3


class ReferenceBuildError(RuntimeError):
    pass


class ClinicClient:
    def __init__(self, *, api_key: str, base_url: str, timeout_s: int = DEFAULT_TIMEOUT_S) -> None:
        self._api_key = api_key
        self._base_url = base_url.rstrip("/")
        self._timeout_s = timeout_s

    def query_range(
        self,
        *,
        cluster_id: str,
        promql: str,
        start: int,
        end: int,
        step_seconds: int,
    ) -> dict[str, Any]:
        return self._get(
            f"/data-proxy/metrics/{cluster_id}/api/v1/query_range",
            params={
                "query": promql,
                "start": str(start),
                "end": str(end),
                "step": str(step_seconds),
            },
        )

    def _get(self, path: str, *, params: dict[str, Any]) -> dict[str, Any]:
        url = f"{self._base_url}{path}?{urllib.parse.urlencode(params, doseq=True)}"
        request = urllib.request.Request(
            url,
            headers={"Authorization": f"Bearer {self._api_key}"},
            method="GET",
        )
        try:
            with urllib.request.urlopen(request, timeout=self._timeout_s) as response:
                payload = response.read().decode("utf-8")
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")
            raise ReferenceBuildError(f"HTTP {exc.code} for {path}: {body}") from exc
        except urllib.error.URLError as exc:
            raise ReferenceBuildError(f"request failed for {path}: {exc}") from exc
        try:
            data = json.loads(payload)
        except json.JSONDecodeError as exc:
            raise ReferenceBuildError(f"invalid JSON response for {path}: {exc}") from exc
        status = str(data.get("status") or "")
        if status and status != "success":
            raise ReferenceBuildError(f"unexpected response status for {path}: {status}")
        return data


def _env_api_key() -> str:
    for key in ("SKILL_CLINIC_API_KEY", "CLINIC_API_KEY", "INVESTIBOOK_CLINIC_API_KEY"):
        value = os.environ.get(key, "").strip()
        if value:
            return value
    raise ReferenceBuildError(
        "missing Clinic API key in SKILL_CLINIC_API_KEY/CLINIC_API_KEY/INVESTIBOOK_CLINIC_API_KEY"
    )


def _parse_iso8601(value: str) -> int:
    normalized = value.strip()
    if normalized.endswith("Z"):
        normalized = normalized[:-1] + "+00:00"
    return int(datetime.fromisoformat(normalized).timestamp())


def _isoformat_utc(ts: int) -> str:
    return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat().replace("+00:00", "Z")


def _safe_float(value: Any) -> float | None:
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        return None
    if not math.isfinite(numeric):
        return None
    return numeric


def _result_series(payload: dict[str, Any]) -> list[dict[str, Any]]:
    data = payload.get("data")
    if not isinstance(data, dict):
        return []
    result = data.get("result")
    return [item for item in result if isinstance(item, dict)] if isinstance(result, list) else []


def _metric_node_name(metric: dict[str, Any]) -> str:
    for label_name in ("pod", "pod_name", "kubernetes_pod_name"):
        pod = str(metric.get(label_name) or "").strip()
        if NODE_PATTERN.search(pod):
            return pod
    for label_name in ("instance", "job", "exported_instance", "pod", "pod_name", "kubernetes_pod_name"):
        raw_value = str(metric.get(label_name) or "").strip()
        match = NODE_PATTERN.search(raw_value)
        if match:
            return match.group(0)
    return ""


def _series_numeric_values(item: dict[str, Any]) -> list[float]:
    values = item.get("values")
    if not isinstance(values, list):
        return []
    observed: list[float] = []
    for row in values:
        if not isinstance(row, list) or len(row) < 2:
            continue
        numeric = _safe_float(row[1])
        if numeric is not None:
            observed.append(numeric)
    return observed


def _mean(values: list[float]) -> float | None:
    if not values:
        return None
    return sum(values) / len(values)


def _max_value(values: list[float]) -> float | None:
    if not values:
        return None
    return max(values)


def _median(values: list[float]) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    return ordered[len(ordered) // 2]


def _resolved_windows(now: datetime | None = None) -> list[tuple[str, str]]:
    now_utc = now or datetime.now(tz=timezone.utc)
    candidate_start = now_utc.replace(
        hour=WINDOW_START_HOUR_UTC,
        minute=0,
        second=0,
        microsecond=0,
    )
    candidate_end = candidate_start + timedelta(minutes=WINDOW_DURATION_MINUTES)
    if now_utc < candidate_end:
        candidate_start -= timedelta(days=1)

    windows: list[tuple[str, str]] = []
    current_start = candidate_start
    for _ in range(WINDOW_COUNT):
        current_end = current_start + timedelta(minutes=WINDOW_DURATION_MINUTES)
        windows.append(
            (
                current_start.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
                current_end.astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
            )
        )
        current_start -= timedelta(days=1)
    return windows


def _select_target_node(*, client: ClinicClient, cluster_id: str, start_iso: str, end_iso: str) -> str:
    payload = client.query_range(
        cluster_id=cluster_id,
        promql=CPU_QUERY,
        start=_parse_iso8601(start_iso),
        end=_parse_iso8601(end_iso),
        step_seconds=60,
    )
    candidates: list[tuple[float, str]] = []
    for item in _result_series(payload):
        metric = item.get("metric")
        if not isinstance(metric, dict):
            continue
        node = _metric_node_name(metric)
        avg_cpu = _mean(_series_numeric_values(item))
        if not node or avg_cpu is None:
            continue
        candidates.append((avg_cpu, node))
    if not candidates:
        raise ReferenceBuildError("failed to resolve a target TiKV node from CPU evidence")
    candidates.sort(key=lambda item: (-item[0], item[1]))
    return candidates[0][1]


def _build_case_input(args: argparse.Namespace) -> dict[str, Any]:
    windows = _resolved_windows()
    metrics_cluster_id = args.metrics_cluster_id or args.cluster_id
    client = ClinicClient(api_key=_env_api_key(), base_url=os.environ.get("CLINIC_ENDPOINT", DEFAULT_BASE_URL))
    return {
        "start": windows[-1][0],
        "end": windows[0][1],
        "target_node": _select_target_node(
            client=client,
            cluster_id=metrics_cluster_id,
            start_iso=windows[0][0],
            end_iso=windows[0][1],
        ),
        "windows_markdown": "\n".join(f"- {start} 到 {end}" for start, end in windows),
    }


def _build_reference(*, cluster_id: str, node: str, metrics_cluster_id: str | None = None) -> dict[str, Any]:
    client = ClinicClient(api_key=_env_api_key(), base_url=os.environ.get("CLINIC_ENDPOINT", DEFAULT_BASE_URL))
    scope_cluster_id = metrics_cluster_id or cluster_id

    quota_candidates: list[float] = []
    windows_output: list[dict[str, Any]] = []

    for start_iso, end_iso in _resolved_windows():
        start_ts = _parse_iso8601(start_iso)
        end_ts = _parse_iso8601(end_iso)
        cpu_payload = client.query_range(
            cluster_id=scope_cluster_id,
            promql=CPU_QUERY,
            start=start_ts,
            end=end_ts,
            step_seconds=60,
        )
        quota_payload = client.query_range(
            cluster_id=scope_cluster_id,
            promql=QUOTA_QUERY,
            start=start_ts,
            end=end_ts,
            step_seconds=60,
        )

        node_cpu_values: list[float] = []
        for item in _result_series(cpu_payload):
            metric = item.get("metric")
            if not isinstance(metric, dict):
                continue
            if _metric_node_name(metric) != node:
                continue
            node_cpu_values.extend(_series_numeric_values(item))

        node_quota_values: list[float] = []
        for item in _result_series(quota_payload):
            metric = item.get("metric")
            if not isinstance(metric, dict):
                continue
            if _metric_node_name(metric) != node:
                continue
            node_quota_values.extend(_series_numeric_values(item))

        avg_cpu = _mean(node_cpu_values)
        peak_cpu = _max_value(node_cpu_values)
        quota = _median(node_quota_values)
        if avg_cpu is None or peak_cpu is None or quota is None or quota <= 0:
            return {
                "cluster_id": cluster_id,
                "node": node,
                "cpu_spec_cores": round(quota, 4) if quota is not None and quota > 0 else None,
                "evidence_sufficient": False,
                "windows": [],
            }

        quota_candidates.extend(node_quota_values)
        windows_output.append(
            {
                "start": start_iso,
                "end": end_iso,
                "avg_cpu_cores": round(avg_cpu, 4),
                "peak_cpu_cores": round(peak_cpu, 4),
            }
        )

    cpu_spec = _median(quota_candidates)
    if cpu_spec is None or cpu_spec <= 0:
        return {
            "cluster_id": cluster_id,
            "node": node,
            "cpu_spec_cores": None,
            "evidence_sufficient": False,
            "windows": [],
        }

    return {
        "cluster_id": cluster_id,
        "node": node,
        "cpu_spec_cores": round(cpu_spec, 4),
        "evidence_sufficient": True,
        "windows": windows_output,
    }


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Runtime hooks for single-node CPU core scope comparison case.")
    subparsers = parser.add_subparsers(dest="command", required=True)

    resolve_parser = subparsers.add_parser("resolve-windows")
    resolve_parser.add_argument("--cluster-id", required=True)
    resolve_parser.add_argument("--metrics-cluster-id")

    build_reference = subparsers.add_parser("build-reference")
    build_reference.add_argument("--cluster-id", required=True)
    build_reference.add_argument("--metrics-cluster-id")
    build_reference.add_argument("--node", required=True)
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()
    args = parser.parse_args(argv)

    if args.command == "resolve-windows":
        payload = _build_case_input(args)
        json.dump(payload, fp=os.sys.stdout, ensure_ascii=False, indent=2)
        os.sys.stdout.write("\n")
        return 0

    if args.command == "build-reference":
        payload = _build_reference(
            cluster_id=args.cluster_id,
            node=args.node,
            metrics_cluster_id=args.metrics_cluster_id,
        )
        json.dump(payload, fp=os.sys.stdout, ensure_ascii=False, indent=2)
        os.sys.stdout.write("\n")
        return 0

    parser.error(f"unsupported command: {args.command}")
    return 2


if __name__ == "__main__":
    raise SystemExit(main())