from __future__ import annotations from copy import deepcopy import json from pathlib import Path from typing import Any from uuid import uuid4 from deeptutor.services.path_service import get_path_service from .embedding_endpoint import normalize_embedding_endpoint_for_display # Fallback only — frozen at admin scope at import time. Production code should # enter through ``get_model_catalog_service()`` so the path is resolved from the # current user's PathService on every call. CATALOG_PATH = get_path_service().get_settings_file("model_catalog") def _service_shell() -> dict[str, Any]: return { "active_profile_id": None, "active_model_id": None, "profiles": [], } def _search_shell() -> dict[str, Any]: return { "active_profile_id": None, "profiles": [], } def _default_catalog() -> dict[str, Any]: return { "version": 1, "services": { "llm": _service_shell(), "embedding": _service_shell(), "search": _search_shell(), "tts": _service_shell(), "stt": _service_shell(), "imagegen": _service_shell(), "videogen": _service_shell(), }, } class ModelCatalogService: _instances: dict[str, "ModelCatalogService"] = {} def __init__(self, path: Path | None = None): self.path = path or CATALOG_PATH @classmethod def get_instance(cls, path: Path | None = None) -> "ModelCatalogService": resolved = (path or get_path_service().get_settings_file("model_catalog")).resolve() key = str(resolved) if key not in cls._instances: cls._instances[key] = cls(resolved) return cls._instances[key] def load(self) -> dict[str, Any]: loaded = self._read_existing_catalog() if loaded: catalog = _default_catalog() catalog.update({k: v for k, v in loaded.items() if k != "services"}) catalog["services"].update(loaded.get("services", {})) merged_defaults = catalog != loaded before = deepcopy(catalog) self._normalize(catalog) if merged_defaults or catalog != before: self.save(catalog) return catalog catalog = _default_catalog() self._normalize(catalog) self.save(catalog) return catalog def _read_existing_catalog(self) -> dict[str, Any]: if not self.path.exists() or self.path.stat().st_size == 0: return {} try: loaded = json.loads(self.path.read_text(encoding="utf-8")) except Exception: return {} return loaded if isinstance(loaded, dict) else {} def save(self, catalog: dict[str, Any]) -> dict[str, Any]: normalized = deepcopy(catalog) self._normalize(normalized) self.path.parent.mkdir(parents=True, exist_ok=True) with open(self.path, "w", encoding="utf-8") as handle: json.dump(normalized, handle, indent=2, ensure_ascii=False) return normalized def apply(self, catalog: dict[str, Any] | None = None) -> dict[str, Any]: current = self.save(catalog or self.load()) return {"catalog_path": str(self.path), "services": list(current.get("services", {}))} def _normalize(self, catalog: dict[str, Any]) -> bool: services = catalog.setdefault("services", {}) changed = False services.setdefault("llm", _service_shell()) services.setdefault("embedding", _service_shell()) services.setdefault("search", _search_shell()) services.setdefault("tts", _service_shell()) services.setdefault("stt", _service_shell()) services.setdefault("imagegen", _service_shell()) services.setdefault("videogen", _service_shell()) for service_name in ("llm", "embedding", "search", "tts", "stt", "imagegen", "videogen"): service = services[service_name] profiles = service.setdefault("profiles", []) for profile in profiles: profile.setdefault("id", f"{service_name}-profile-{uuid4().hex[:8]}") profile.setdefault("name", "Untitled Profile") profile.setdefault("api_version", "") profile.setdefault("base_url", "") profile.setdefault("api_key", "") if service_name == "search": profile.setdefault("provider", "brave") profile.setdefault("proxy", "") profile["models"] = [] else: profile.setdefault("binding", "openai") profile.setdefault("extra_headers", {}) if service_name == "embedding": before = str(profile.get("base_url") or "") after = normalize_embedding_endpoint_for_display( profile.get("binding"), before, ) if after != before: profile["base_url"] = after changed = True models = profile.setdefault("models", []) for model in models: model.setdefault("id", f"{service_name}-model-{uuid4().hex[:8]}") model.setdefault("name", model.get("model") or "Untitled Model") model.setdefault("model", "") if service_name == "embedding": # Empty default → test_runner auto-fills from the # actual API response on first connection test. model.setdefault("dimension", "") # CSV of supported dims discovered during the last # successful "Test connection" — drives the UI # dropdown. Empty when the model is not in any # adapter's MODELS_INFO map. model.setdefault("supported_dimensions", "") elif service_name == "tts": # Provider/model-specific free-form voice string # (e.g. "alloy", "autumn", "model:voice"). model.setdefault("voice", "") model.setdefault("response_format", "mp3") elif service_name == "imagegen": # Generation knobs; empty → provider default. model.setdefault("size", "") model.setdefault("quality", "") model.setdefault("style", "") model.setdefault("response_format", "") elif service_name == "videogen": model.setdefault("aspect_ratio", "") model.setdefault("duration", "") model.setdefault("resolution", "") profile_ids = {profile.get("id") for profile in profiles} if profiles and service.get("active_profile_id") not in profile_ids: service["active_profile_id"] = profiles[0]["id"] changed = True if service_name in {"llm", "embedding", "tts", "stt", "imagegen", "videogen"}: active_profile = self.get_active_profile(catalog, service_name) models = (active_profile or {}).get("models") or [] model_ids = {model.get("id") for model in models} if models and service.get("active_model_id") not in model_ids: service["active_model_id"] = models[0]["id"] changed = True return changed def get_active_profile( self, catalog: dict[str, Any], service_name: str ) -> dict[str, Any] | None: service = catalog.get("services", {}).get(service_name, {}) active_id = service.get("active_profile_id") for profile in service.get("profiles", []): if profile.get("id") == active_id: return profile profiles = service.get("profiles", []) return profiles[0] if profiles else None def get_active_model(self, catalog: dict[str, Any], service_name: str) -> dict[str, Any] | None: if service_name == "search": return None service = catalog.get("services", {}).get(service_name, {}) active_model_id = service.get("active_model_id") profile = self.get_active_profile(catalog, service_name) if not profile: return None for model in profile.get("models", []): if model.get("id") == active_model_id: return model models = profile.get("models", []) return models[0] if models else None def get_model_catalog_service() -> ModelCatalogService: try: from deeptutor.multi_user.context import get_current_user from deeptutor.multi_user.paths import get_admin_path_service if not get_current_user().is_admin: return ModelCatalogService.get_instance( get_admin_path_service().get_settings_file("model_catalog") ) except Exception: pass return ModelCatalogService.get_instance(get_path_service().get_settings_file("model_catalog")) __all__ = ["CATALOG_PATH", "ModelCatalogService", "get_model_catalog_service"]