#!/usr/bin/env python3
"""Clinic API Python helper — zero external dependencies (stdlib only).

Usage:
    from clinic_api import ClinicAPI
    api = ClinicAPI()                       # reads .env automatically
    api.query_range("CLUSTER_ID", "sum(rate(tidb_server_query_total[1m]))", start, end)
    api.query_sql("CLUSTER_ID", "SELECT ... FROM clinic_data_proxy.slow_query_logs WHERE date='20260405'")
"""
from __future__ import annotations

import json
import hashlib
import os
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

_ENV_ENDPOINTS = {
    "dev": "https://dev-clinic.pingcap.com",
    "staging": "https://staging-clinic.pingcap.com",
    "prod": "https://clinic.pingcap.com",
}
_DEFAULT_HTTP_TIMEOUT_SECONDS = 60
_DEFAULT_HTTP_ATTEMPTS = 2
_RETRYABLE_HTTP_STATUS = {502, 503, 504}
_PERF_LOCK = threading.Lock()


def _short_hash(value: str) -> str:
    return hashlib.sha1(value.encode(errors="replace")).hexdigest()[:12]


def _perf_event(event: dict) -> None:
    path = os.environ.get("CLINIC_PERF_LOG", "").strip()
    if not path:
        return
    event.setdefault("ts", time.time())
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with _PERF_LOCK:
            with open(path, "a", encoding="utf-8") as f:
                f.write(json.dumps(event, ensure_ascii=False, sort_keys=True) + "\n")
                f.flush()
    except Exception:
        pass


def _request_meta(req: urllib.request.Request) -> dict:
    parsed = urllib.parse.urlparse(req.full_url)
    params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
    flat_params = {k: v[-1] for k, v in params.items() if v}
    query_text = str(flat_params.get("query") or flat_params.get("sql") or "")
    body_preview = ""
    data = getattr(req, "data", None)
    if data:
        body_text = data.decode(errors="replace")
        body_preview = body_text[:240]
        try:
            body = json.loads(body_text)
            sql = body.get("sql") or body.get("query") or ""
            if sql:
                query_text = str(sql)
        except Exception:
            pass
    return {
        "method": req.get_method(),
        "path": parsed.path,
        "params": {
            key: flat_params[key]
            for key in ("start", "end", "step", "time", "limit", "direction", "timeout")
            if key in flat_params
        },
        "query_hash": _short_hash(query_text) if query_text else "",
        "query_preview": query_text[:240],
        "body_preview": body_preview,
    }


def _load_dotenv(path: str = ".env") -> None:
    """Minimal .env loader (KEY=VALUE, no quotes handling needed)."""
    p = Path(path)
    if not p.exists():
        return
    for line in p.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            continue
        k, _, v = line.partition("=")
        k, v = k.strip(), v.strip().strip("'\"")
        if k and k not in os.environ:
            os.environ[k] = v


class ClinicAPI:
    """Thin wrapper around Clinic Data Proxy HTTP APIs."""

    def __init__(self, env_file: str = ".env"):
        _load_dotenv(env_file)
        clinic_env = os.environ.get("CLINIC_ENV", "prod")
        self.endpoint = os.environ.get(
            "CLINIC_ENDPOINT", _ENV_ENDPOINTS.get(clinic_env, _ENV_ENDPOINTS["prod"])
        )
        self.api_base = f"{self.endpoint}/clinic/api/v1"
        self.api_key = os.environ.get("CLINIC_API_KEY", "")
        if not self.api_key:
            raise RuntimeError("CLINIC_API_KEY not set. Check .env or environment.")
        self._canonical_org_ids: dict[str, str] = {}

    # ── helpers ──────────────────────────────────────────────

    def _get(
        self,
        path: str,
        params: dict | None = None,
        extra_headers: dict | None = None,
    ) -> dict:
        qs = urllib.parse.urlencode(params or {}, doseq=True)
        url = f"{self.endpoint}{path}" + (f"?{qs}" if qs else "")
        headers = {"Authorization": f"Bearer {self.api_key}"}
        if extra_headers:
            headers.update(extra_headers)
        req = urllib.request.Request(url, headers=headers)
        return self._do(req)

    def _api_get(self, path: str, params: dict | None = None) -> dict:
        qs = urllib.parse.urlencode(params or {}, doseq=True)
        url = f"{self.api_base}{path}" + (f"?{qs}" if qs else "")
        req = urllib.request.Request(url, headers={"Authorization": f"Bearer {self.api_key}"})
        return self._do(req)

    def _post_json(self, path: str, body: object) -> dict | list:
        url = f"{self.endpoint}{path}"
        data = json.dumps(body).encode()
        req = urllib.request.Request(
            url,
            data=data,
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json",
            },
        )
        return self._do(req)

    @staticmethod
    def _path_segment(value: object) -> str:
        return urllib.parse.quote(str(value), safe="")

    def get_cluster_by_id(self, cluster_id: str) -> dict:
        """Return authoritative metadata for an exact logical or physical cluster ID."""
        result = self._get(
            f"/data-proxy/api/v1/clusters/{self._path_segment(cluster_id)}"
        )
        if isinstance(result, dict) and not result.get("error"):
            metadata = result.get("data", result)
            if isinstance(metadata, dict) and metadata.get("orgId"):
                self._canonical_org_ids[str(cluster_id)] = str(metadata["orgId"])
        return result

    @staticmethod
    def _deploy_type(metadata: dict) -> str:
        return str(
            metadata.get("deployType")
            or metadata.get("clusterDeployTypeV2")
            or metadata.get("clusterDeployType")
            or ""
        ).lower().replace("_", "-").replace(" ", "-")

    @classmethod
    def is_byoc_metadata(cls, metadata: dict) -> bool:
        return cls._deploy_type(metadata) in {"byoc", "byoc-host", "byoc-resource-pool"}

    def byoc_physical_cluster_id(self, cluster_id: str, metadata: dict | None = None) -> str:
        """Resolve a logical BYOC cluster to its physical ``byoc-host`` ID."""
        resolved = metadata if isinstance(metadata, dict) else self.get_cluster_by_id(cluster_id)
        if not isinstance(resolved, dict) or resolved.get("error"):
            raise RuntimeError(f"Could not resolve metadata for cluster {cluster_id}")
        if isinstance(resolved.get("data"), dict):
            resolved = resolved["data"]
        org_id = resolved.get("orgId")
        if org_id:
            self._canonical_org_ids[str(cluster_id)] = str(org_id)
        deploy_type = self._deploy_type(resolved)
        if deploy_type == "byoc-host":
            return str(cluster_id)
        if deploy_type != "byoc":
            raise RuntimeError(f"Cluster {cluster_id} is not a logical or physical BYOC cluster")
        context = resolved.get("diagnosticContext") or {}
        context_parent = context.get("parentId") if isinstance(context, dict) else None
        physical_id = str(context_parent or resolved.get("parentId") or "")
        if not physical_id:
            raise RuntimeError(
                f"Could not resolve physical parent ID for logical BYOC cluster {cluster_id}"
            )
        return physical_id

    def byoc_data_plane_namespace(self, cluster_id: str) -> str:
        """Return the physical Kubernetes namespace for BYOC infrastructure metrics."""
        metadata = self.get_cluster_by_id(cluster_id)
        if isinstance(metadata, dict) and isinstance(metadata.get("data"), dict):
            metadata = metadata["data"]
        if not isinstance(metadata, dict) or not metadata.get("orgId"):
            raise RuntimeError(f"Could not resolve metadata for cluster {cluster_id}")
        self._canonical_org_ids[str(cluster_id)] = str(metadata["orgId"])
        if self._deploy_type(metadata) == "byoc":
            return f"tidb{self.byoc_physical_cluster_id(cluster_id, metadata)}"
        return f"tidb{cluster_id}"

    def _canonical_org_id(self, cluster_id: str) -> str:
        cluster_id = str(cluster_id)
        cached = self._canonical_org_ids.get(cluster_id)
        if cached:
            return cached
        metadata = self.get_cluster_by_id(cluster_id)
        if isinstance(metadata, dict) and isinstance(metadata.get("data"), dict):
            metadata = metadata["data"]
        org_id = metadata.get("orgId") if isinstance(metadata, dict) else None
        if not org_id:
            raise RuntimeError(f"Could not resolve DBAAS org ID for cluster {cluster_id}")
        self._canonical_org_ids[cluster_id] = str(org_id)
        return str(org_id)

    def _canonical_cluster_path(self, cluster_id: str, suffix: str = "") -> str:
        org_id = self._canonical_org_id(cluster_id)
        path = (
            f"/data-proxy/api/v1/orgs/{self._path_segment(org_id)}/clusters/"
            f"{self._path_segment(cluster_id)}"
        )
        return f"{path}/{suffix.lstrip('/')}" if suffix else path

    @staticmethod
    def _positive_numeric_id(value: object, *, name: str) -> str:
        text = str(value).strip()
        if not text.isdigit() or int(text) <= 0:
            raise ValueError(f"{name} must be a positive unsigned integer")
        return text

    def ticdc_changefeeds_list(self, cluster_id: str, *, org_id: str = "") -> dict:
        """List TiCDC changefeeds through the canonical Clinic Data Proxy."""
        cluster_id = self._positive_numeric_id(cluster_id, name="cluster_id")
        canonical_org_id = self._positive_numeric_id(
            org_id or self._canonical_org_id(cluster_id),
            name="org_id",
        )
        return self._get(
            f"/data-proxy/api/v1/orgs/{self._path_segment(canonical_org_id)}/clusters/"
            f"{self._path_segment(cluster_id)}/ticdc/changefeeds"
        )

    def ticdc_changefeed_get(
        self,
        cluster_id: str,
        changefeed_id: str,
        *,
        org_id: str = "",
    ) -> dict:
        """Get one TiCDC changefeed by its kernel/internal ID."""
        cluster_id = self._positive_numeric_id(cluster_id, name="cluster_id")
        changefeed_id = self._positive_numeric_id(
            changefeed_id,
            name="kernel_changefeed_id",
        )
        canonical_org_id = self._positive_numeric_id(
            org_id or self._canonical_org_id(cluster_id),
            name="org_id",
        )
        return self._get(
            f"/data-proxy/api/v1/orgs/{self._path_segment(canonical_org_id)}/clusters/"
            f"{self._path_segment(cluster_id)}/ticdc/changefeeds/"
            f"{self._path_segment(changefeed_id)}"
        )

    @staticmethod
    def _do(req: urllib.request.Request) -> dict:
        last_error = ""
        meta = _request_meta(req)
        for attempt in range(1, _DEFAULT_HTTP_ATTEMPTS + 1):
            request_id = f"{int(time.time() * 1000)}-{threading.get_ident()}-{attempt}"
            _perf_event({
                "event": "http_start",
                "request_id": request_id,
                "attempt": attempt,
                "timeout_s": _DEFAULT_HTTP_TIMEOUT_SECONDS,
                **meta,
            })
            t0 = time.monotonic()
            try:
                with urllib.request.urlopen(req, timeout=_DEFAULT_HTTP_TIMEOUT_SECONDS) as resp:
                    raw = resp.read()
                    _perf_event({
                        "event": "http_end",
                        "request_id": request_id,
                        "attempt": attempt,
                        "elapsed_ms": round((time.monotonic() - t0) * 1000, 3),
                        "status": "ok",
                        "status_code": getattr(resp, "status", None),
                        "bytes": len(raw),
                        **meta,
                    })
                    return json.loads(raw)
            except urllib.error.HTTPError as e:
                body = e.read().decode(errors="replace")
                trace_id = e.headers.get("X-Trace-Id") or e.headers.get("X-Request-Id")
                suffix = f" trace_id={trace_id}" if trace_id else ""
                last_error = f"HTTP {e.code}{suffix}: {body[:500]}"
                _perf_event({
                    "event": "http_end",
                    "request_id": request_id,
                    "attempt": attempt,
                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 3),
                    "status": "http_error",
                    "status_code": e.code,
                    "error": last_error[:500],
                    **meta,
                })
                if e.code not in _RETRYABLE_HTTP_STATUS or attempt >= _DEFAULT_HTTP_ATTEMPTS:
                    return {"error": last_error}
            except (TimeoutError, urllib.error.URLError) as e:
                last_error = str(e)
                _perf_event({
                    "event": "http_end",
                    "request_id": request_id,
                    "attempt": attempt,
                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 3),
                    "status": "exception",
                    "error": last_error[:500],
                    "exception_type": type(e).__name__,
                    **meta,
                })
                if attempt >= _DEFAULT_HTTP_ATTEMPTS:
                    return {"error": last_error}
            except Exception as e:
                _perf_event({
                    "event": "http_end",
                    "request_id": request_id,
                    "attempt": attempt,
                    "elapsed_ms": round((time.monotonic() - t0) * 1000, 3),
                    "status": "exception",
                    "error": str(e)[:500],
                    "exception_type": type(e).__name__,
                    **meta,
                })
                return {"error": str(e)}
            time.sleep(float(attempt))
        return {"error": last_error}

    # ── Prometheus metrics ───────────────────────────────────

    def query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        step: str = "60",
    ) -> dict:
        """PromQL range query through the canonical org/cluster-scoped API."""
        return self._get(
            self._canonical_cluster_path(cluster_id, "metrics/query-range"),
            {"query": query, "start": start, "end": end, "step": step},
        )

    def query_instant(
        self,
        cluster_id: str,
        query: str,
        time: int | str | None = None,
    ) -> dict:
        """PromQL instant query through the canonical org/cluster-scoped API."""
        params: dict = {"query": query}
        if time is not None:
            params["time"] = time
        return self._get(self._canonical_cluster_path(cluster_id, "metrics/query"), params)

    def query_series(
        self,
        cluster_id: str,
        match: str,
        start: int | str,
        end: int | str,
    ) -> dict:
        """Series discovery through the canonical org/cluster-scoped API."""
        return self._get(
            self._canonical_cluster_path(cluster_id, "metrics/series"),
            {"match[]": match, "start": start, "end": end},
        )

    def legacy_query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        step: str = "60",
    ) -> dict:
        """Compatibility call to the historical metrics proxy path."""
        return self._get(
            f"/data-proxy/metrics/{self._path_segment(cluster_id)}/api/v1/query_range",
            {"query": query, "start": start, "end": end, "step": step},
        )

    def build_node_to_pod_map(
        self,
        cluster_id: str,
        namespace: str,
        pod_pattern: str,
        start: int | str,
        end: int | str,
    ) -> dict:
        """Discover `node -> pod` mapping from kube_pod_info via /series."""
        result = self.query_series(
            cluster_id,
            f'kube_pod_info{{namespace="{namespace}", pod=~"{pod_pattern}"}}',
            start,
            end,
        )
        if result.get("status") != "success":
            return result

        mapping: dict[str, str] = {}
        for series in result.get("data", []):
            node = series.get("node")
            pod = series.get("pod")
            if node and pod:
                mapping[node] = pod

        return {
            "status": "success",
            "data": {
                "resultType": "map",
                "result": [{"node": node, "pod": pod} for node, pod in sorted(mapping.items())],
            },
        }

    def query_range_join_instance_to_pod(
        self,
        cluster_id: str,
        query: str,
        namespace: str,
        pod_pattern: str,
        start: int | str,
        end: int | str,
        step: str = "60",
        instance_label: str = "instance",
        pod_label: str = "pod",
        source_pod_label: str = "source_pod",
    ) -> dict:
        """Run a range query, then attach pod labels client-side via `instance -> node -> pod`.

        This is intended for infrastructure metrics where a server-side PromQL join against
        `kube_pod_info` may return empty through Clinic Data Proxy even though both sides
        individually have data.
        """
        mapping_result = self.build_node_to_pod_map(
            cluster_id, namespace, pod_pattern, start, end
        )
        if mapping_result.get("status") != "success":
            return mapping_result

        node_to_pod = {
            item["node"]: item["pod"]
            for item in mapping_result.get("data", {}).get("result", [])
            if item.get("node") and item.get("pod")
        }

        raw = self.query_range(cluster_id, query, start, end, step=step)
        if raw.get("status") != "success":
            return raw

        joined = []
        for series in raw.get("data", {}).get("result", []):
            metric = dict(series.get("metric", {}))
            instance = metric.get(instance_label)
            pod = node_to_pod.get(instance)
            if not pod:
                continue

            if pod_label in metric and metric[pod_label] != pod:
                metric[source_pod_label] = metric[pod_label]
            metric[pod_label] = pod

            joined.append({"metric": metric, "values": series.get("values", [])})

        return {
            "status": "success",
            "data": {
                "resultType": raw.get("data", {}).get("resultType", "matrix"),
                "result": joined,
            },
        }

    def query_exceeded_iops_by_pod(
        self,
        cluster_id: str,
        component: str,
        start: int | str,
        end: int | str,
        step: str = "60",
        rate_window: str = "2m",
        instance_filter: str = ".*",
        namespace: str | None = None,
    ) -> dict:
        """Client-side joined `aws_ebs_csi_exceeded_iops_seconds_total` by storage pod.

        `component` must be `tikv` or `tiflash`.
        Use the same `rate_window` as the dashboard's `$__rate_interval` when comparing results.
        """
        patterns = {"tikv": ".*tikv.*", "tiflash": ".*tiflash.*"}
        if component not in patterns:
            return {"error": f"unsupported component: {component}"}

        resolved_namespace = namespace or self.byoc_data_plane_namespace(cluster_id)
        query = (
            f'rate(aws_ebs_csi_exceeded_iops_seconds_total'
            f'{{instance=~"{instance_filter}"}}[{rate_window}])'
        )
        return self.query_range_join_instance_to_pod(
            cluster_id,
            query,
            resolved_namespace,
            patterns[component],
            start,
            end,
            step=step,
        )

    def query_node_disk_iops_by_pod(
        self,
        cluster_id: str,
        component: str,
        start: int | str,
        end: int | str,
        step: str = "60",
        rate_window: str = "1m",
        instance_filter: str = ".*",
        namespace: str | None = None,
    ) -> dict:
        """Client-side joined node disk IOPS by storage pod."""
        patterns = {"tikv": ".*tikv.*", "tiflash": ".*tiflash.*"}
        if component not in patterns:
            return {"error": f"unsupported component: {component}"}

        resolved_namespace = namespace or self.byoc_data_plane_namespace(cluster_id)
        query = (
            f'('
            f'rate(node_disk_reads_completed_total{{instance=~"{instance_filter}"}}[{rate_window}]) + '
            f'rate(node_disk_writes_completed_total{{instance=~"{instance_filter}"}}[{rate_window}])'
            f')'
        )
        return self.query_range_join_instance_to_pod(
            cluster_id,
            query,
            resolved_namespace,
            patterns[component],
            start,
            end,
            step=step,
        )

    # ── Loki business logs ──────────────────────────────────

    def logs_query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        limit: int | None = 100,
        direction: str | None = "backward",
        step: int | str | None = None,
    ) -> dict:
        """Canonical Loki range query with parameters selected for stream or metric results."""
        params: dict = {
            "query": query,
            "start": start,
            "end": end,
        }
        if step is not None:
            params["step"] = step
        if limit is not None:
            params["limit"] = int(limit)
        if direction is not None:
            params["direction"] = direction
        return self._get(
            self._canonical_cluster_path(cluster_id, "logs/query-range"),
            params,
        )

    def logs_labels(
        self,
        cluster_id: str,
        start: int | str | None = None,
        end: int | str | None = None,
    ) -> dict:
        params: dict = {}
        if start is not None and end is not None:
            params.update({"start": start, "end": end})
        return self._get(self._canonical_cluster_path(cluster_id, "logs/labels"), params)

    def logs_label_values(
        self,
        cluster_id: str,
        label_name: str,
        start: int | str | None = None,
        end: int | str | None = None,
    ) -> dict:
        params: dict = {}
        if start is not None and end is not None:
            params.update({"start": start, "end": end})
        return self._get(
            self._canonical_cluster_path(
                cluster_id,
                f"logs/labels/{self._path_segment(label_name)}/values",
            ),
            params,
        )

    def logs_series(
        self,
        cluster_id: str,
        matches: str | list[str],
        start: int | str | None = None,
        end: int | str | None = None,
    ) -> dict:
        params: dict = {"match[]": [matches] if isinstance(matches, str) else matches}
        if start is not None and end is not None:
            params.update({"start": start, "end": end})
        return self._get(self._canonical_cluster_path(cluster_id, "logs/series"), params)

    def legacy_logs_query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        limit: int = 100,
        direction: str = "backward",
    ) -> dict:
        return self._get(
            f"/data-proxy/loki/{self._path_segment(cluster_id)}/api/v1/query_range",
            {
                "query": query,
                "start": start,
                "end": end,
                "limit": int(limit),
                "direction": direction,
            },
        )

    # ── Data Proxy SQL (slow query / TopSQL / LogStats) ──────

    def query_sql(
        self,
        cluster_id: str,
        sql: str,
        timeout: int = 60,
    ) -> dict:
        """SQL query through canonical POST .../sql-queries.

        Note: ``rows`` in the response may be ``None`` when no rows match.
        This method normalises it to ``[]`` so callers can safely iterate.
        All row values are **strings** regardless of column type.
        """
        result = self._post_json(
            self._canonical_cluster_path(cluster_id, "sql-queries"),
            {"sql": sql, "timeout": timeout},
        )
        if isinstance(result, dict) and result.get("rows") is None:
            result["rows"] = []
        return result

    def legacy_query_sql(
        self,
        cluster_id: str,
        sql: str,
        timeout: int = 60,
    ) -> dict:
        """Compatibility call to POST /data-proxy/query."""
        result = self._post_json(
            "/data-proxy/query",
            {"sql": sql, "clusterId": cluster_id, "timeout": timeout},
        )
        if isinstance(result, dict) and result.get("rows") is None:
            result["rows"] = []
        return result

    # ── BYOC customer-side O11y diagnosis APIs ──────────────

    def byoc_slow_query_available_fields(self, cluster_id: str) -> list | dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "slow-queries/available-fields")
        )

    def byoc_slow_query_list(
        self,
        cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
        limit: int = 100,
        order_by: str = "timestamp",
        desc: bool = False,
        fields: str = "query,timestamp,query_time,memory_max",
        text: str | None = None,
        digest: str | None = None,
        show_internal: bool = False,
        db: list[str] | None = None,
        plans: list[str] | None = None,
    ) -> list | dict:
        params: dict = {
            "begin_time": begin_time,
            "end_time": end_time,
            "limit": int(limit),
            "orderBy": order_by,
            "desc": str(desc).lower(),
            "fields": fields,
            "show_internal": str(show_internal).lower(),
        }
        if text is not None:
            params["text"] = text
        if digest is not None:
            params["digest"] = digest
        if db:
            params["db"] = db
        if plans:
            params["plans"] = plans
        return self._get(self._canonical_cluster_path(cluster_id, "slow-queries"), params)

    def byoc_slow_query_detail(
        self,
        cluster_id: str,
        digest: str,
        timestamp: int | float | str,
        connect_id: int | str,
        show_internal: bool = False,
    ) -> dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "slow-queries/detail"),
            {
                "digest": digest,
                "timestamp": timestamp,
                "connect_id": connect_id,
                "show_internal": str(show_internal).lower(),
            },
        )

    def byoc_top_sql_instances(
        self,
        cluster_id: str,
        start: int | float | str,
        end: int | float | str,
    ) -> dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "top-sql/instances"),
            {"start": start, "end": end},
        )

    def byoc_top_sql_summary(
        self,
        cluster_id: str,
        start: int | float | str,
        end: int | float | str,
        window: str,
        top: int,
        instance: str,
        instance_type: str,
    ) -> dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "top-sql/summary"),
            {
                "start": start,
                "end": end,
                "window": window,
                "top": int(top),
                "instance": instance,
                "instance_type": instance_type,
            },
        )

    def byoc_statements_available_fields(self, cluster_id: str) -> list | dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "statements/available-fields")
        )

    def byoc_statements_statement_types(self, cluster_id: str) -> list | dict:
        return self._get(
            self._canonical_cluster_path(cluster_id, "statements/statement-types")
        )

    def byoc_statements_list(
        self,
        cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
        schemas: list[str] | None = None,
        stmt_types: list[str] | None = None,
        text: str | None = None,
        fields: str | None = None,
    ) -> list | dict:
        params: dict = {"begin_time": begin_time, "end_time": end_time}
        if schemas:
            params["schemas"] = schemas
        if stmt_types:
            params["stmt_types"] = stmt_types
        if text is not None:
            params["text"] = text
        if fields is not None:
            params["fields"] = fields
        return self._get(self._canonical_cluster_path(cluster_id, "statements"), params)

    def byoc_statements_plans(
        self,
        cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
        digest: str,
        schema_name: str | None = None,
    ) -> list | dict:
        params: dict = {
            "begin_time": begin_time,
            "end_time": end_time,
            "digest": digest,
        }
        if schema_name is not None:
            params["schema_name"] = schema_name
        return self._get(self._canonical_cluster_path(cluster_id, "statements/plans"), params)

    def byoc_statements_plan_detail(
        self,
        cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
        digest: str,
        plans: list[str] | None = None,
        schema_name: str | None = None,
    ) -> dict:
        params: dict = {
            "begin_time": begin_time,
            "end_time": end_time,
            "digest": digest,
        }
        if plans:
            params["plans"] = plans
        if schema_name is not None:
            params["schema_name"] = schema_name
        return self._get(
            self._canonical_cluster_path(cluster_id, "statements/plan-detail"),
            params,
        )

    def byoc_continuous_profiling_group_profiles(
        self,
        physical_cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
    ) -> list | dict:
        return self._get(
            self._canonical_cluster_path(
                physical_cluster_id,
                "continuous-profiling/group-profiles",
            ),
            {"begin_time": begin_time, "end_time": end_time},
        )

    def byoc_continuous_profiling_group_detail(
        self,
        physical_cluster_id: str,
        ts: int | str,
    ) -> dict | None:
        return self._get(
            self._canonical_cluster_path(
                physical_cluster_id,
                f"continuous-profiling/group-profiles/{self._path_segment(ts)}",
            )
        )

    # ── Schema ───────────────────────────────────────────────

    def get_schema(self, cluster_id: str, tables: str = "slow_query_logs,topsql") -> dict:
        """Table schema → GET /data-proxy/schema"""
        return self._get("/data-proxy/schema", {"tables": tables, "clusterId": cluster_id})

    # ── LogStats convenience helpers ─────────────────────────

    @staticmethod
    def _validate_log_stats_date(date: int | str) -> str:
        value = str(date)
        if len(value) != 8 or not value.isdigit():
            raise ValueError("LogStats date must be YYYYMMDD, for example 20260618")
        return value

    @staticmethod
    def _log_stats_int(value: object) -> int:
        try:
            return int(float(str(value)))
        except (TypeError, ValueError):
            return 0

    @staticmethod
    def _log_stats_rows_as_dicts(result: dict) -> list[dict]:
        columns = [str(column) for column in result.get("columns") or []]
        rows = []
        for row in result.get("rows") or []:
            if isinstance(row, dict):
                rows.append(row)
            elif isinstance(row, list):
                rows.append({
                    columns[index]: row[index] if index < len(row) else ""
                    for index in range(len(columns))
                })
        return rows

    @classmethod
    def _log_stats_group_rows(
        cls,
        result: dict,
        group_columns: list[str],
        limit: int,
    ) -> dict:
        if result.get("error"):
            return result
        grouped: dict[tuple, dict] = {}
        for row in cls._log_stats_rows_as_dicts(result):
            key = tuple(str(row.get(column) or "") for column in group_columns)
            item = grouped.setdefault(
                key,
                {column: key[index] for index, column in enumerate(group_columns)} | {"total_count": 0},
            )
            item["total_count"] += cls._log_stats_int(row.get("count"))
        columns = [*group_columns, "total_count"]
        rows = [
            [item.get(column, "") for column in columns]
            for item in sorted(grouped.values(), key=lambda value: value["total_count"], reverse=True)[:int(limit)]
        ]
        return {"columns": columns, "rows": rows}

    def log_stats_schema(self, cluster_id: str) -> dict:
        """LogStats table schema → GET /data-proxy/schema?tables=log_stats"""
        return self.get_schema(cluster_id, "log_stats")

    def log_stats_summary(
        self,
        cluster_id: str,
        date: int | str,
        limit: int = 50,
        timeout: int = 120,
    ) -> dict:
        """Aggregate LogStats by level/component for one YYYYMMDD date."""
        d = self._validate_log_stats_date(date)
        raw_limit = max(int(limit) * 100, 1000)
        sql = f"""
SELECT
  level,
  component,
  count
FROM log_stats
WHERE date = {d}
LIMIT {raw_limit}
""".strip()
        return self._log_stats_group_rows(
            self.query_sql(cluster_id, sql, timeout=timeout),
            ["level", "component"],
            limit,
        )

    def log_stats_top(
        self,
        cluster_id: str,
        date: int | str,
        limit: int = 50,
        timeout: int = 120,
    ) -> dict:
        """Top LogStats messages for one YYYYMMDD date."""
        d = self._validate_log_stats_date(date)
        raw_limit = max(int(limit) * 100, 1000)
        sql = f"""
SELECT
  level,
  component,
  file,
  line,
  msg_digest,
  sample,
  count
FROM log_stats
WHERE date = {d}
LIMIT {raw_limit}
""".strip()
        return self._log_stats_group_rows(
            self.query_sql(cluster_id, sql, timeout=timeout),
            ["level", "component", "file", "line", "msg_digest", "sample"],
            limit,
        )

    def log_stats_search(
        self,
        cluster_id: str,
        date: int | str,
        pattern: str,
        limit: int = 50,
        timeout: int = 120,
    ) -> dict:
        """Search LogStats sample/digest/file/component for a substring."""
        d = self._validate_log_stats_date(date)
        escaped = pattern.replace("\\", "\\\\").replace("'", "''")
        like = f"%{escaped}%"
        raw_limit = max(int(limit) * 100, 1000)
        sql = f"""
SELECT
  level,
  component,
  file,
  line,
  msg_digest,
  sample,
  count
FROM log_stats
WHERE date = {d}
  AND (
    sample LIKE '{like}'
    OR msg_digest LIKE '{like}'
    OR file LIKE '{like}'
    OR component LIKE '{like}'
  )
LIMIT {raw_limit}
""".strip()
        return self._log_stats_group_rows(
            self.query_sql(cluster_id, sql, timeout=timeout),
            ["level", "component", "file", "line", "msg_digest", "sample"],
            limit,
        )


# ── CLI: python3 clinic_api.py <method> <args...> ─────────

def _cli():
    """Minimal CLI for quick testing."""
    if len(sys.argv) < 2:
        print("Usage: python3 clinic_api.py <method> <args...>")
        print("  query_range  <cluster_id> <query> <start> <end> [step]")
        print("  query_sql    <cluster_id> <sql> [timeout]")
        print("  query_instant <cluster_id> <query> [time]")
        print("  log_stats_schema  <cluster_id>")
        print("  log_stats_summary <cluster_id> <YYYYMMDD> [limit] [timeout]")
        print("  log_stats_top     <cluster_id> <YYYYMMDD> [limit] [timeout]")
        print("  log_stats_search  <cluster_id> <YYYYMMDD> <pattern> [limit] [timeout]")
        print("  query_exceeded_iops_by_pod <cluster_id> <tikv|tiflash> <start> <end> [step] [rate_window=2m]")
        print("  query_node_disk_iops_by_pod <cluster_id> <tikv|tiflash> <start> <end> [step] [rate_window=1m]")
        sys.exit(1)

    api = ClinicAPI()
    method = sys.argv[1]

    if method == "query_range":
        result = api.query_range(sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5],
                                  sys.argv[6] if len(sys.argv) > 6 else "60")
    elif method == "query_sql":
        result = api.query_sql(sys.argv[2], sys.argv[3],
                                int(sys.argv[4]) if len(sys.argv) > 4 else 60)
    elif method == "query_instant":
        result = api.query_instant(sys.argv[2], sys.argv[3],
                                    sys.argv[4] if len(sys.argv) > 4 else None)
    elif method == "log_stats_schema":
        result = api.log_stats_schema(sys.argv[2])
    elif method == "log_stats_summary":
        result = api.log_stats_summary(
            sys.argv[2],
            sys.argv[3],
            int(sys.argv[4]) if len(sys.argv) > 4 else 50,
            int(sys.argv[5]) if len(sys.argv) > 5 else 120,
        )
    elif method == "log_stats_top":
        result = api.log_stats_top(
            sys.argv[2],
            sys.argv[3],
            int(sys.argv[4]) if len(sys.argv) > 4 else 50,
            int(sys.argv[5]) if len(sys.argv) > 5 else 120,
        )
    elif method == "log_stats_search":
        result = api.log_stats_search(
            sys.argv[2],
            sys.argv[3],
            sys.argv[4],
            int(sys.argv[5]) if len(sys.argv) > 5 else 50,
            int(sys.argv[6]) if len(sys.argv) > 6 else 120,
        )
    elif method == "query_exceeded_iops_by_pod":
        result = api.query_exceeded_iops_by_pod(
            sys.argv[2],
            sys.argv[3],
            sys.argv[4],
            sys.argv[5],
            sys.argv[6] if len(sys.argv) > 6 else "60",
            sys.argv[7] if len(sys.argv) > 7 else "2m",
        )
    elif method == "query_node_disk_iops_by_pod":
        result = api.query_node_disk_iops_by_pod(
            sys.argv[2],
            sys.argv[3],
            sys.argv[4],
            sys.argv[5],
            sys.argv[6] if len(sys.argv) > 6 else "60",
            sys.argv[7] if len(sys.argv) > 7 else "1m",
        )
    else:
        print(f"Unknown method: {method}")
        sys.exit(1)

    json.dump(result, sys.stdout, indent=2, ensure_ascii=False)
    print()


if __name__ == "__main__":
    _cli()