"""CRUD-операции над кластерами kind и сводная статистика.

Автор: Сергей Антропов
Сайт: https://devops.org.ru
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import shutil
import tempfile
from functools import partial
from pathlib import Path
from typing import Any

from fastapi import APIRouter, BackgroundTasks, HTTPException, Query
from fastapi.responses import FileResponse, JSONResponse
from starlette.background import BackgroundTask

from core.cluster_config_edit import (
    ClusterConfigEditError,
    NOTE_NOT_IN_KIND,
    NOTE_REGISTERED_IN_KIND,
    apply_cluster_config_update,
    read_cluster_config_bundle,
)
from core.cluster_lifecycle import (
    KindClusterError,
    cluster_summary_for_api,
    configured_container_cli,
    create_cluster_non_interactive,
    delete_kind_cluster_and_data,
    delete_kind_cluster_keep_data,
    kubectl_delete_pod,
    kubectl_get_json,
    kubectl_get_all_namespaces_wide,
    kubectl_nodes_wide,
    kubectl_pods_all_namespaces,
    list_registered_kind_clusters,
    read_meta_json,
    start_kind_cluster_containers,
    stop_kind_cluster_containers,
    validate_cluster_name,
)
from core.container_resource_stats import (
    aggregate_kind_cluster_resources,
    collect_kind_clusters_resource_stats,
    collect_single_cluster_resource_stats,
    running_kind_clusters_mask,
)
from core.job_store import (
    JobRecord,
    end_job_tracking,
    get_logs_snapshot_sync,
    get_progress_sync,
    job_store,
    register_uncapped_job_log,
    request_cancel_sync,
    take_uncapped_log_finalize,
)
from core.provision_log import provision_log_file_path, write_cluster_provision_log
from core.kind_guard import kind_cluster_lock
from kind_k8s_paths import clusters_dir
from kubeconfig_patch import kubeconfig_host_file, patch_kubeconfig_server_for_host
from models.schemas import (
    AggregateResourcesSummary,
    ClusterConfigGetResponse,
    ClusterConfigUpdateRequest,
    ClusterConfigUpdateResponse,
    ClusterCreateAccepted,
    ClusterCreateRequest,
    ClusterOverviewResponse,
    ClusterSummary,
    ClusterWorkloadsResponse,
    JobView,
    K8sListJsonBlock,
    PodRestartRequest,
    PodRestartResponse,
    StatsResponse,
)

logger = logging.getLogger("kind_k8s.api.clusters")

# Сообщение UI: узлы kind остановлены — kubectl до API кластера не достучится (DNS control-plane и т.д.).
_K8S_CLUSTER_STOPPED_MSG = (
    "Кластер остановлен. Данные Kubernetes появятся после запуска узлов (кнопка «Старт» в панели)."
)

# Безопасные имена namespace/pod для kubectl (без shell-метасимволов).
_K8S_NAME_SAFE = re.compile(r"^[a-zA-Z0-9][-a-zA-Z0-9.]*[a-zA-Z0-9]$|^[a-zA-Z0-9]$")


def _k8s_ns_pod_valid(s: str) -> bool:
    if not s or len(s) > 253:
        return False
    return bool(_K8S_NAME_SAFE.match(s))


def _k8s_list_from_kubectl(
    rc: int,
    items: list[dict[str, Any]] | None,
    err: str,
) -> K8sListJsonBlock:
    """Собрать блок JSON для UI из результата :func:`kubectl_get_json`."""
    if items is not None and rc == 0:
        return K8sListJsonBlock(ok=True, rc=0, items=items, message=None)
    return K8sListJsonBlock(ok=False, rc=rc, items=[], message=err or "ошибка kubectl")


def _unlink_temp_kubeconfig(path: Path) -> None:
    """Удалить временный файл после отдачи ответа (скачивание kubeconfig)."""
    try:
        path.unlink(missing_ok=True)
    except OSError as e:
        logger.debug("Не удалось удалить временный kubeconfig %s: %s", path, e)

router = APIRouter(tags=["clusters"])


def _api_job_log_limit_detail() -> int:
    """Сколько строк журнала отдавать в GET /jobs/{id} (полный хвост буфера задания)."""
    raw = (os.environ.get("KIND_K8S_JOB_API_LOG_MAX_LINES") or "5000").strip()
    try:
        return max(200, min(int(raw), 20000))
    except ValueError:
        return 5000


def _record_to_job_view(rec: JobRecord, *, include_progress_log: bool = True) -> JobView:
    """JobRecord → JobView с полями прогресса из потокобезопасного снимка."""
    prog = get_progress_sync(rec.job_id)
    stage, pct = (None, None)
    if prog is not None:
        stage, pct = prog[0], prog[1]
    if not include_progress_log:
        log_tail: list[str] = []
    else:
        if rec.status in ("queued", "running"):
            log_tail = get_logs_snapshot_sync(rec.job_id)
        else:
            log_tail = list(rec.log_lines or [])
        cap = _api_job_log_limit_detail()
        if len(log_tail) > cap:
            log_tail = log_tail[-cap:]
    return JobView(
        job_id=rec.job_id,
        kind=rec.kind,
        status=rec.status,
        cluster_name=rec.cluster_name,
        created_at_utc=rec.created_at_utc,
        message=rec.message,
        result=rec.result,
        progress_stage=stage,
        progress_percent=pct,
        progress_log=log_tail,
    )


def _stats_sync() -> StatsResponse:
    """Собрать статистику (синхронно; вызывать из thread при необходимости)."""
    kind_names = list_registered_kind_clusters()
    cdir = clusters_dir()
    subdirs: list[str] = []
    if cdir.is_dir():
        subdirs = sorted(p.name for p in cdir.iterdir() if p.is_dir() and not p.name.startswith("."))

    total_workers = 0
    for name in subdirs:
        meta = read_meta_json(name)
        if not meta:
            continue
        w = meta.get("worker_nodes")
        if w is None:
            continue
        try:
            total_workers += int(w)
        except (TypeError, ValueError):
            continue

    jobs = job_store.snapshot_all()
    failed = sum(1 for j in jobs if j.status == "failed")

    res_blocks, res_err = collect_kind_clusters_resource_stats()
    agg: AggregateResourcesSummary
    if res_err:
        agg = AggregateResourcesSummary()
    else:
        agg = aggregate_kind_cluster_resources(res_blocks)

    return StatsResponse(
        container_cli=configured_container_cli(),
        kind_clusters_count=len(kind_names),
        local_cluster_dirs_count=len(subdirs),
        total_workers_from_meta=total_workers,
        jobs_total=len(jobs),
        jobs_recent_failed=failed,
        cluster_resources=res_blocks,
        cluster_resources_error=res_err,
        aggregate_cluster_resources=agg,
    )


@router.get("/stats", response_model=StatsResponse, summary="Статистика")
async def get_stats() -> StatsResponse:
    """Число кластеров kind, локальных каталогов, сумма workers из meta (если есть), счётчики заданий."""
    return await asyncio.to_thread(_stats_sync)


@router.get("/jobs", response_model=list[JobView], summary="Список заданий")
async def list_jobs(limit: int = Query(30, ge=1, le=200, description="Сколько последних заданий")) -> list[JobView]:
    """История создания кластеров (в памяти процесса; после перезапуска контейнера пусто)."""
    items = job_store.snapshot_recent_sorted(limit=limit)
    # В списке не тащим progress_log — экономим трафик; полный журнал только в GET /jobs/{id}.
    return [_record_to_job_view(r, include_progress_log=False) for r in items]


@router.delete("/jobs", summary="Очистить завершённые задания из памяти")
async def delete_completed_jobs() -> dict[str, int]:
    """
    Удаляет записи со статусом success, failed, cancelled. Задания в очереди и в работе не трогаются.
    """
    removed = await job_store.purge_completed()
    logger.info("Очистка заданий: удалено завершённых записей: %s", removed)
    return {"removed": removed}


@router.get("/clusters", response_model=list[ClusterSummary], summary="Список кластеров")
async def list_clusters() -> list[ClusterSummary]:
    """Объединение: зарегистрированные в kind + каталоги в ``clusters/`` (без дубликатов в выдаче)."""
    kind_names = set(list_registered_kind_clusters())
    cdir = clusters_dir()
    dir_names: set[str] = set()
    if cdir.is_dir():
        dir_names = {p.name for p in cdir.iterdir() if p.is_dir() and not p.name.startswith(".")}
    all_names = sorted(kind_names | dir_names)
    running_map = running_kind_clusters_mask(all_names)
    out: list[ClusterSummary] = []
    for name in all_names:
        summary = cluster_summary_for_api(name)
        out.append(
            ClusterSummary(
                name=str(summary["name"]),
                registered_in_kind=bool(summary["registered_in_kind"]),
                kind_nodes_running=bool(running_map.get(name, False)),
                has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
                has_provision_log=bool(summary.get("has_provision_log")),
                meta=dict(summary["meta"]) if isinstance(summary.get("meta"), dict) else {},
            )
        )
    logger.debug("list_clusters: %s записей", len(out))
    return out


def _cluster_summary_model(name: str) -> ClusterSummary:
    """Сводка кластера для ответов API (как в GET /clusters)."""
    summary = cluster_summary_for_api(name)
    running_map = running_kind_clusters_mask([name])
    return ClusterSummary(
        name=str(summary["name"]),
        registered_in_kind=bool(summary["registered_in_kind"]),
        kind_nodes_running=bool(running_map.get(name, False)),
        has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
        has_provision_log=bool(summary.get("has_provision_log")),
        meta=dict(summary["meta"]) if isinstance(summary.get("meta"), dict) else {},
    )


@router.get(
    "/clusters/{name}/config",
    response_model=ClusterConfigGetResponse,
    summary="Конфигурация кластера (meta + kind-config.yaml)",
    responses={404: {"description": "Каталог кластера не найден"}},
)
async def get_cluster_config(name: str) -> ClusterConfigGetResponse:
    """
    Текущие ``meta.json`` и текст ``kind-config.yaml`` из ``clusters/<имя>/``.
    Поле ``kind_note`` напоминает: у уже созданного в kind кластера смена YAML на диске не меняет узлы до пересоздания.
    """
    n = name.strip()
    if not validate_cluster_name(n):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    try:
        bundle = await asyncio.to_thread(read_cluster_config_bundle, n)
    except ClusterConfigEditError as e:
        if e.not_found:
            raise HTTPException(status_code=404, detail=str(e)) from e
        raise HTTPException(status_code=400, detail=str(e)) from e
    summary = await asyncio.to_thread(cluster_summary_for_api, n)
    reg = bool(summary["registered_in_kind"])
    note = NOTE_REGISTERED_IN_KIND if reg else NOTE_NOT_IN_KIND
    return ClusterConfigGetResponse(
        cluster_name=str(bundle["cluster_name"]),
        meta=dict(bundle["meta"]) if isinstance(bundle.get("meta"), dict) else {},
        kind_config_yaml=bundle.get("kind_config_yaml"),
        has_kind_config=bool(bundle.get("has_kind_config")),
        registered_in_kind=reg,
        kind_note=note,
    )


@router.put(
    "/clusters/{name}/config",
    response_model=ClusterConfigUpdateResponse,
    summary="Обновить конфигурацию кластера на диске",
    responses={404: {"description": "Каталог кластера не найден"}},
)
async def put_cluster_config(name: str, body: ClusterConfigUpdateRequest) -> ClusterConfigUpdateResponse:
    """
    Сохраняет изменения в ``kind-config.yaml`` и/или ``meta.json``.

    Режимы: (1) непустой ``kind_config_yaml`` — полная замена файла с проверкой структуры kind Cluster;
    (2) ``kubernetes_version`` и/или ``workers`` — пересборка простого конфига (как при создании);
    (3) ``description`` — заметка в meta (пустая строка сбрасывает поле).

    Kind не применяет новый YAML к уже работающим нодам — см. ``kind_note`` в ответе.
    """
    n = name.strip()
    if not validate_cluster_name(n):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    try:
        updated_yaml, msg = await asyncio.to_thread(
            apply_cluster_config_update,
            n,
            kubernetes_version=body.kubernetes_version,
            workers=body.workers,
            kind_config_yaml=body.kind_config_yaml,
            description=body.description,
        )
    except ClusterConfigEditError as e:
        if e.not_found:
            raise HTTPException(status_code=404, detail=str(e)) from e
        raise HTTPException(status_code=400, detail=str(e)) from e
    summary_api = await asyncio.to_thread(cluster_summary_for_api, n)
    reg = bool(summary_api["registered_in_kind"])
    note = NOTE_REGISTERED_IN_KIND if reg else NOTE_NOT_IN_KIND
    summ = await asyncio.to_thread(_cluster_summary_model, n)
    logger.info("Обновлена конфигурация кластера «%s», yaml=%s", n, updated_yaml)
    return ClusterConfigUpdateResponse(
        cluster_name=n,
        updated_kind_config_yaml=updated_yaml,
        message=msg,
        registered_in_kind=reg,
        kind_note=note,
        summary=summ,
    )


@router.get(
    "/clusters/{name}/kubeconfig",
    summary="Скачать kubeconfig",
    responses={404: {"description": "Файл не найден"}},
)
async def download_kubeconfig(name: str) -> FileResponse:
    """
    Kubeconfig для **kubectl на машине пользователя**: копия ``clusters/<имя>/kubeconfig``
    с ``server=https://<KIND_K8S_KUBECONFIG_CLIENT_HOST или localhost>:<порт>`` (порт из ``docker port``).

    Патч выполняется при каждом скачивании, чтобы учитывать переменные окружения и актуальный порт.
    Если патч невозможен — отдаётся сохранённый ``kubeconfig.host`` или сырой ``kubeconfig``.
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    n = name.strip()
    base = clusters_dir() / n
    primary = base / "kubeconfig"
    if not primary.is_file():
        raise HTTPException(status_code=404, detail="kubeconfig не найден")

    fd, tmp_name = tempfile.mkstemp(suffix=".yaml", prefix="kind-k8s-kube-")
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        shutil.copyfile(primary, tmp_path, follow_symlinks=True)
    except OSError as e:
        tmp_path.unlink(missing_ok=True)
        logger.warning("Копирование kubeconfig для скачивания: %s", e)
        raise HTTPException(status_code=500, detail="Не удалось подготовить kubeconfig") from e

    patched = await asyncio.to_thread(
        partial(patch_kubeconfig_server_for_host, cluster_name=n, kube_path=tmp_path),
    )
    if patched:
        logger.info("Отдача kubeconfig для кластера %s (сгенерировано с актуальным server)", n)
        return FileResponse(
            path=str(tmp_path),
            filename=f"kubeconfig-{n}.yaml",
            media_type="application/x-yaml",
            background=BackgroundTask(_unlink_temp_kubeconfig, tmp_path),
        )

    tmp_path.unlink(missing_ok=True)
    host_path = kubeconfig_host_file(base)
    if host_path.is_file():
        logger.info("Отдача kubeconfig для кластера %s (файл %s, патч при скачивании не удался)", n, host_path.name)
        return FileResponse(
            path=host_path,
            filename=f"kubeconfig-{n}.yaml",
            media_type="application/x-yaml",
        )
    logger.info("Отдача kubeconfig для кластера %s (сырой kubeconfig)", n)
    return FileResponse(
        path=primary,
        filename=f"kubeconfig-{n}.yaml",
        media_type="application/x-yaml",
    )


@router.get(
    "/clusters/{name}/provision-log",
    summary="Журнал развёртывания (JSON)",
    responses={404: {"description": "Файл не найден"}},
)
async def get_cluster_provision_log(name: str) -> JSONResponse:
    """Полный журнал последнего создания/старта кластера (``provision_log.json`` в каталоге кластера)."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    path = provision_log_file_path(name)
    if not path.is_file():
        raise HTTPException(status_code=404, detail="Журнал развёртывания не найден")
    try:
        raw = await asyncio.to_thread(path.read_text, encoding="utf-8")
        data = json.loads(raw)
    except (OSError, json.JSONDecodeError) as e:
        logger.warning("provision_log для %s не прочитан: %s", name, e)
        raise HTTPException(status_code=500, detail="Не удалось прочитать журнал") from e
    return JSONResponse(content=data)


@router.get(
    "/clusters/{name}/workloads",
    response_model=ClusterWorkloadsResponse,
    summary="Узлы и поды (kubectl)",
)
async def cluster_workloads(name: str) -> ClusterWorkloadsResponse:
    """``kubectl get nodes`` и ``kubectl get pods -A`` по сохранённому kubeconfig."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    n = name.strip()
    kc = clusters_dir() / n / "kubeconfig"
    if not kc.is_file():
        return ClusterWorkloadsResponse(cluster_name=n, error="Нет сохранённого kubeconfig в clusters/<имя>/")

    summary = cluster_summary_for_api(n)
    run_mask = await asyncio.to_thread(running_kind_clusters_mask, [n])
    if bool(summary["registered_in_kind"]) and not bool(run_mask.get(n, False)):
        return ClusterWorkloadsResponse(cluster_name=n, error=_K8S_CLUSTER_STOPPED_MSG)

    nodes_rc, nodes_out = await asyncio.to_thread(
        kubectl_nodes_wide,
        cluster_name=n,
        kubeconfig=kc,
    )
    pods_rc, pods_out = await asyncio.to_thread(
        kubectl_pods_all_namespaces,
        cluster_name=n,
        kubeconfig=kc,
    )
    return ClusterWorkloadsResponse(
        cluster_name=n,
        nodes_rc=nodes_rc,
        nodes_output=nodes_out,
        pods_rc=pods_rc,
        pods_output=pods_out,
    )


@router.get(
    "/clusters/{name}/overview",
    response_model=ClusterOverviewResponse,
    summary="Сводка кластера (страница UI)",
)
async def cluster_overview(name: str) -> ClusterOverviewResponse:
    """
    Мета, метрики узлов (docker/podman), агрегаты как на главной, вывод kubectl по **всему кластеру**:
    узлы; для namespace-ресурсов везде **-A** (все пространства имён); отдельно namespaces, RS, jobs и т.д.
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    n = name.strip()
    summary = cluster_summary_for_api(n)
    run_mask = await asyncio.to_thread(running_kind_clusters_mask, [n])
    kind_running = bool(run_mask.get(n, False))

    block, res_err = await asyncio.to_thread(collect_single_cluster_resource_stats, n)
    agg = aggregate_kind_cluster_resources([block])

    kc = clusters_dir() / n / "kubeconfig"
    kube_err: str | None = None
    k8s_nodes = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_pods = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_deployments = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_statefulsets = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_daemonsets = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_services = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_ingresses = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_namespaces = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_replicasets = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_jobs = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_cronjobs = K8sListJsonBlock(ok=False, items=[], message=None)
    k8s_pvcs = K8sListJsonBlock(ok=False, items=[], message=None)

    reg_in_kind = bool(summary["registered_in_kind"])

    if not kc.is_file():
        kube_err = "Нет сохранённого kubeconfig в clusters/<имя>/"
        _kube_missing = K8sListJsonBlock(ok=False, items=[], message=kube_err)
        k8s_nodes = _kube_missing
        k8s_pods = k8s_deployments = k8s_statefulsets = k8s_daemonsets = _kube_missing
        k8s_services = k8s_ingresses = k8s_namespaces = k8s_replicasets = _kube_missing
        k8s_jobs = k8s_cronjobs = k8s_pvcs = _kube_missing
    elif reg_in_kind and not kind_running:
        # Не вызываем kubectl: контейнеры узлов остановлены — в логах был бы шум про dev-control-plane / DNS.
        kube_err = _K8S_CLUSTER_STOPPED_MSG
        _stopped = K8sListJsonBlock(ok=False, items=[], message=_K8S_CLUSTER_STOPPED_MSG)
        k8s_nodes = _stopped
        k8s_pods = k8s_deployments = k8s_statefulsets = k8s_daemonsets = _stopped
        k8s_services = k8s_ingresses = k8s_namespaces = k8s_replicasets = _stopped
        k8s_jobs = k8s_cronjobs = k8s_pvcs = _stopped
        logger.debug("overview %s: kubectl пропущен — кластер в kind, узлы не запущены", n)
    else:
        (
            n_res,
            p_res,
            d_res,
            st_res,
            dm_res,
            svc_res,
            ing_res,
            ns_res,
            rs_res,
            job_res,
            cj_res,
            pvc_res,
        ) = await asyncio.gather(
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["nodes"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["pods", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["deployments", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["statefulsets", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["daemonsets", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["services", "-A"]),
            asyncio.to_thread(
                kubectl_get_json,
                cluster_name=n,
                kubeconfig=kc,
                resource_args=["ingresses.networking.k8s.io", "-A"],
            ),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["namespaces"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["replicasets", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["jobs", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["cronjobs", "-A"]),
            asyncio.to_thread(kubectl_get_json, cluster_name=n, kubeconfig=kc, resource_args=["pvc", "-A"]),
        )
        k8s_nodes = _k8s_list_from_kubectl(*n_res)
        k8s_pods = _k8s_list_from_kubectl(*p_res)
        k8s_deployments = _k8s_list_from_kubectl(*d_res)
        k8s_statefulsets = _k8s_list_from_kubectl(*st_res)
        k8s_daemonsets = _k8s_list_from_kubectl(*dm_res)
        k8s_services = _k8s_list_from_kubectl(*svc_res)
        k8s_ingresses = _k8s_list_from_kubectl(*ing_res)
        k8s_namespaces = _k8s_list_from_kubectl(*ns_res)
        k8s_replicasets = _k8s_list_from_kubectl(*rs_res)
        k8s_jobs = _k8s_list_from_kubectl(*job_res)
        k8s_cronjobs = _k8s_list_from_kubectl(*cj_res)
        k8s_pvcs = _k8s_list_from_kubectl(*pvc_res)

    meta = summary.get("meta")
    return ClusterOverviewResponse(
        cluster_name=n,
        registered_in_kind=reg_in_kind,
        kind_nodes_running=kind_running,
        has_local_kubeconfig=bool(summary["has_local_kubeconfig"]),
        has_provision_log=bool(summary.get("has_provision_log")),
        meta=dict(meta) if isinstance(meta, dict) else {},
        resources_error=res_err,
        cluster_resources=block,
        aggregate_resources=agg,
        kubeconfig_error=kube_err,
        nodes_rc=None,
        nodes_output=None,
        pods_rc=None,
        pods_output=None,
        deployments_rc=None,
        deployments_output=None,
        statefulsets_rc=None,
        statefulsets_output=None,
        daemonsets_rc=None,
        daemonsets_output=None,
        services_rc=None,
        services_output=None,
        ingresses_rc=None,
        ingresses_output=None,
        k8s_nodes=k8s_nodes,
        k8s_pods=k8s_pods,
        k8s_deployments=k8s_deployments,
        k8s_statefulsets=k8s_statefulsets,
        k8s_daemonsets=k8s_daemonsets,
        k8s_services=k8s_services,
        k8s_ingresses=k8s_ingresses,
        k8s_namespaces=k8s_namespaces,
        k8s_replicasets=k8s_replicasets,
        k8s_jobs=k8s_jobs,
        k8s_cronjobs=k8s_cronjobs,
        k8s_pvcs=k8s_pvcs,
    )


@router.post(
    "/clusters/{name}/pods/restart",
    response_model=PodRestartResponse,
    summary="Рестарт пода (kubectl delete pod)",
)
async def restart_cluster_pod(name: str, body: PodRestartRequest) -> PodRestartResponse:
    """
    Удаление пода: Deployment/RS/Job при необходимости создают новый Pod (типичный «рестарт»).
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")
    n = name.strip()
    ns = body.namespace.strip()
    pod = body.pod.strip()
    if not _k8s_ns_pod_valid(ns) or not _k8s_ns_pod_valid(pod):
        raise HTTPException(status_code=400, detail="Некорректное имя namespace или пода")
    kc = clusters_dir() / n / "kubeconfig"
    if not kc.is_file():
        raise HTTPException(status_code=404, detail="kubeconfig не найден")
    rc, msg = await asyncio.to_thread(
        kubectl_delete_pod,
        cluster_name=n,
        kubeconfig=kc,
        namespace=ns,
        pod_name=pod,
        grace_period=30,
    )
    if rc != 0:
        logger.warning("restart pod %s/%s в кластере %s: rc=%s %s", ns, pod, n, rc, msg[:500])
        raise HTTPException(status_code=502, detail=msg[:4000] if msg else "kubectl delete pod завершился с ошибкой")
    logger.info("Рестарт пода %s/%s в кластере %s", ns, pod, n)
    return PodRestartResponse(cluster_name=n, namespace=ns, pod=pod, rc=rc, message=msg or "ok")


@router.get("/clusters/{name}", summary="Детали кластера")
async def get_cluster(name: str) -> dict[str, object]:
    """Сводка и попытка ``kubectl get nodes`` (предпочтительно сохранённый kubeconfig)."""
    summary = cluster_summary_for_api(name)
    saved = clusters_dir() / name / "kubeconfig"
    kubectl_rc: int | None = None
    kubectl_msg: str | None = None
    if saved.is_file():
        rc, msg = await asyncio.to_thread(
            kubectl_nodes_wide,
            cluster_name=name.strip(),
            kubeconfig=saved,
        )
        kubectl_rc = rc
        kubectl_msg = msg
    return {
        **summary,
        "kubectl_get_nodes_rc": kubectl_rc,
        "kubectl_get_nodes": kubectl_msg,
    }


async def _run_create_job(job_id: str, body: ClusterCreateRequest) -> None:
    register_uncapped_job_log(job_id)
    cname = body.name.strip()
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id)
            try:
                result = await asyncio.to_thread(
                    create_cluster_non_interactive,
                    name=cname,
                    kubernetes_version_tag=body.kubernetes_version.strip(),
                    workers=body.workers,
                    job_id=job_id,
                )
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("create job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("create job %s: непредвиденная ошибка", job_id)
                return

            payload: dict[str, Any] = {
                "cluster_name": result.cluster_name,
                "kubernetes_version_tag": result.ver_tag,
                "node_image": result.node_image,
                "workers": result.workers,
                "kubeconfig_path": str(result.kubeconfig_path),
                "kubeconfig_patched_for_host": result.kubeconfig_patched_for_host,
                "nodes_ready": result.nodes_ready,
                "nodes_ready_message": result.nodes_ready_message,
            }
            await job_store.set_success(job_id, result=payload, message="Кластер создан")
            logger.info("create job %s: успех, кластер %s", job_id, result.cluster_name)
    finally:
        lines = take_uncapped_log_finalize(job_id)
        rec = await job_store.get(job_id)
        if rec is not None:
            await asyncio.to_thread(
                lambda: write_cluster_provision_log(
                    cluster_name=cname,
                    job_id=job_id,
                    job_kind="create_cluster",
                    status=rec.status,
                    message=rec.message,
                    lines=lines,
                    result=rec.result,
                ),
            )
        end_job_tracking(job_id)


async def _run_start_cluster_job(job_id: str, name: str, kubernetes_version_tag: str, workers: int) -> None:
    """Фоновое создание кластера по уже сохранённому ``kind-config.yaml`` (без kind в списке)."""
    register_uncapped_job_log(job_id)
    n = name.strip()
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id)
            try:
                result = await asyncio.to_thread(
                    create_cluster_non_interactive,
                    name=n,
                    kubernetes_version_tag=kubernetes_version_tag.strip(),
                    workers=workers,
                    job_id=job_id,
                    use_existing_config=True,
                )
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("start_cluster job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("start_cluster job %s: непредвиденная ошибка", job_id)
                return

            payload: dict[str, Any] = {
                "cluster_name": result.cluster_name,
                "kubernetes_version_tag": result.ver_tag,
                "node_image": result.node_image,
                "workers": result.workers,
                "kubeconfig_path": str(result.kubeconfig_path),
                "kubeconfig_patched_for_host": result.kubeconfig_patched_for_host,
                "nodes_ready": result.nodes_ready,
                "nodes_ready_message": result.nodes_ready_message,
            }
            await job_store.set_success(job_id, result=payload, message="Кластер поднят по сохранённому конфигу")
            logger.info("start_cluster job %s: успех, кластер %s", job_id, result.cluster_name)
    finally:
        lines = take_uncapped_log_finalize(job_id)
        rec = await job_store.get(job_id)
        if rec is not None:
            await asyncio.to_thread(
                lambda: write_cluster_provision_log(
                    cluster_name=n,
                    job_id=job_id,
                    job_kind="start_cluster",
                    status=rec.status,
                    message=rec.message,
                    lines=lines,
                    result=rec.result,
                ),
            )
        end_job_tracking(job_id)


async def _run_reapply_kind_config_start_job(
    job_id: str, name: str, kubernetes_version_tag: str, workers: int
) -> None:
    """
    После правки kind-config.yaml: удалить кластер из kind без удаления каталога данных,
    затем ``kind create`` по сохранённому файлу (новый ``meta.json`` без флага reapply).
    """
    register_uncapped_job_log(job_id)
    n = name.strip()
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id)
            try:
                await asyncio.to_thread(delete_kind_cluster_keep_data, name=n, job_id=job_id)
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("reapply start job %s: delete_kind: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("reapply start job %s: ошибка delete", job_id)
                return

            try:
                result = await asyncio.to_thread(
                    create_cluster_non_interactive,
                    name=n,
                    kubernetes_version_tag=kubernetes_version_tag.strip(),
                    workers=workers,
                    job_id=job_id,
                    use_existing_config=True,
                )
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("reapply start job %s: create: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("reapply start job %s: непредвиденная ошибка create", job_id)
                return

            payload: dict[str, Any] = {
                "cluster_name": result.cluster_name,
                "kubernetes_version_tag": result.ver_tag,
                "node_image": result.node_image,
                "workers": result.workers,
                "kubeconfig_path": str(result.kubeconfig_path),
                "kubeconfig_patched_for_host": result.kubeconfig_patched_for_host,
                "nodes_ready": result.nodes_ready,
                "nodes_ready_message": result.nodes_ready_message,
            }
            await job_store.set_success(
                job_id,
                result=payload,
                message="Кластер пересоздан в kind по обновлённому kind-config.yaml",
            )
            logger.info("reapply start job %s: успех, кластер %s", job_id, result.cluster_name)
    finally:
        lines = take_uncapped_log_finalize(job_id)
        rec = await job_store.get(job_id)
        if rec is not None:
            await asyncio.to_thread(
                lambda: write_cluster_provision_log(
                    cluster_name=n,
                    job_id=job_id,
                    job_kind="start_cluster_reapply",
                    status=rec.status,
                    message=rec.message,
                    lines=lines,
                    result=rec.result,
                ),
            )
        end_job_tracking(job_id)


async def _run_stop_cluster_job(job_id: str, name: str) -> None:
    """Фоновая остановка узлов с журналом в GET /jobs/{id}."""
    n = name.strip()
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id, stage="Остановка узлов", percent=6)
            try:
                ok, summary = await asyncio.to_thread(stop_kind_cluster_containers, name=n, job_id=job_id)
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("stop_containers job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("stop_containers job %s: непредвиденная ошибка", job_id)
                return

            await job_store.set_success(
                job_id,
                result={"name": n, "containers_stopped_ok": ok},
                message=summary,
            )
            logger.info("stop_containers job %s: ok=%s", job_id, ok)
    finally:
        end_job_tracking(job_id)


async def _run_start_containers_job(job_id: str, name: str) -> None:
    """Фоновый запуск уже существующих узлов (кластер зарегистрирован в kind)."""
    n = name.strip()
    try:
        async with kind_cluster_lock:
            await job_store.set_running(job_id, stage="Запуск узлов", percent=6)
            try:
                ok, summary = await asyncio.to_thread(start_kind_cluster_containers, name=n, job_id=job_id)
            except KindClusterError as e:
                msg = str(e)
                if "отмен" in msg.lower():
                    await job_store.set_cancelled(job_id, msg)
                else:
                    await job_store.set_failed(job_id, msg)
                logger.warning("start_containers job %s: %s", job_id, e)
                return
            except Exception as e:
                await job_store.set_failed(job_id, f"{type(e).__name__}: {e}")
                logger.exception("start_containers job %s: непредвиденная ошибка", job_id)
                return

            if not ok:
                await job_store.set_failed(job_id, summary)
                return

            await job_store.set_success(
                job_id,
                result={"name": n, "containers_started_ok": ok},
                message=summary,
            )
            logger.info("start_containers job %s: ok=%s", job_id, ok)
    finally:
        end_job_tracking(job_id)


@router.post(
    "/clusters",
    response_model=ClusterCreateAccepted,
    status_code=202,
    summary="Создать кластер (фон)",
)
async def post_create_cluster(
    body: ClusterCreateRequest,
    background_tasks: BackgroundTasks,
) -> ClusterCreateAccepted:
    """Поставить создание кластера в фон; идентификатор задания — в ответе."""
    if not validate_cluster_name(body.name.strip()):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    existing = await asyncio.to_thread(list_registered_kind_clusters)
    if body.name.strip() in existing:
        raise HTTPException(status_code=409, detail="Кластер с таким именем уже есть в kind")

    rec = await job_store.create_job("create_cluster", cluster_name=body.name.strip())
    background_tasks.add_task(_run_create_job, rec.job_id, body)
    logger.info("Принят запрос на создание кластера %s, job_id=%s", body.name, rec.job_id)
    return ClusterCreateAccepted(job_id=rec.job_id)


@router.delete("/clusters/{name}", summary="Удалить кластер")
async def delete_cluster(name: str) -> dict[str, object]:
    """``kind delete`` и удаление локальной папки ``clusters/<имя>/``."""
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    async with kind_cluster_lock:

        def _do() -> tuple[bool, str]:
            return delete_kind_cluster_and_data(name=name, log_to_stdout=False)

        try:
            kind_ok, summary = await asyncio.to_thread(_do)
        except KindClusterError as e:
            raise HTTPException(status_code=500, detail=str(e)) from e

    logger.info("Удаление кластера %s: kind_ok=%s", name, kind_ok)
    return {"name": name, "kind_delete_ok": kind_ok, "summary": summary}


@router.post(
    "/clusters/{name}/stop",
    summary="Остановить узлы кластера (фон, журнал в /jobs)",
    responses={400: {"description": "Некорректное имя"}},
)
async def stop_cluster_nodes(name: str, background_tasks: BackgroundTasks) -> JSONResponse:
    """
    Поставить остановку узлов в фон: тот же журнал в реальном времени, что и при создании.

    Запись кластера в kind сохраняется; «Старт» снова поднимет узлы.
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    n = name.strip()
    rec = await job_store.create_job("stop_containers", cluster_name=n)
    background_tasks.add_task(_run_stop_cluster_job, rec.job_id, n)
    logger.info("Остановка узлов %s в фоне, job_id=%s", n, rec.job_id)
    return JSONResponse(
        status_code=202,
        content={
            "job_id": rec.job_id,
            "status": "queued",
            "mode": "stop",
            "message": "Остановка узлов; опросите GET /api/v1/jobs/{job_id}",
        },
    )


@router.post(
    "/clusters/{name}/start",
    summary="Запустить кластер (контейнеры или kind create по конфигу)",
    responses={400: {"description": "Нет kind и нет kind-config.yaml"}},
)
async def start_cluster_nodes(
    name: str,
    background_tasks: BackgroundTasks,
) -> JSONResponse:
    """
    Если кластер уже зарегистрирован — фоновый запуск сохранённых узлов (журнал в GET /jobs).

    Иначе, при наличии сохранённого конфига в каталоге кластера — фоновый подъём как при создании.
    """
    if not validate_cluster_name(name):
        raise HTTPException(status_code=400, detail="Некорректное имя кластера")

    n = name.strip()

    meta = read_meta_json(n) or {}
    ver_raw = str(meta.get("kubernetes_version_tag") or "v1.29.4").strip() or "v1.29.4"
    w_raw = meta.get("worker_nodes")
    try:
        w = int(w_raw) if w_raw is not None else 0
    except (TypeError, ValueError):
        w = 0

    async with kind_cluster_lock:
        in_kind = n in await asyncio.to_thread(list_registered_kind_clusters)

    if in_kind:
        if bool(meta.get("apply_kind_config_on_next_start")):
            cfg = clusters_dir() / n / "kind-config.yaml"
            if not cfg.is_file():
                raise HTTPException(
                    status_code=400,
                    detail="Нет файла clusters/<имя>/kind-config.yaml — восстановите конфиг или снимите флаг пересоздания в meta.json.",
                )
            rec = await job_store.create_job("start_cluster_reapply", cluster_name=n)
            background_tasks.add_task(_run_reapply_kind_config_start_job, rec.job_id, n, ver_raw, w)
            logger.info(
                "Пересоздание кластера %s в kind по новому конфигу (фон), job_id=%s",
                n,
                rec.job_id,
            )
            return JSONResponse(
                status_code=202,
                content={
                    "job_id": rec.job_id,
                    "status": "queued",
                    "mode": "kind_config_reapply",
                    "message": "Удаление записи в kind и создание кластера по обновлённому kind-config.yaml; GET /api/v1/jobs/{job_id}",
                },
            )
        rec = await job_store.create_job("start_containers", cluster_name=n)
        background_tasks.add_task(_run_start_containers_job, rec.job_id, n)
        logger.info("Запуск контейнеров кластера %s в фоне, job_id=%s", n, rec.job_id)
        return JSONResponse(
            status_code=202,
            content={
                "job_id": rec.job_id,
                "status": "queued",
                "mode": "containers",
                "message": "Запуск узлов; опросите GET /api/v1/jobs/{job_id}",
            },
        )

    cfg = clusters_dir() / n / "kind-config.yaml"
    if not cfg.is_file():
        raise HTTPException(
            status_code=400,
            detail="Кластер не в kind и нет файла clusters/<имя>/kind-config.yaml — создайте кластер или восстановите конфиг.",
        )

    rec = await job_store.create_job("start_cluster", cluster_name=n)
    background_tasks.add_task(_run_start_cluster_job, rec.job_id, n, ver_raw, w)
    logger.info("Фоновый старт кластера %s по конфигу, job_id=%s", n, rec.job_id)
    return JSONResponse(
        status_code=202,
        content={
            "job_id": rec.job_id,
            "status": "queued",
            "mode": "kind_config",
            "message": "Подъём кластера по сохранённому конфигу; опросите GET /api/v1/jobs/{job_id}",
        },
    )


@router.post(
    "/jobs/{job_id}/cancel",
    summary="Прервать фоновое задание",
    responses={400: {"description": "Задание уже завершено"}, 404: {"description": "Нет задания"}},
)
async def cancel_create_job(job_id: str) -> dict[str, object]:
    """
    Прервать активное задание: скачивание образа, создание кластера, запуск/остановка узлов.

    Для длительных команд дочерний процесс завершается принудительно; между узлами останов
    также учитывает флаг отмены.
    """
    rec = await job_store.get(job_id)
    if not rec:
        raise HTTPException(status_code=404, detail="Задание не найдено")
    if rec.status not in ("queued", "running"):
        raise HTTPException(status_code=400, detail="Задание уже завершено; отмена невозможна")
    if not request_cancel_sync(job_id):
        raise HTTPException(status_code=404, detail="Задание не найдено")
    logger.info("Принят запрос отмены задания %s", job_id)
    return {
        "job_id": job_id,
        "cancel_requested": True,
        "message": "Запрошено прерывание; текущая команда будет остановлена, задание перейдёт в отменено",
    }


@router.get("/jobs/{job_id}", response_model=JobView, summary="Статус одного задания")
async def get_job(job_id: str) -> JobView:
    """Узнать состояние фонового создания кластера."""
    rec = await job_store.get(job_id)
    if not rec:
        raise HTTPException(status_code=404, detail="Задание не найдено")
    return _record_to_job_view(rec)