#!/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 math
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from threading import Lock

_ENV_ENDPOINTS = {
    "dev": "https://dev-clinic.pingcap.com",
    "staging": "https://staging-clinic.pingcap.com",
    "prod": "https://clinic.pingcap.com",
}


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 _do(req: urllib.request.Request) -> dict | list:
        try:
            with urllib.request.urlopen(req, timeout=120) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError as e:
            body = e.read().decode(errors="replace")
            return {"error": f"HTTP {e.code}: {body[:500]}"}
        except Exception as e:
            return {"error": str(e)}

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

    def get_cluster_by_id(self, cluster_id: str) -> dict:
        """Canonical exact metadata → GET /data-proxy/api/v1/clusters/{cluster_id}."""
        return self._get(
            f"/data-proxy/api/v1/clusters/{self._path_segment(cluster_id)}"
        )

    def byoc_data_plane_namespace(self, cluster_id: str) -> str:
        """Return the Kubernetes namespace used by BYOC infrastructure metrics.

        Logical BYOC clusters use their physical ``diagnosticContext.parentId``
        (with top-level ``parentId`` compatibility fallback); physical
        ``byoc-host`` clusters use their own ID. Other tiers keep the historical
        ``tidb<cluster_id>`` convention.
        """
        metadata = self.get_cluster_by_id(cluster_id)
        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"])
        deploy_type = str(metadata.get("deployType") or "").lower()
        physical_id = str(cluster_id)
        if deploy_type == "byoc":
            context = metadata.get("diagnosticContext") or {}
            context_parent = context.get("parentId") if isinstance(context, dict) else None
            physical_id = str(context_parent or metadata.get("parentId") or "")
            if not physical_id:
                raise RuntimeError(
                    f"Could not resolve physical parent ID for logical BYOC cluster {cluster_id}"
                )
        return f"tidb{physical_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)
        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/{org_id}/clusters/"
            f"{self._path_segment(cluster_id)}"
        )
        return f"{path}/{suffix.lstrip('/')}" if suffix else path

    # ── 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_range_for_target(
        self,
        org_id: str,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        step: str = "60",
    ) -> dict:
        """Range query for an already resolved Cloud org/cluster pair.

        This deliberately skips metadata lookup. Callers that already own the
        typed target should not spend another request rediscovering it.
        """
        path = (
            f"/data-proxy/api/v1/orgs/{self._path_segment(org_id)}/clusters/"
            f"{self._path_segment(cluster_id)}/metrics/query-range"
        )
        return self._get(
            path,
            {"query": query, "start": start, "end": end, "step": step},
        )

    def query_range_batch(self, plan: dict, context: dict | None = None) -> dict:
        """Execute a bounded metrics plan and recover cross-source queries.

        Independent initial queries run concurrently. If Clinic rejects one
        query with ``cross_source_metrics_rejected``, the helper executes the
        query's predeclared source-homogeneous ``fallbackQueries``. When no
        fallback is declared, a top-level PromQL ``or`` is split without
        rewriting either arm. The helper never asks an agent to devise a retry
        after the provider error.
        """
        normalized = _validate_metrics_batch_plan(plan, context)
        target = normalized["target"]
        started_at = time.time()
        recovery_budget = {"remaining": 6 - len(normalized["queries"])}
        recovery_budget_lock = Lock()

        def run(query: dict) -> dict:
            query_started = time.time()
            response = self.query_range_for_target(
                target["orgId"],
                target["clusterId"],
                query["query"],
                normalized["start"],
                normalized["end"],
                normalized["step"],
            )
            recovery = None
            if _cross_source_metrics_rejected(response):
                fallback_queries = query.get("fallbackQueries") or [
                    {"source": "auto", "query": part}
                    for part in _split_top_level_or(query["query"])
                ]
                with recovery_budget_lock:
                    may_recover = 1 < len(fallback_queries) <= recovery_budget["remaining"]
                    if may_recover:
                        recovery_budget["remaining"] -= len(fallback_queries)
                if may_recover:
                    recovery = self._run_cross_source_recovery(
                        normalized,
                        target,
                        fallback_queries,
                    )
                    response = recovery["combinedResponse"]
            summary = _summarize_metrics_response(response)
            status = "failed" if summary["errorCode"] else (
                "empty" if summary["seriesCount"] == 0 else "complete"
            )
            statement = _metrics_batch_statement(
                query["queryId"],
                query.get("description"),
                status,
                summary,
                recovery,
            )
            return {
                "queryId": query["queryId"],
                "source": query["source"],
                "status": status,
                "statement": statement,
                "durationMs": round((time.time() - query_started) * 1000),
                "summary": summary,
                "recovery": _public_recovery(recovery),
            }

        queries = normalized["queries"]
        by_index: dict[int, dict] = {}
        with ThreadPoolExecutor(max_workers=min(6, len(queries))) as executor:
            futures = {executor.submit(run, query): index for index, query in enumerate(queries)}
            for future in as_completed(futures):
                index = futures[future]
                try:
                    by_index[index] = future.result()
                except Exception as error:
                    query = queries[index]
                    by_index[index] = {
                        "queryId": query["queryId"],
                        "source": query["source"],
                        "status": "failed",
                        "statement": f"Metric query {query['queryId']} failed before producing evidence.",
                        "durationMs": 0,
                        "summary": _failed_metrics_summary(str(error)),
                        "recovery": None,
                    }

        results = [by_index[index] for index in range(len(queries))]
        completion = "complete" if all(
            item["status"] == "complete" for item in results
        ) else "partial"
        return {
            "schemaVersion": "clinic-metrics-batch-v1",
            "scenarioId": normalized["scenarioId"],
            "timeWindow": normalized["timeWindow"],
            "completion": completion,
            "startedAt": _utc_timestamp(started_at),
            "durationMs": round((time.time() - started_at) * 1000),
            "queries": results,
        }

    def _run_cross_source_recovery(
        self,
        plan: dict,
        target: dict,
        fallback_queries: list[dict],
    ) -> dict:
        def run(part: dict) -> dict:
            return self.query_range_for_target(
                target["orgId"],
                target["clusterId"],
                part["query"],
                plan["start"],
                plan["end"],
                plan["step"],
            )

        with ThreadPoolExecutor(max_workers=min(6, len(fallback_queries))) as executor:
            responses = list(executor.map(run, fallback_queries))
        success = all(not _response_error(response) for response in responses)
        return {
            "triggered": True,
            "code": "cross_source_metrics_rejected",
            "method": "source_homogeneous_parallel_split",
            "succeeded": success,
            "parts": [
                {
                    "source": part.get("source", "auto"),
                    "status": "complete" if not _response_error(response) else "failed",
                }
                for part, response in zip(fallback_queries, responses)
            ],
            "combinedResponse": _combine_metrics_responses(responses),
        }

    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},
        )

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

    def logs_query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        limit: int = 100,
        direction: str = "backward",
    ) -> dict:
        """Canonical business-log range query for any supported tier, including BYOC."""
        return self._get(
            self._canonical_cluster_path(cluster_id, "logs/query-range"),
            {
                "query": query,
                "start": start,
                "end": end,
                "limit": int(limit),
                "direction": direction,
            },
        )

    def logs_labels(
        self,
        cluster_id: str,
        start: int | str | None = None,
        end: int | str | None = None,
    ) -> dict:
        """List canonical business-log label names."""
        params = {}
        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:
        """List values for one canonical business-log label."""
        params = {}
        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:
        """Discover canonical business-log series with one or more match selectors."""
        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_query_range(
        self,
        cluster_id: str,
        query: str,
        start: int | str,
        end: int | str,
        step: str = "60",
    ) -> dict:
        """Compatibility call to legacy cluster-only metrics query_range."""
        return self._get(
            f"/data-proxy/metrics/{cluster_id}/api/v1/query_range",
            {"query": query, "start": start, "end": end, "step": step},
        )

    def legacy_query_instant(
        self,
        cluster_id: str,
        query: str,
        time: int | str | None = None,
    ) -> dict:
        """Compatibility call to legacy cluster-only metrics query."""
        params: dict = {"query": query}
        if time is not None:
            params["time"] = time
        return self._get(f"/data-proxy/metrics/{cluster_id}/api/v1/query", params)

    def legacy_query_series(
        self,
        cluster_id: str,
        match: str,
        start: int | str,
        end: int | str,
    ) -> dict:
        """Compatibility call to legacy cluster-only metrics series."""
        return self._get(
            f"/data-proxy/metrics/{cluster_id}/api/v1/series",
            {"match[]": match, "start": start, "end": end},
        )

    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,
        )

    # ── 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 query_sql_batch(self, cluster_id: str, queries: list[dict]) -> dict | list:
        """Run up to 100 same-cluster canonical SQL queries in request order."""
        return self._post_json(
            self._canonical_cluster_path(cluster_id, "sql-query-batches"),
            queries,
        )

    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

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

    def get_schema(self, cluster_id: str, tables: str = "slow_query_logs") -> dict:
        """Table schemas through canonical GET .../table-schemas."""
        return self._get(
            self._canonical_cluster_path(cluster_id, "table-schemas"),
            {"tables": tables},
        )

    def get_capabilities(self) -> dict:
        """Service capabilities → GET /data-proxy/api/v1/capabilities."""
        return self._get("/data-proxy/api/v1/capabilities")

    def legacy_get_schema(
        self,
        cluster_id: str,
        tables: str = "slow_query_logs,topsql",
    ) -> dict:
        """Compatibility call to GET /data-proxy/schema."""
        return self._get("/data-proxy/schema", {"tables": tables, "clusterId": cluster_id})

    # ── NGM slow query (starter / essential / shared) ────────

    def search_cluster_by_id(
        self,
        cluster_id: str,
        show_deleted: bool = True,
        limit: int = 10,
    ) -> dict:
        """Legacy Dashboard lookup used only for NGM context discovery."""
        return self._api_get(
            "/dashboard/clusters2",
            {
                "cluster_id": cluster_id,
                "show_deleted": str(show_deleted).lower(),
                "limit": limit,
                "page": 1,
                "include_total": "false",
            },
        )

    def search_clusters(
        self,
        query: str = "",
        show_deleted: bool = False,
        limit: int = 10,
        page: int = 1,
        *,
        cluster_ids: list[str] | None = None,
        cluster_statuses: list[str] | None = None,
        cluster_versions: list[str] | None = None,
        deploy_types: list[str] | None = None,
        deploy_types_v2: list[str] | None = None,
        tenant_plans: list[str] | None = None,
        control_planes: list[str] | None = None,
        providers: list[str] | None = None,
        regions: list[str] | None = None,
        sort: str = "",
        order: str = "",
        include_total: bool = False,
    ) -> dict:
        """Global cluster discovery with server-side Dashboard filters.

        This route-neutral public helper uses the compatibility transport only
        because the canonical org list does not expose these filter dimensions.
        """
        if int(limit) < 1 or int(page) < 1:
            raise ValueError("limit and page must be positive integers")
        params: dict = {
            "query": query,
            "show_deleted": str(show_deleted).lower(),
            "limit": int(limit),
            "page": int(page),
            "include_total": str(include_total).lower(),
        }
        repeated = {
            "cluster_id": cluster_ids,
            "cluster_status": cluster_statuses,
            "cluster_version": cluster_versions,
            "deploy_type": deploy_types,
            "deploy_type_v2": deploy_types_v2,
            "tenant_plan": tenant_plans,
            "control_plane": control_planes,
            "provider": providers,
            "region": regions,
        }
        params.update({key: value for key, value in repeated.items() if value})
        if sort:
            params["sort"] = sort
        if order:
            params["order"] = order
        return self._api_get("/dashboard/clusters2", params)

    def get_cluster(self, org_id: str, cluster_id: str) -> dict:
        """Canonical cluster detail; org_id is the DBAAS/TiDB Cloud org ID."""
        return self._get(
            f"/data-proxy/api/v1/orgs/{org_id}/clusters/{self._path_segment(cluster_id)}"
        )

    def list_clusters(
        self,
        org_id: str,
        query: str = "",
        show_deleted: bool = False,
        page_size: int = 50,
        page: int = 1,
    ) -> dict:
        """Canonical org-scoped cluster list."""
        return self._get(
            f"/data-proxy/api/v1/orgs/{org_id}/clusters",
            {
                "query": query,
                "showDeleted": str(show_deleted).lower(),
                "pageSize": page_size,
                "page": page,
            },
        )

    def get_components(self, org_id: str, cluster_id: str) -> dict | list:
        """Canonical topology/components for supported non-shared tiers."""
        return self._get(
            f"/data-proxy/api/v1/orgs/{org_id}/clusters/"
            f"{self._path_segment(cluster_id)}/components"
        )

    def get_activities(
        self,
        org_id: str,
        cluster_id: str,
        begin_ts: int | str,
        end_ts: int | str,
        limit: int = 100,
        cursor: str = "",
        catalog: str = "",
        activity_type: str = "",
        name: str = "",
        severity: int | None = None,
    ) -> dict:
        """Canonical Activity Events; pagination cursor is a query parameter."""
        params: dict = {"begin_ts": begin_ts, "end_ts": end_ts, "limit": limit}
        optional = {
            "cursor": cursor,
            "catalog": catalog,
            "type": activity_type,
            "name": name,
        }
        params.update({key: value for key, value in optional.items() if value != ""})
        if severity is not None:
            params["severity"] = severity
        return self._get(
            f"/data-proxy/api/v1/orgs/{org_id}/clusters/"
            f"{self._path_segment(cluster_id)}/activities",
            params,
        )

    def top_ru_summary(
        self,
        cluster_id: str,
        start: int | str,
        end: int | str,
        top: int,
        order_by: str | None = None,
        data_source: str | None = None,
        step: str | None = None,
        user: str | None = None,
    ) -> dict:
        """Canonical Top RU summary for supported Premium/Essential clusters."""
        params: dict = {"start": start, "end": end, "top": top}
        for key, value in (
            ("order_by", order_by),
            ("data_source", data_source),
            ("step", step),
            ("user", user),
        ):
            if value is not None:
                params[key] = value
        return self._get(self._canonical_cluster_path(cluster_id, "top-ru/summary"), params)

    def top_ru_summary_plan(
        self,
        cluster_id: str,
        start: int | str,
        end: int | str,
        top: int,
        digest: str | None = None,
        order_by: str | None = None,
        data_source: str | None = None,
        step: str | None = None,
        user: str | None = None,
    ) -> dict:
        """Canonical Top RU SQL/plan aggregation for Premium/Essential.

        ``user=None`` omits the filter; ``user=""`` preserves the distinct
        empty/NULL-user filter.
        """
        params: dict = {"start": start, "end": end, "top": top}
        for key, value in (
            ("digest", digest),
            ("order_by", order_by),
            ("data_source", data_source),
            ("step", step),
            ("user", user),
        ):
            if value is not None:
                params[key] = value
        return self._get(
            self._canonical_cluster_path(cluster_id, "top-ru/summary-plan"),
            params,
        )

    def top_ru_user_stats(
        self,
        cluster_id: str,
        start: int | str,
        end: int | str,
        top: int | None = None,
    ) -> dict:
        """Canonical Top RU user statistics."""
        params: dict = {"start": start, "end": end}
        if top is not None:
            params["top"] = top
        return self._get(
            self._canonical_cluster_path(cluster_id, "top-ru/user-stats"),
            params,
        )

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

    def byoc_slow_query_available_fields(self, cluster_id: str) -> list | dict:
        """List available customer-side O11y Slow Query fields."""
        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:
        """List BYOC Slow Query rows through the canonical O11y adapter."""
        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:
        """Fetch one BYOC Slow Query detail using keys returned by list.

        Map list field ``connection_id`` to query parameter ``connect_id`` and
        preserve the list timestamp token without unit conversion.
        """
        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:
        """List BYOC Top SQL instances in a time window."""
        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:
        """Query BYOC Top SQL summary for one discovered instance."""
        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:
        """List available BYOC SQL Statements fields."""
        return self._get(
            self._canonical_cluster_path(cluster_id, "statements/available-fields")
        )

    def byoc_statements_statement_types(self, cluster_id: str) -> list | dict:
        """List available BYOC SQL statement types."""
        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:
        """List BYOC SQL Statements; repeated filters are encoded with doseq."""
        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:
        """List plans for one BYOC SQL digest."""
        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:
        """Fetch BYOC Statements plan detail; omitted plans means all plans."""
        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:
        """List profiling groups for a physical BYOC host cluster."""
        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:
        """Fetch one physical BYOC profiling group inventory detail."""
        return self._get(
            self._canonical_cluster_path(
                physical_cluster_id,
                f"continuous-profiling/group-profiles/{self._path_segment(ts)}",
            )
        )

    def legacy_get_cluster(self, clinic_internal_org_id: str, cluster_id: str) -> dict:
        """Compatibility call to Clinic detail using its internal org primary key."""
        return self._api_get(
            f"/orgs/{clinic_internal_org_id}/clusters/{cluster_id}"
        )

    legacy_search_cluster_by_id = search_cluster_by_id

    @staticmethod
    def _normalize_ngm_deploy_type(deploy_type: str) -> str:
        value = deploy_type.lower().replace(" ", "_")
        if value in {"starter", "essential", "shared"}:
            return "shared"
        return value

    def _ngm_context_from_cluster_id(self, cluster_id: str) -> dict:
        result = self.search_cluster_by_id(cluster_id, show_deleted=True)
        items = result.get("items") or []
        item = next((x for x in items if str(x.get("clusterID")) == str(cluster_id)), None)
        if not item:
            raise RuntimeError(f"Cluster not found: {cluster_id}")

        def pick(*names: str) -> str:
            for name in names:
                value = item.get(name)
                if value not in (None, "", "null"):
                    return str(value)
            return ""

        org_id = pick("orgID")
        detail = None
        fallback_keys = (
            ("tenantID",),
            ("clusterDeployType",),
            ("projectID", "projectId"),
            ("cloudProvider", "provider", "vendor", "clusterProviderName"),
            ("region", "clusterRegionName"),
        )
        if org_id and any(not pick(*names) for names in fallback_keys):
            detail = self.legacy_get_cluster(org_id, cluster_id)

        def pick_detail(*names: str) -> str:
            nonlocal detail
            value = pick(*names)
            if value:
                return value
            if not detail:
                return ""
            for name in names:
                detail_value = detail.get(name)
                if detail_value not in (None, "", "null"):
                    return str(detail_value)
            return ""

        context = {
            "x-cluster-id": cluster_id,
            "x-org-id": pick_detail("tenantID"),
            "x-project-id": pick_detail("projectID", "projectId"),
            "x-deploy-type": self._normalize_ngm_deploy_type(
                pick_detail("clusterDeployType", "deployType")
                or pick_detail("clusterDeployTypeV2", "deployTypeV2")
            ),
            "x-provider": pick_detail("cloudProvider", "provider", "vendor", "clusterProviderName").lower(),
            "x-region": pick_detail("region", "clusterRegionName"),
        }
        missing = [name for name, value in context.items() if not value]
        if missing:
            raise RuntimeError(
                f"Could not resolve NGM headers for cluster {cluster_id}: {', '.join(missing)}"
            )
        return context

    def ngm_slow_query_list(
        self,
        cluster_id: str,
        begin_time: int | str,
        end_time: int | str,
        limit: int = 100,
        order_by: str = "query_time",
        desc: bool = True,
        fields: str = "query,timestamp,query_time,memory_max,digest,index_names",
        text: str = "",
        digest: str = "",
        show_internal: bool = True,
    ) -> dict | list:
        """Slow query list via NGM → GET /ngm/api/v1/slow_query/list.

        Use this for starter/essential/shared clusters whose slow query data is
        not exposed through Data Proxy SQL `slow_query_logs`.
        """
        headers = self._ngm_context_from_cluster_id(cluster_id)
        return self._get(
            "/ngm/api/v1/slow_query/list",
            {
                "begin_time": begin_time,
                "end_time": end_time,
                "fields": fields,
                "limit": int(limit),
                "orderBy": order_by,
                "desc": str(desc).lower(),
                "text": text,
                "digest": digest,
                "show_internal": str(show_internal).lower(),
            },
            extra_headers=headers,
        )

    # ── 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

    def log_stats_schema(self, cluster_id: str) -> dict:
        """LogStats table schema through canonical table-schemas."""
        return self.get_schema(cluster_id, "log_stats")

    def statements(
        self,
        cluster_id: str,
        date: str,
        limit: int = 100,
        timeout: int = 120,
    ) -> dict:
        """Query persisted SQL Statements through canonical SQL.

        ``date`` must be a single ``YYYY-MM-DD`` partition. The physical
        provider URL and execution engine remain server-resolved.
        """
        parts = date.split("-")
        if (
            len(parts) != 3
            or list(map(len, parts)) != [4, 2, 2]
            or not date.replace("-", "").isdigit()
        ):
            raise ValueError("Statements date must be YYYY-MM-DD, for example 2026-07-22")
        sql = (
            "SELECT * FROM clinic_data_proxy.statements "
            f"WHERE date = '{date}' LIMIT {int(limit)}"
        )
        return self.query_sql(cluster_id, sql, timeout=timeout)

    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.

        LogStats currently uses AWS Delta data. Use an unquoted numeric
        YYYYMMDD date in SQL so Athena can compare the partition field without
        a varchar/bigint mismatch.
        """
        d = self._validate_log_stats_date(date)
        sql = f"""
SELECT
  level,
  component,
  SUM(count) AS total_count
FROM log_stats
WHERE date = {d}
GROUP BY level, component
ORDER BY total_count DESC
LIMIT {int(limit)}
""".strip()
        return self.query_sql(cluster_id, sql, timeout=timeout)

    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)
        sql = f"""
SELECT
  level,
  component,
  file,
  line,
  msg_digest,
  sample,
  SUM(count) AS total_count
FROM log_stats
WHERE date = {d}
GROUP BY level, component, file, line, msg_digest, sample
ORDER BY total_count DESC
LIMIT {int(limit)}
""".strip()
        return self.query_sql(cluster_id, sql, timeout=timeout)

    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}%"
        sql = f"""
SELECT
  level,
  component,
  file,
  line,
  msg_digest,
  sample,
  SUM(count) AS total_count
FROM log_stats
WHERE date = {d}
  AND (
    sample LIKE '{like}'
    OR msg_digest LIKE '{like}'
    OR file LIKE '{like}'
    OR component LIKE '{like}'
  )
GROUP BY level, component, file, line, msg_digest, sample
ORDER BY total_count DESC
LIMIT {int(limit)}
""".strip()
        return self.query_sql(cluster_id, sql, timeout=timeout)


_QUERY_ID_PATTERN = re.compile(r"^[a-z][a-z0-9-]{0,63}$")
_METRIC_LABEL_KEYS = {
    "__name__",
    "component",
    "instance",
    "job",
    "name",
    "namespace",
    "pod",
    "store",
    "type",
}

_METRICS_BATCH_RECIPES = {
    "memory-oom": [
        {
            "queryId": "alignment",
            "source": "data_plane",
            "description": "TiKV working-set memory percentage of the declared pod limit",
            "query": "100 * max by (namespace, pod) (container_memory_working_set_bytes{container=\"tikv\", image!=\"\"}) / max by (namespace, pod) (kube_pod_container_resource_limits{container=\"tikv\", resource=\"memory\", unit=\"byte\"})",
        },
        {
            "queryId": "mechanism",
            "source": "cluster_store",
            "description": "TiKV allocator statistics by instance and statistic type",
            "query": "avg by (instance, type) (tikv_allocator_stats)",
        },
        {
            "queryId": "strongest-alternative",
            "source": "cluster_store",
            "description": "TiKV engine and block-cache memory by instance, database, and column family",
            "query": "avg by (instance, db, cf, type) (tikv_engine_memory_bytes) or avg by (instance, db, cf) (tikv_engine_block_cache_size_bytes)",
        },
    ],
    "capacity-resource-pressure": [
        {
            "queryId": "alignment",
            "source": "data_plane",
            "description": "TiKV working-set memory percentage of the declared pod limit",
            "query": "100 * max by (namespace, pod) (container_memory_working_set_bytes{container=\"tikv\", image!=\"\"}) / max by (namespace, pod) (kube_pod_container_resource_limits{container=\"tikv\", resource=\"memory\", unit=\"byte\"})",
        },
        {
            "queryId": "mechanism",
            "source": "cluster_store",
            "description": "TiKV thread CPU consumption by instance and thread name",
            "query": "sum by (instance, name) (rate(tikv_thread_cpu_seconds_total[1m]))",
        },
        {
            "queryId": "strongest-alternative",
            "source": "cluster_store",
            "description": "TiKV Region and leader distribution by instance",
            "query": "sum by (instance, type) (tikv_raftstore_region_count) or sum by (instance) (tikv_raftstore_leader_count)",
        },
    ],
}


def _validate_metrics_batch_plan(plan: dict, context: dict | None = None) -> dict:
    if not isinstance(plan, dict) or plan.get("schemaVersion") != "clinic-metrics-batch-plan-v1":
        raise ValueError("clinic_metrics_batch_plan_invalid")
    context_values = _validate_metrics_batch_context(context) if context is not None else None
    target = context_values["target"] if context_values else plan.get("target")
    if not isinstance(target, dict) or not str(target.get("orgId", "")).strip() \
            or not str(target.get("clusterId", "")).strip():
        raise ValueError("clinic_metrics_batch_target_missing")
    recipe = str(plan.get("recipe", "")).strip()
    queries = _METRICS_BATCH_RECIPES.get(recipe) if recipe else plan.get("queries")
    if recipe and context_values and recipe != context_values["scenarioId"]:
        raise ValueError("clinic_metrics_batch_recipe_scenario_mismatch")
    if not isinstance(queries, list) or not 1 <= len(queries) <= 6:
        raise ValueError("clinic_metrics_batch_query_count_invalid")
    normalized_queries = []
    seen = set()
    for item in queries:
        if not isinstance(item, dict):
            raise ValueError("clinic_metrics_batch_query_invalid")
        query_id = str(item.get("queryId", ""))
        query = str(item.get("query", "")).strip()
        source = str(item.get("source", "auto"))
        if not _QUERY_ID_PATTERN.fullmatch(query_id) or query_id in seen or not query:
            raise ValueError("clinic_metrics_batch_query_invalid")
        if source not in {"auto", "data_plane", "cluster_store", "mixed"}:
            raise ValueError("clinic_metrics_batch_source_invalid")
        fallback_queries = item.get("fallbackQueries", [])
        if not isinstance(fallback_queries, list) or len(fallback_queries) > 6:
            raise ValueError("clinic_metrics_batch_fallback_invalid")
        normalized_fallbacks = []
        for fallback in fallback_queries:
            if not isinstance(fallback, dict):
                raise ValueError("clinic_metrics_batch_fallback_invalid")
            fallback_source = str(fallback.get("source", "auto"))
            fallback_query = str(fallback.get("query", "")).strip()
            if fallback_source not in {"auto", "data_plane", "cluster_store"} \
                    or not fallback_query:
                raise ValueError("clinic_metrics_batch_fallback_invalid")
            normalized_fallbacks.append({
                "source": fallback_source,
                "query": fallback_query,
            })
        seen.add(query_id)
        normalized_queries.append({
            "queryId": query_id,
            "source": source,
            "query": query,
            "description": str(item.get("description", "")).strip()[:240],
            "fallbackQueries": normalized_fallbacks,
        })
    if len(normalized_queries) + sum(
        len(item["fallbackQueries"]) for item in normalized_queries
    ) > 6:
        raise ValueError("clinic_metrics_batch_provider_query_budget_exceeded")
    time_window = context_values["timeWindow"] if context_values else plan.get("timeWindow")
    start, end = _historical_window_bounds(time_window)
    return {
        "schemaVersion": "clinic-metrics-batch-plan-v1",
        "scenarioId": context_values["scenarioId"] if context_values \
            else str(plan.get("scenarioId", "")).strip(),
        "timeWindow": time_window,
        "target": {
            "orgId": str(target["orgId"]).strip(),
            "clusterId": str(target["clusterId"]).strip(),
        },
        "start": start if start is not None else plan.get("start"),
        "end": end if end is not None else plan.get("end"),
        "step": str(plan.get("step", "60")),
        "queries": normalized_queries,
    }


def _validate_metrics_batch_context(context: dict) -> dict:
    if not isinstance(context, dict) \
            or context.get("schemaVersion") != "auto-oncall-clinic-metrics-context-v1":
        raise ValueError("clinic_metrics_batch_context_invalid")
    target = context.get("target")
    scenario_id = str(context.get("scenarioId", "")).strip()
    if not scenario_id or not isinstance(target, dict) \
            or not str(target.get("orgId", "")).strip() \
            or not str(target.get("clusterId", "")).strip():
        raise ValueError("clinic_metrics_batch_context_invalid")
    return {
        "scenarioId": scenario_id,
        "timeWindow": context.get("timeWindow"),
        "target": {
            "orgId": str(target["orgId"]).strip(),
            "clusterId": str(target["clusterId"]).strip(),
        },
    }


def _historical_window_bounds(time_window: object) -> tuple[int | None, int | None]:
    if not isinstance(time_window, str) or not time_window.startswith("historical:absolute;"):
        return None, None
    start = re.search(r"(?:^|;)start=([0-9]+)(?:;|$)", time_window)
    end = re.search(r"(?:^|;)end=([0-9]+)(?:;|$)", time_window)
    if not start or not end:
        raise ValueError("clinic_metrics_batch_historical_window_invalid")
    return int(start.group(1)), int(end.group(1))


def _response_error(response: object) -> str | None:
    if not isinstance(response, dict):
        return "clinic_metrics_response_invalid"
    error = response.get("error")
    if error:
        return str(error)[:500]
    if str(response.get("status", "")).lower() == "error":
        return str(response.get("errorType") or response.get("message") or "provider_error")[:500]
    return None


def _cross_source_metrics_rejected(response: object) -> bool:
    if not isinstance(response, (dict, list)):
        return False
    encoded = json.dumps(response, sort_keys=True).lower()
    return "cross_source_metrics_rejected" in encoded


def _split_top_level_or(query: str) -> list[str]:
    parts = []
    start = 0
    paren = brace = bracket = 0
    quote = None
    escaped = False
    index = 0
    while index < len(query):
        char = query[index]
        if quote:
            if escaped:
                escaped = False
            elif char == "\\":
                escaped = True
            elif char == quote:
                quote = None
            index += 1
            continue
        if char in {'"', "'", "`"}:
            quote = char
        elif char == "(":
            paren += 1
        elif char == ")":
            paren = max(0, paren - 1)
        elif char == "{":
            brace += 1
        elif char == "}":
            brace = max(0, brace - 1)
        elif char == "[":
            bracket += 1
        elif char == "]":
            bracket = max(0, bracket - 1)
        elif paren == brace == bracket == 0 and query[index:index + 2].lower() == "or":
            before = query[index - 1] if index else " "
            after = query[index + 2] if index + 2 < len(query) else " "
            if before.isspace() and after.isspace():
                part = query[start:index].strip()
                if part:
                    parts.append(part)
                start = index + 2
                index += 2
                continue
        index += 1
    tail = query[start:].strip()
    if tail:
        parts.append(tail)
    return parts if len(parts) > 1 else []


def _metric_values(series: dict) -> list[float]:
    raw_values = series.get("values")
    if raw_values is None and series.get("value") is not None:
        raw_values = [series["value"]]
    values = []
    for item in raw_values or []:
        if not isinstance(item, (list, tuple)) or len(item) < 2:
            continue
        try:
            value = float(item[1])
        except (TypeError, ValueError):
            continue
        if math.isfinite(value):
            values.append(value)
    return values


def _summarize_metrics_response(response: object) -> dict:
    error = _response_error(response)
    if error:
        code = "cross_source_metrics_rejected" if "cross_source_metrics_rejected" in error.lower() \
            else "clinic_metrics_query_failed"
        return _failed_metrics_summary(error, code)
    data = response.get("data", {}) if isinstance(response, dict) else {}
    result = data.get("result", []) if isinstance(data, dict) else []
    if not isinstance(result, list):
        return _failed_metrics_summary("Clinic metrics result is not an array.")
    summaries = []
    all_values = []
    sample_count = 0
    for series in result[:50]:
        if not isinstance(series, dict):
            continue
        values = _metric_values(series)
        if not values:
            continue
        labels = series.get("metric", {})
        safe_labels = {
            str(key): str(value)[:160]
            for key, value in labels.items()
            if key in _METRIC_LABEL_KEYS
        } if isinstance(labels, dict) else {}
        sample_count += len(values)
        all_values.extend(values)
        summaries.append({
            "labels": safe_labels,
            "samples": len(values),
            "minimum": min(values),
            "maximum": max(values),
            "average": sum(values) / len(values),
            "latest": values[-1],
        })
    return {
        "seriesCount": len(summaries),
        "sampleCount": sample_count,
        "minimum": min(all_values) if all_values else None,
        "maximum": max(all_values) if all_values else None,
        "average": sum(all_values) / len(all_values) if all_values else None,
        "series": summaries,
        "errorCode": None,
        "errorMessage": None,
    }


def _failed_metrics_summary(message: str, code: str = "clinic_metrics_query_failed") -> dict:
    return {
        "seriesCount": 0,
        "sampleCount": 0,
        "minimum": None,
        "maximum": None,
        "average": None,
        "series": [],
        "errorCode": code,
        "errorMessage": str(message)[:500],
    }


def _combine_metrics_responses(responses: list[object]) -> dict:
    errors = [error for response in responses if (error := _response_error(response))]
    results = []
    for response in responses:
        if not isinstance(response, dict):
            continue
        data = response.get("data", {})
        if isinstance(data, dict) and isinstance(data.get("result"), list):
            results.extend(data["result"])
    if errors:
        return {"error": "; ".join(errors)[:500]}
    return {"status": "success", "data": {"resultType": "matrix", "result": results}}


def _public_recovery(recovery: dict | None) -> dict | None:
    if recovery is None:
        return None
    return {key: recovery[key] for key in (
        "triggered", "code", "method", "succeeded", "parts"
    )}


def _metrics_batch_statement(
    query_id: str,
    description: str | None,
    status: str,
    summary: dict,
    recovery: dict | None,
) -> str:
    subject = description or f"Metric query {query_id}"
    if status == "failed":
        return f"{subject} failed without usable metric evidence."
    if status == "empty":
        return f"{subject} returned no numeric metric series in the requested window."
    recovery_text = " after automatic source-homogeneous recovery" if recovery else ""
    return (
        f"{subject} returned {summary['seriesCount']} series and "
        f"{summary['sampleCount']} numeric samples{recovery_text}; "
        f"observed range was {summary['minimum']:.6g} to {summary['maximum']:.6g}."
    )[:1000]


def _utc_timestamp(value: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(value))


# ── 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_range_batch [context_json] <plan_json> <output_json>")
        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("  ngm_slow_query_list <cluster_id> <begin_unix_sec> <end_unix_sec> [limit] [orderBy] [desc] [fields] [text] [digest] [show_internal]")
        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_batch":
        if len(sys.argv) not in {4, 5}:
            raise SystemExit(
                "usage: clinic_api.py query_range_batch [context_json] <plan_json> <output_json>"
            )
        context_path = Path(sys.argv[2]) if len(sys.argv) == 5 else None
        plan_path = Path(sys.argv[3] if context_path else sys.argv[2])
        output_path = Path(sys.argv[4] if context_path else sys.argv[3])
        context = json.loads(context_path.read_text()) if context_path else None
        result = api.query_range_batch(json.loads(plan_path.read_text()), context)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        temporary_path = output_path.with_suffix(output_path.suffix + ".tmp")
        temporary_path.write_text(json.dumps(result, separators=(",", ":")))
        os.replace(temporary_path, output_path)
    elif 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 == "ngm_slow_query_list":
        result = api.ngm_slow_query_list(
            sys.argv[2],
            sys.argv[3],
            sys.argv[4],
            int(sys.argv[5]) if len(sys.argv) > 5 else 100,
            sys.argv[6] if len(sys.argv) > 6 else "query_time",
            sys.argv[7].lower() == "true" if len(sys.argv) > 7 else True,
            sys.argv[8] if len(sys.argv) > 8 else "query,timestamp,query_time,memory_max,digest,index_names",
            sys.argv[9] if len(sys.argv) > 9 else "",
            sys.argv[10] if len(sys.argv) > 10 else "",
            sys.argv[11].lower() == "true" if len(sys.argv) > 11 else True,
        )
    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()