config/loader.py - master - YoungFreddy/FinalWork

config/loader.py
117 строк5 KB
Igor
Initial commit: Q&A LLM service
23 июл 2026, 18:02
23 июл 2026, 18:02e5e37db
"""Загрузчик конфигурации: читает config.json и .env, валидирует значения."""

import json
import os
import re
from pathlib import Path
from typing import Any, Dict, Optional

CONFIG_DIR = Path(__file__).parent.resolve()
DEFAULT_CONFIG_PATH = CONFIG_DIR / "config.json"


def _resolve_env_vars(value: Any) -> Any:
    """Заменяет ${VAR_NAME} в строке на значение переменной окружения."""
    if isinstance(value, str):
        def replace_var(match):
            var_name = match.group(1)
            return os.environ.get(var_name, match.group(0))
        return re.sub(r"\$\{(\w+)\}", replace_var, value)
    if isinstance(value, dict):
        return {k: _resolve_env_vars(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_resolve_env_vars(v) for v in value]
    return value


def load_json_config(path: str) -> Optional[Dict[str, Any]]:
    """Загружает конфигурацию из JSON-файла. Возвращает None при ошибке."""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return None


def _validate_positive_int(value: Any, name: str) -> int:
    """Проверяет, что значение — положительное целое число."""
    try:
        v = int(value)
    except (TypeError, ValueError):
        raise ValueError(f"{name} must be a positive integer, got {value!r}")
    if v <= 0:
        raise ValueError(f"{name} must be positive, got {v}")
    return v


def _validate_float_range(value: Any, name: str, lo: float, hi: float) -> float:
    """Проверяет, что значение — число в диапазоне [lo, hi]."""
    try:
        v = float(value)
    except (TypeError, ValueError):
        raise ValueError(f"{name} must be a number, got {value!r}")
    if not (lo <= v <= hi):
        raise ValueError(f"{name} must be between {lo} and {hi}, got {v}")
    return v


def load_config() -> Dict[str, Any]:
    """Загружает конфиг из config.json с переопределением из .env. Валидирует значения."""
    config: Dict[str, Any] = {}

    if DEFAULT_CONFIG_PATH.exists():
        loaded = load_json_config(str(DEFAULT_CONFIG_PATH))
        if loaded:
            config.update(loaded)

    config = _resolve_env_vars(config)

    env_prefix = "LLM_"
    env_overrides = {
        "api_key": os.environ.get(f"{env_prefix}API_KEY"),
        "model": os.environ.get(f"{env_prefix}MODEL"),
        "temperature": os.environ.get(f"{env_prefix}TEMPERATURE"),
        "max_tokens": os.environ.get(f"{env_prefix}MAX_TOKENS"),
        "cache_enabled": os.environ.get(f"{env_prefix}CACHE_ENABLED"),
        "api_timeout": os.environ.get(f"{env_prefix}API_TIMEOUT"),
        "max_retries": os.environ.get(f"{env_prefix}MAX_RETRIES"),
        "retry_delay_base": os.environ.get(f"{env_prefix}RETRY_DELAY_BASE"),
        "cache_ttl": os.environ.get(f"{env_prefix}CACHE_TTL"),
        "system_prompt": os.environ.get(f"{env_prefix}SYSTEM_PROMPT"),
        "fallback_response": os.environ.get(f"{env_prefix}FALLBACK_RESPONSE"),
        "api_base": os.environ.get(f"{env_prefix}API_BASE"),
    }
    for key, val in env_overrides.items():
        if val is not None:
            if key in ("temperature", "api_timeout", "retry_delay_base"):
                config[key] = float(val)
            elif key in ("max_tokens", "max_retries", "cache_ttl"):
                config[key] = int(val)
            elif key == "cache_enabled":
                config[key] = val.lower() == "true"
            else:
                config[key] = val

    config.setdefault("model", "glm-4.7-flash")
    config.setdefault("temperature", 0.7)
    config.setdefault("max_tokens", 1000)
    config.setdefault("api_timeout", 30.0)
    config.setdefault("max_retries", 3)
    config.setdefault("retry_delay_base", 1.0)
    config.setdefault("cache_ttl", 600)
    config.setdefault("cache_enabled", True)
    config.setdefault("system_prompt", "Ты — полезный ассистент. Отвечай на вопросы точно и по делу.")
    config.setdefault("fallback_response", "Сервис временно недоступен, попробуйте позже")
    config.setdefault("api_base", "https://api.z.ai/api/paas/v4")
    config.setdefault("api_key", "")
    config.setdefault("qa_max_question_length", 2000)
    config.setdefault("qa_min_question_length", 1)

    config["max_retries"] = _validate_positive_int(config["max_retries"], "max_retries")
    config["max_tokens"] = _validate_positive_int(config["max_tokens"], "max_tokens")
    config["cache_ttl"] = _validate_positive_int(config["cache_ttl"], "cache_ttl")
    config["api_timeout"] = _validate_float_range(config["api_timeout"], "api_timeout", 0.1, 300.0)
    config["retry_delay_base"] = _validate_float_range(config["retry_delay_base"], "retry_delay_base", 0.01, 60.0)
    config["temperature"] = _validate_float_range(config["temperature"], "temperature", 0.0, 2.0)

    return config