fckbot/rag/utils/config_loader.py

# -*- coding: utf-8 -*-
"""
Загрузчик конфигурации с использованием Pydantic.
Читает YAML-файлы (bot.conf и rag.conf), подставляет переменные окружения,
создаёт экземпляр AppConfig.
"""

import os
import yaml
import logging
from pathlib import Path
from typing import Dict, Any

from dotenv import load_dotenv

from ..config_models import AppConfig

logger = logging.getLogger(__name__)

# Загружаем переменные окружения из .env
load_dotenv()


def load_config(profile_dir: str) -> AppConfig:
    """
    Загружает конфигурацию из профиля.

    Аргументы:
        profile_dir: путь к каталогу профиля бота (например, /usr/local/etc/bots/metabot)

    Возвращает:
        AppConfig: экземпляр модели конфигурации.
    """
    profile_path = Path(profile_dir).resolve()
    bot_config_path = profile_path / "bot.conf"
    if not bot_config_path.exists():
        raise FileNotFoundError(f"Файл конфигурации бота не найден: {bot_config_path}")

    # Загружаем bot.conf
    with open(bot_config_path, 'r', encoding='utf-8') as f:
        bot_raw = yaml.safe_load(f) or {}

    # Загружаем rag.conf (общий для всех ботов) – ищем в родительском каталоге или рядом
    rag_config_path = profile_path.parent / "rag" / "rag.conf"
    if not rag_config_path.exists():
        # fallback: возможно, rag.conf лежит в той же директории, что и profile
        rag_config_path = profile_path / "rag.conf"
    if not rag_config_path.exists():
        raise FileNotFoundError(f"Файл rag.conf не найден: {rag_config_path}")

    with open(rag_config_path, 'r', encoding='utf-8') as f:
        rag_raw = yaml.safe_load(f) or {}

    # Глубокое слияние: bot.conf имеет приоритет над rag.conf
    merged = deep_merge(rag_raw, bot_raw)

    # Извлекаем переменные окружения
    env_vars = {
        'XMPP_PASSWORD': os.getenv('XMPP_PASSWORD'),
        'DB_PASSWORD': os.getenv('DB_PASSWORD'),
        'GIGACHAT_API_KEY': os.getenv('GIGACHAT_API_KEY'),
        'SALUTE_SPEECH_AUTH': os.getenv('SALUTE_SPEECH_AUTH'),
    }

    # Подставляем пароли и ключи в секции
    if 'database' in merged and 'password_env' in merged['database']:
        env_name = merged['database']['password_env']
        merged['database']['password'] = env_vars.get(env_name)
        if not merged['database']['password']:
            raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
        del merged['database']['password_env']

    if 'gigachat' in merged and 'credentials_env' in merged['gigachat']:
        env_name = merged['gigachat']['credentials_env']
        merged['gigachat']['api_key'] = env_vars.get(env_name)
        if not merged['gigachat']['api_key']:
            raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
        del merged['gigachat']['credentials_env']

    if 'password_env' in merged:
        env_name = merged['password_env']
        merged['xmpp_password'] = env_vars.get(env_name)
        if not merged['xmpp_password']:
            raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
        del merged['password_env']

    merged['salute_speech_auth'] = env_vars.get('SALUTE_SPEECH_AUTH')

    # Преобразуем пути в абсолютные с учётом profile_dir
    paths_section = merged.get('paths', {})
    for key in ['data_dir', 'temp_dir', 'prompts_dir', 'upload_dir', 'log_dir']:
        if key in paths_section:
            p = Path(paths_section[key])
            if not p.is_absolute():
                paths_section[key] = str(profile_path / p)

    # Добавляем profile_dir
    merged['profile_dir'] = str(profile_path)

    # Добавляем web_scraper (если отсутствует)
    if 'web_scraper' not in merged:
        merged['web_scraper'] = {'max_pages': 5, 'max_depth': 1}

    # Добавляем секцию auth, если её нет
    if 'auth' not in merged:
        merged['auth'] = {}

    # Добавляем секцию fewshot, если её нет
    if 'fewshot' not in merged:
        merged['fewshot'] = {}

    # Создаём экземпляр AppConfig
    try:
        config = AppConfig(**merged)
    except Exception as e:
        raise ValueError(f"Ошибка валидации конфигурации: {e}")

    # Загружаем API-ключ для RAG-сервера
    api_key_env = merged.get('auth', {}).get('api_key_env', 'RAG_API_KEY')
    config._rag_api_key = os.getenv(api_key_env)
    if not config._rag_api_key:
        logger.warning(f"Переменная окружения {api_key_env} не задана. Аутентификация на RAG-сервере отключена (НЕ БЕЗОПАСНО!)")

    # Загружаем промпты из файлов
    prompts_dir = config.prompts_dir_path
    if prompts_dir and prompts_dir.exists():
        prompts_content = {}
        for prompt_name, filename in config.ai.prompts.items():
            file_path = prompts_dir / filename
            if file_path.exists():
                try:
                    with open(file_path, 'r', encoding='utf-8') as f:
                        prompts_content[prompt_name] = f.read()
                except Exception as e:
                    logger.warning(f"Не удалось загрузить промпт {prompt_name} из {file_path}: {e}")
            else:
                logger.warning(f"Файл промпта {filename} не найден в {prompts_dir}")
        config.prompts_content = prompts_content
    else:
        logger.warning(f"Каталог промптов не существует: {prompts_dir}")

    return config


def deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
    """
    Рекурсивное слияние словарей: override имеет приоритет над base.
    """
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result