core/services/model_config.py

"""模型配置解析服务：用户 > 管理员 > 环境变量 的优先级（需求 5 / 15.5）。

本模块把"如何为本次会话挑选并构建 LLM 提供商链"这一**纯逻辑**与 UI、存储后端
解耦，便于单元测试。它产出一个有序的 :class:`ProviderSpec` 列表，交给
:class:`services.llm_service.LLMService` 执行多模型路由与 Fallback。

优先级（design「6. LLMService」、用户诉求）：

1. **用户自配**（最高）：用户在"模型设置"面板填写的提供商 / 模型 / 密钥
   （仅存于该用户的会话状态，不落库）。一旦用户配置了可用模型，本次会话**优先**
   使用用户配置。
2. **管理员配置**（次之）：管理员在后台设置并启用的模型（存于本地 DB
   ``admin_config``，不入公开仓库）。用户**未自行配置**时，可直接使用管理员配置的
   模型，无需自己填密钥。
3. **环境变量**（兜底）：进程环境中已配置密钥的提供商（如 HF Space Secrets /
   本地 ``set XXX_API_KEY``）。

为提高鲁棒性，解析后会把三档**按优先级拼成一条 Fallback 链**（去重）：用户档在前、
管理员档居中、环境变量档兜底。这样即便用户档暂时不可用（如密钥失效），仍能自动
降级到管理员 / 环境变量配置，符合"优雅降级"目标。
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any, Mapping, Optional

from layers.llm_providers import (
    API_KEY_ENV_VARS,
    CUSTOM_BASE_URL_ENV,
    CUSTOM_MODEL_ENV,
    LLMProvider,
)
from services.llm_service import LLMService, ProviderSpec, SlidingWindowRateLimiter

logger = logging.getLogger(__name__)


@dataclass
class UserModelConfig:
    """单个用户在会话内自配的模型设置（不落库，仅存会话状态）。"""

    provider: str = ""
    model: str = ""
    api_key: str = ""
    base_url: str = ""  # 仅 custom 需要

    def is_usable(self) -> bool:
        """是否构成一个可用配置：需有提供商与密钥；custom 还需 base_url 与模型。"""
        if not self.provider or not self.api_key:
            return False
        if self.provider == LLMProvider.CUSTOM.value:
            return bool(self.base_url and self.model)
        return True

    def to_spec(self) -> Optional[ProviderSpec]:
        if not self.is_usable():
            return None
        return ProviderSpec(
            provider=self.provider,
            model=self.model or None,
            api_key=self.api_key,
            base_url=self.base_url or None,
        )


def _admin_spec(admin_cfg: Optional[Mapping[str, Any]]) -> Optional[ProviderSpec]:
    """把管理员配置（来自 ``utils.database.get_default_llm_config``）转为 ProviderSpec。"""
    if not admin_cfg or not admin_cfg.get("enabled"):
        return None
    provider = (admin_cfg.get("provider") or "").strip()
    api_key = (admin_cfg.get("api_key") or "").strip()
    if not provider or not api_key:
        return None
    base_url = (admin_cfg.get("base_url") or "").strip() or None
    model = (admin_cfg.get("model") or "").strip() or None
    if provider == LLMProvider.CUSTOM.value and (not base_url or not model):
        return None
    return ProviderSpec(provider=provider, model=model, api_key=api_key, base_url=base_url)


def _env_specs(env: Mapping[str, str]) -> list[ProviderSpec]:
    """枚举环境中已配置密钥的提供商，构成兜底链（保持声明顺序）。"""
    specs: list[ProviderSpec] = []
    for provider, env_var in API_KEY_ENV_VARS.items():
        key = env.get(env_var)
        if not key:
            continue
        if provider == LLMProvider.CUSTOM:
            base_url = env.get(CUSTOM_BASE_URL_ENV)
            model = env.get(CUSTOM_MODEL_ENV)
            if not base_url or not model:
                continue
            specs.append(ProviderSpec(provider.value, model=model, api_key=key, base_url=base_url))
        else:
            specs.append(ProviderSpec(provider.value, api_key=key))
    return specs


def _dedup(specs: list[ProviderSpec]) -> list[ProviderSpec]:
    """按 (provider, model, base_url) 去重，保留首次出现（即更高优先级）。"""
    seen: set[tuple] = set()
    out: list[ProviderSpec] = []
    for s in specs:
        sig = (s.provider, s.resolved_model(), s.base_url)
        if sig in seen:
            continue
        seen.add(sig)
        out.append(s)
    return out


@dataclass
class ResolvedModelChain:
    """解析结果：有序 ProviderSpec 链 + 来源标注（便于 UI 显示"当前使用谁的配置"）。"""

    specs: list = field(default_factory=list)
    source: str = "none"  # "user" | "admin" | "env" | "none"

    @property
    def active_provider(self) -> Optional[str]:
        return self.specs[0].provider if self.specs else None

    @property
    def active_model(self) -> Optional[str]:
        return self.specs[0].resolved_model() if self.specs else None


def resolve_chain(
    *,
    user_cfg: Optional[UserModelConfig] = None,
    admin_cfg: Optional[Mapping[str, Any]] = None,
    env: Optional[Mapping[str, str]] = None,
) -> ResolvedModelChain:
    """按 用户 > 管理员 > 环境变量 的优先级解析出有序 ProviderSpec 链。

    返回的链已去重；``source`` 标注**首选档**的来源（user / admin / env / none），
    供 UI 告知用户"当前优先使用谁的配置"。即便首选档不可用，链中仍保留较低优先级
    的档作为 Fallback。
    """
    import os

    env = os.environ if env is None else env

    user_spec = user_cfg.to_spec() if user_cfg is not None else None
    admin_spec = _admin_spec(admin_cfg)
    env_specs = _env_specs(env)

    ordered: list[ProviderSpec] = []
    if user_spec is not None:
        ordered.append(user_spec)
    if admin_spec is not None:
        ordered.append(admin_spec)
    ordered.extend(env_specs)
    ordered = _dedup(ordered)

    if user_spec is not None:
        source = "user"
    elif admin_spec is not None:
        source = "admin"
    elif env_specs:
        source = "env"
    else:
        source = "none"

    return ResolvedModelChain(specs=ordered, source=source)


def build_llm_service(
    chain: ResolvedModelChain,
    *,
    usage: Any = None,
    rate_limiter: Any = None,
) -> LLMService:
    """用解析出的链构建 :class:`LLMService`（保持优先级顺序）。

    未提供 ``rate_limiter`` 时套用与 app 一致的默认全局滑动窗口（60s / 30 次）。
    """
    if rate_limiter is None:
        rate_limiter = SlidingWindowRateLimiter(max_calls=30, window=60.0)
    return LLMService(chain.specs, usage=usage, rate_limiter=rate_limiter)


__all__ = [
    "UserModelConfig",
    "ResolvedModelChain",
    "resolve_chain",
    "build_llm_service",
]