from __future__ import annotations

import copy
import os
from pathlib import Path
from typing import Any

from niannian.core.config import (
    _load_yaml,
    load_providers_config,
    resolve_providers_config_candidates,
    resolve_providers_config_path,
)
from niannian.llm.model_registry import (
    add_provider_model,
    init_free_models,
    list_provider_model_entries,
    model_ids_for_provider,
    remove_provider_model,
)
from niannian.llm.custom_providers import (
    CustomProviderError,
    create_custom_provider,
    delete_custom_provider,
    is_custom_provider_entry,
)
from niannian.llm.provider_catalog import catalog_by_id, list_catalog
from niannian.llm.providers.registry import is_storage_provider

_LLM_SEED_KEYS = frozenset(
    {
        "openai_compat",
        "siliconflow",
        "deepseek",
        "moonshot",
        "dashscope",
        "zhipu",
        "minimax",
        "baichuan",
        "volcengine",
        "qianfan",
        "spark",
        "anthropic",
    }
)


class LlmConfigError(Exception):
    pass


def _mask_key(key: str) -> str:
    key = (key or "").strip()
    if len(key) <= 8:
        return "***" if key else ""
    return key[:4] + "***" + key[-2:]


def _resolve_raw_api_key(doc: dict[str, Any], name: str) -> str:
    entry = doc.get(name)
    if not isinstance(entry, dict):
        return ""
    raw = entry.get("api_key")
    if not isinstance(raw, str):
        return ""
    raw = raw.strip()
    if raw.startswith("${") and raw.endswith("}"):
        return (os.environ.get(raw[2:-1]) or "").strip()
    return raw


def _provider_api_key(doc: dict[str, Any], resolved: dict[str, Any], provider_id: str) -> str:
    item = catalog_by_id().get(provider_id) or {}
    names = [provider_id, *(item.get("aliases") or [])]
    for name in names:
        key = ""
        res = resolved.get(name)
        if isinstance(res, dict) and res.get("api_key"):
            key = str(res["api_key"]).strip()
        if not key:
            key = _resolve_raw_api_key(doc, name)
        if key:
            return key
    if provider_id == "siliconflow":
        return (os.environ.get("SILICONFLOW_API_KEY") or "").strip()
    if provider_id == "zhipu":
        return (os.environ.get("ZHIPU_API_KEY") or "").strip()
    return ""


def sync_keys_from_file() -> dict[str, Any]:
    """从 providers.local / 环境变量同步硅基与智谱 Key。"""
    doc = _load_raw_providers_doc()
    synced: dict[str, str] = {}
    dirty = False

    sf_key = (
        _resolve_raw_api_key(doc, "openai_compat")
        or _resolve_raw_api_key(doc, "siliconflow")
        or (os.environ.get("SILICONFLOW_API_KEY") or "").strip()
    )
    if sf_key:
        sf_entry = doc.get("openai_compat") if isinstance(doc.get("openai_compat"), dict) else {}
        base_url = str((sf_entry or {}).get("base_url") or "https://api.siliconflow.cn/v1").rstrip("/")
        for pid in ("siliconflow", "openai_compat"):
            entry = doc.setdefault(pid, {})
            if isinstance(entry, dict):
                if entry.get("api_key") != sf_key or entry.get("base_url") != base_url:
                    dirty = True
                entry["api_key"] = sf_key
                entry["base_url"] = base_url
        synced["siliconflow"] = _mask_key(sf_key)

    zp_key = _resolve_raw_api_key(doc, "zhipu") or (os.environ.get("ZHIPU_API_KEY") or "").strip()
    if zp_key:
        zp_base = "https://open.bigmodel.cn/api/paas/v4"
        entry = doc.setdefault("zhipu", {})
        if isinstance(entry, dict):
            if entry.get("api_key") != zp_key or entry.get("base_url", zp_base) != zp_base:
                dirty = True
            entry["api_key"] = zp_key
            entry.setdefault("base_url", zp_base)
        synced["zhipu"] = _mask_key(zp_key)

    path = str(resolve_providers_config_path(for_write=True))
    if dirty:
        path = _save_raw_providers_doc(doc)
    return {"providers_path": path, "synced": synced}


def _seed_llm_providers() -> dict[str, Any]:
    """从 example 或目录默认生成国内厂商空 Key 模板。"""
    for p in resolve_providers_config_candidates():
        if p.name == "providers.example.yaml" and p.is_file():
            raw = _load_yaml(p)
            return {k: copy.deepcopy(v) for k, v in raw.items() if k in _LLM_SEED_KEYS}
    out: dict[str, Any] = {}
    for item in list_catalog():
        pid = str(item["id"])
        out[pid] = {"api_key": "", "base_url": item["base_url"]}
        for alias in item.get("aliases") or []:
            out[str(alias)] = {"api_key": "", "base_url": item["base_url"]}
    return out


def _load_raw_providers_doc() -> dict[str, Any]:
    path = resolve_providers_config_path(for_write=True)
    if path.is_file():
        return _load_yaml(path)
    return _seed_llm_providers()


def _save_raw_providers_doc(doc: dict[str, Any]) -> str:
    from niannian.core.config import save_providers_config

    path = save_providers_config(doc)
    return str(path)


def list_provider_views() -> tuple[dict[str, Any], str]:
    doc = _load_raw_providers_doc()
    resolved = load_providers_config()
    cat = catalog_by_id(doc)
    views: dict[str, Any] = {}
    for item in list_catalog(doc):
        pid = str(item["id"])
        raw = doc.get(pid) if isinstance(doc.get(pid), dict) else {}
        key = _provider_api_key(doc, resolved, pid)
        model_entries = list_provider_model_entries(pid)
        base_url = str(raw.get("base_url") or item.get("base_url") or "").rstrip("/")
        has_key = bool(key.strip())
        is_custom = bool(item.get("custom_provider")) or bool(item.get("builtin_custom"))
        configured = has_key and bool(base_url) if is_custom else has_key
        views[pid] = {
            "id": pid,
            "label": str(raw.get("label") or item.get("label") or pid),
            "base_url": base_url,
            "api_key_masked": _mask_key(key),
            "configured": configured,
            "models": model_ids_for_provider(pid) or list(item.get("models") or []),
            "model_entries": model_entries,
            "docs_url": item.get("docs_url"),
            "exclusive": bool(item.get("exclusive")),
            "custom_provider": bool(item.get("custom_provider")),
            "builtin_custom": bool(item.get("builtin_custom")),
            "deletable": bool(item.get("custom_provider")),
        }
    path = str(resolve_providers_config_path(for_write=True))
    return views, path


def update_provider(provider_id: str, *, api_key: str | None = None, base_url: str | None = None) -> dict[str, Any]:
    doc = _load_raw_providers_doc()
    cat = catalog_by_id(doc)
    if provider_id not in cat:
        raise LlmConfigError(f"unknown provider: {provider_id}")
    entry = doc.setdefault(provider_id, {})
    if not isinstance(entry, dict):
        entry = {}
        doc[provider_id] = entry
    if is_custom_provider_entry(entry) or cat[provider_id].get("custom_provider"):
        entry.setdefault("custom_provider", True)
    if base_url is not None:
        entry["base_url"] = base_url.strip().rstrip("/")
    elif not entry.get("base_url"):
        entry["base_url"] = cat[provider_id].get("base_url") or ""
    if api_key is not None:
        key = api_key.strip()
        if key:
            entry["api_key"] = key
    # 硅基别名同步 Key
    item = cat[provider_id]
    if provider_id == "siliconflow" and api_key is not None:
        for alias in item.get("aliases") or []:
            alias_entry = doc.setdefault(str(alias), {})
            if isinstance(alias_entry, dict):
                if api_key.strip():
                    alias_entry["api_key"] = api_key.strip()
                if entry.get("base_url"):
                    alias_entry["base_url"] = entry["base_url"]
    _save_raw_providers_doc(doc)
    views, _ = list_provider_views()
    return views[provider_id]


def init_default_providers() -> str:
    doc = _load_raw_providers_doc()
    seed = _seed_llm_providers()
    for k, v in seed.items():
        if k not in doc and not is_storage_provider(k):
            doc[k] = copy.deepcopy(v)
    return _save_raw_providers_doc(doc)


def ensure_default_free_models() -> None:
    """首次启动时写入硅基/智谱免费模型列表。"""
    from niannian.llm.model_registry import load_model_registry, resolve_llm_models_path

    path = resolve_llm_models_path()
    doc = load_model_registry()
    pm = doc.get("provider_models") or {}
    missing = [pid for pid in ("siliconflow", "zhipu") if not pm.get(pid)]
    if not missing and path.is_file():
        return
    if not path.is_file():
        template = Path(__file__).resolve().parents[2] / "config" / "llm_models.local.yaml.template"
        if template.is_file():
            import shutil

            dest = resolve_llm_models_path(for_write=True)
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(template, dest)
            return
    if missing:
        init_free_models(*missing)


def bootstrap_llm_config_from_files() -> None:
    """从 providers.local / llm_models.local 自动归一化 Key 与默认模型（启动与读配置时调用）。"""
    sync_keys_from_file()
    ensure_default_free_models()


def sync_all_from_file() -> dict[str, Any]:
    """同步 Key + 初始化硅基/智谱免费模型列表。"""
    key_result = sync_keys_from_file()
    ensure_default_free_models()
    from niannian.llm.model_registry import load_model_registry

    pm = (load_model_registry().get("provider_models") or {})
    key_result["models_initialized"] = {k: list(v) for k, v in pm.items() if k in ("siliconflow", "zhipu")}
    return key_result


def add_model_for_provider(provider_id: str, model_id: str) -> list[dict[str, Any]]:
    from niannian.llm.model_registry import ModelRegistryError

    doc = _load_raw_providers_doc()
    if provider_id not in catalog_by_id(doc):
        raise LlmConfigError(f"unknown provider: {provider_id}")
    try:
        return add_provider_model(provider_id, model_id)
    except ModelRegistryError as exc:
        raise LlmConfigError(str(exc)) from exc


def delete_model_for_provider(provider_id: str, model_id: str) -> list[dict[str, Any]]:
    from niannian.llm.model_registry import ModelRegistryError

    doc = _load_raw_providers_doc()
    if provider_id not in catalog_by_id(doc):
        raise LlmConfigError(f"unknown provider: {provider_id}")
    try:
        return remove_provider_model(provider_id, model_id)
    except ModelRegistryError as exc:
        raise LlmConfigError(str(exc)) from exc


def add_custom_llm_provider(
    *,
    provider_id: str,
    label: str,
    base_url: str,
    api_key: str = "",
) -> dict[str, Any]:
    try:
        create_custom_provider(
            provider_id=provider_id,
            label=label,
            base_url=base_url,
            api_key=api_key,
        )
    except CustomProviderError as exc:
        raise LlmConfigError(str(exc)) from exc
    views, _ = list_provider_views()
    return views[provider_id.strip().lower()]


def remove_custom_llm_provider(provider_id: str) -> None:
    try:
        delete_custom_provider(provider_id)
    except CustomProviderError as exc:
        raise LlmConfigError(str(exc)) from exc


BRAIN_MODULE_IDS = ("right", "left", "hippocampus", "cerebellum", "retrieval_rerank", "ocr", "brainstem")


def sync_brain_api_into_config(config: dict[str, Any], config_path: Path | None = None) -> None:
    """从 default.yaml + api.local.yaml 重新合并脑区路由到运行中 config。"""
    from niannian.core.config import load_config

    fresh = load_config(config_path)
    fresh_api = fresh.get("api") if isinstance(fresh.get("api"), dict) else {}
    api = config.setdefault("api", {})
    if not isinstance(api, dict):
        api = {}
        config["api"] = api
    for key in ("global_fallback", "provider_limits"):
        if key in fresh_api:
            api[key] = copy.deepcopy(fresh_api[key])
    for mid in BRAIN_MODULE_IDS:
        if mid in fresh_api:
            api[mid] = copy.deepcopy(fresh_api[mid])


def save_brain_api_from_config(api_cfg: dict[str, Any]) -> str:
    """将脑区路由持久化到 api.local.yaml（仅写入非空 slot）。"""
    from niannian.core.config import save_api_local_config

    payload: dict[str, Any] = {}
    gf = api_cfg.get("global_fallback")
    if gf not in ("", None):
        payload["global_fallback"] = gf
    provider_limits = api_cfg.get("provider_limits")
    if isinstance(provider_limits, dict) and provider_limits:
        payload["provider_limits"] = copy.deepcopy(provider_limits)
    for mid in BRAIN_MODULE_IDS:
        mod = api_cfg.get(mid)
        if not isinstance(mod, dict):
            continue
        slots = {
            slot: mod.get(slot)
            for slot in ("primary", "module_fallback", "global_fallback")
            if mod.get(slot) not in ("", None)
        }
        if slots:
            payload[mid] = slots
    if not payload:
        from niannian.core.config import resolve_api_local_config_path

        path = resolve_api_local_config_path(for_write=True)
        if path.is_file():
            path.unlink()
        return str(path)
    return str(save_api_local_config(payload))


def persist_brain_api_patch(patch: dict[str, Any]) -> str:
    """按 PATCH 增量更新 api.local.yaml；null/空值表示删除覆盖、回退 default.yaml。"""
    from niannian.core.config import _load_yaml, resolve_api_local_config_path, save_api_local_config

    path = resolve_api_local_config_path(for_write=True)
    overlay: dict[str, Any] = _load_yaml(path) if path.is_file() else {}

    if "global_fallback" in patch:
        gf = patch["global_fallback"]
        if gf in ("", None):
            overlay.pop("global_fallback", None)
        else:
            overlay["global_fallback"] = gf

    brains_patch = patch.get("brains")
    if isinstance(brains_patch, dict):
        for mid, slots in brains_patch.items():
            if not isinstance(slots, dict):
                continue
            mod = overlay.setdefault(mid, {})
            if not isinstance(mod, dict):
                mod = {}
                overlay[mid] = mod
            for slot in ("primary", "module_fallback", "global_fallback"):
                if slot not in slots:
                    continue
                val = slots[slot]
                if val in ("", None):
                    mod.pop(slot, None)
                else:
                    mod[slot] = val
            if not mod:
                overlay.pop(mid, None)

    overlay = {k: v for k, v in overlay.items() if v not in ({}, None, "")}
    if not overlay:
        if path.is_file():
            path.unlink()
        return str(path)
    return str(save_api_local_config(overlay))


def extract_brain_api(api_cfg: dict[str, Any]) -> dict[str, Any]:
    brains: dict[str, Any] = {}
    for mid in BRAIN_MODULE_IDS:
        mod = api_cfg.get(mid)
        if not isinstance(mod, dict):
            mod = {}
        brains[mid] = {slot: mod.get(slot) for slot in ("primary", "module_fallback", "global_fallback")}
    return {
        "brains": brains,
        "global_fallback": api_cfg.get("global_fallback"),
        "provider_limits": api_cfg.get("provider_limits") or {},
    }


def apply_brain_patch(api_cfg: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]:
    out = copy.deepcopy(api_cfg)
    brains_patch = patch.get("brains")
    if isinstance(brains_patch, dict):
        for mid, slots in brains_patch.items():
            if not isinstance(slots, dict):
                continue
            mod = out.setdefault(mid, {})
            if not isinstance(mod, dict):
                mod = {}
                out[mid] = mod
            for slot in ("primary", "module_fallback", "global_fallback"):
                if slot in slots:
                    val = slots[slot]
                    mod[slot] = val if val not in ("", None) else None
    if "global_fallback" in patch:
        gf = patch["global_fallback"]
        out["global_fallback"] = gf if gf not in ("", None) else None
    return out
