Редактировать config_loader.py
This commit is contained in:
Markov Andrey
@@ -1,290 +1,148 @@
|
|||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
"""
|
"""
|
||||||
Загрузчик конфигурации бота из bot.conf и переменных окружения.
|
Загрузчик конфигурации с использованием Pydantic.
|
||||||
Собирает все настройки в единый объект BotConfig.
|
Читает YAML-файлы (bot.conf и rag.conf), подставляет переменные окружения,
|
||||||
|
создаёт экземпляр AppConfig.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import yaml
|
import yaml
|
||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import List, Optional, Any, Dict
|
from typing import Optional, Dict, Any
|
||||||
|
|
||||||
try:
|
from dotenv import load_dotenv
|
||||||
from dotenv import load_dotenv
|
|
||||||
load_dotenv()
|
from .config_models import AppConfig, DatabaseConfig, QdrantConfig, GigaChatConfig, \
|
||||||
except ImportError:
|
EmbeddingConfig, ChunkingConfig, RAGConfig, HTTPServerConfig, HTTPConfig, \
|
||||||
pass
|
SummarizationConfig, PathsConfig, FeaturesConfig, CleanupConfig, FilePermissionsConfig, \
|
||||||
|
AIConfig, PromptsConfig, PromptTemperatures
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
# Загружаем переменные окружения из .env
|
||||||
|
load_dotenv()
|
||||||
|
|
||||||
|
|
||||||
class BotConfig:
|
def load_config(profile_dir: str) -> AppConfig:
|
||||||
"""
|
"""
|
||||||
Единый объект конфигурации для одного профиля бота.
|
Загружает конфигурацию из профиля.
|
||||||
Все параметры читаются из bot.conf, core.conf и переменных окружения.
|
|
||||||
При отсутствии любого параметра – ошибка.
|
Аргументы:
|
||||||
|
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}")
|
||||||
|
|
||||||
def __init__(self, profile_dir: str):
|
# Загружаем bot.conf
|
||||||
self.profile_dir = Path(profile_dir).resolve()
|
with open(bot_config_path, 'r', encoding='utf-8') as f:
|
||||||
self.bot_config_path = self.profile_dir / "bot.conf"
|
bot_raw = yaml.safe_load(f) or {}
|
||||||
if not self.bot_config_path.exists():
|
|
||||||
raise FileNotFoundError(f"Файл профиля не найден: {self.bot_config_path}")
|
|
||||||
|
|
||||||
with open(self.bot_config_path, 'r', encoding='utf-8') as f:
|
# Загружаем rag.conf (общий для всех ботов) – ищем в родительском каталоге или рядом
|
||||||
self.bot_raw = yaml.safe_load(f)
|
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}")
|
||||||
|
|
||||||
# Загружаем core.conf
|
with open(rag_config_path, 'r', encoding='utf-8') as f:
|
||||||
core_config_path = self.profile_dir.parent / "core" / "core.conf"
|
rag_raw = yaml.safe_load(f) or {}
|
||||||
if core_config_path.exists():
|
|
||||||
with open(core_config_path, 'r', encoding='utf-8') as f:
|
|
||||||
self.core_raw = yaml.safe_load(f) or {}
|
|
||||||
else:
|
|
||||||
self.core_raw = {}
|
|
||||||
|
|
||||||
# Объединяем: bot.conf имеет приоритет над core.conf
|
# Глубокое слияние: bot.conf имеет приоритет над rag.conf
|
||||||
self.raw = self._merge_configs(self.core_raw, self.bot_raw)
|
merged = deep_merge(rag_raw, bot_raw)
|
||||||
|
|
||||||
# ----- 1. Обязательные поля (без них бот не запустится) -----
|
# Извлекаем переменные окружения
|
||||||
self.name: str = self._get_required("name")
|
env_vars = {
|
||||||
self.jid: str = self._get_required("jid")
|
'XMPP_PASSWORD': os.getenv('XMPP_PASSWORD'),
|
||||||
self.password_env: str = self._get_required("password_env")
|
'DB_PASSWORD': os.getenv('DB_PASSWORD'),
|
||||||
self.xmpp_password: str = self._get_env(self.password_env)
|
'GIGACHAT_API_KEY': os.getenv('GIGACHAT_API_KEY'),
|
||||||
|
'SALUTE_SPEECH_AUTH': os.getenv('SALUTE_SPEECH_AUTH'),
|
||||||
|
}
|
||||||
|
|
||||||
# ----- 2. XMPP параметры (обязательные) -----
|
# Подставляем пароли и ключи в секции
|
||||||
self.xmpp_resource = self._get_required("resource")
|
# В database: password_env -> фактический пароль
|
||||||
self.status_message = self._get_required("status_message")
|
if 'database' in merged and 'password_env' in merged['database']:
|
||||||
self.xmpp_server = self._get_required("xmpp_server")
|
env_name = merged['database']['password_env']
|
||||||
self.xmpp_port = int(self._get_required("xmpp_port"))
|
merged['database']['password'] = env_vars.get(env_name)
|
||||||
self.xmpp_use_ssl = bool(self._get_required("xmpp_use_ssl"))
|
if not merged['database']['password']:
|
||||||
|
raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
|
||||||
|
# удаляем password_env, чтобы не передавать в модель
|
||||||
|
del merged['database']['password_env']
|
||||||
|
|
||||||
# ----- 3. Настройки базы данных (PostgreSQL) -----
|
# Аналогично для gigachat
|
||||||
db_cfg = self._get_required_section("database")
|
if 'gigachat' in merged and 'credentials_env' in merged['gigachat']:
|
||||||
self.db_name = self._get_required_from_section(db_cfg, "database")
|
env_name = merged['gigachat']['credentials_env']
|
||||||
self.db_type = self._get_required_from_section(db_cfg, "type")
|
merged['gigachat']['api_key'] = env_vars.get(env_name)
|
||||||
self.db_host = self._get_required_from_section(db_cfg, "host")
|
if not merged['gigachat']['api_key']:
|
||||||
self.db_port = int(self._get_required_from_section(db_cfg, "port"))
|
raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
|
||||||
self.db_user = self._get_required_from_section(db_cfg, "user")
|
del merged['gigachat']['credentials_env']
|
||||||
self.db_password_env = self._get_required_from_section(db_cfg, "password_env")
|
|
||||||
self.db_password = self._get_env(self.db_password_env)
|
|
||||||
|
|
||||||
# SSL параметры (опциональны, но если секция ssl не задана – будет ошибка, если нет)
|
# Для XMPP пароля
|
||||||
self.db_ssl = self._get_optional_from_section(db_cfg, "ssl", default=None)
|
if 'password_env' in merged:
|
||||||
if self.db_ssl is None:
|
env_name = merged['password_env']
|
||||||
raise ValueError("В секции database отсутствует обязательный параметр 'ssl'")
|
merged['xmpp_password'] = env_vars.get(env_name)
|
||||||
self.db_ssl_mode = self._get_optional_from_section(db_cfg, "ssl_mode", default=None)
|
if not merged['xmpp_password']:
|
||||||
if self.db_ssl_mode is None:
|
raise ValueError(f"Переменная окружения {env_name} не задана или пуста")
|
||||||
raise ValueError("В секции database отсутствует обязательный параметр 'ssl_mode'")
|
del merged['password_env']
|
||||||
|
|
||||||
# ----- 4. Настройки Qdrant -----
|
# Для SaluteSpeech (если есть в секции features или отдельно)
|
||||||
qdrant_cfg = self._get_required_section("qdrant")
|
# У нас он может быть в .env, но в модели нет отдельного поля; добавим в итоговый объект
|
||||||
self.qdrant_collection = self._get_required_from_section(qdrant_cfg, "collection")
|
merged['salute_speech_auth'] = env_vars.get('SALUTE_SPEECH_AUTH')
|
||||||
self.qdrant_host = self._get_required_from_section(qdrant_cfg, "host")
|
|
||||||
self.qdrant_port = int(self._get_required_from_section(qdrant_cfg, "port"))
|
|
||||||
self.qdrant_grpc_port = int(self._get_required_from_section(qdrant_cfg, "grpc_port"))
|
|
||||||
self.qdrant_vector_size = int(self._get_required_from_section(qdrant_cfg, "vector_size"))
|
|
||||||
self.qdrant_distance = self._get_required_from_section(qdrant_cfg, "distance")
|
|
||||||
|
|
||||||
# ----- 5. Настройки GigaChat -----
|
# Преобразуем пути в абсолютные с учётом profile_dir
|
||||||
gc_cfg = self._get_required_section("gigachat")
|
paths_section = merged.get('paths', {})
|
||||||
self.gigachat_credentials_env = self._get_required_from_section(gc_cfg, "credentials_env")
|
# Если пути относительные, дополняем их относительно profile_dir
|
||||||
self.gigachat_api_key = self._get_env(self.gigachat_credentials_env)
|
for key in ['data_dir', 'temp_dir', 'prompts_dir', 'upload_dir', 'log_dir']:
|
||||||
self.gigachat_model_embedding = self._get_required_from_section(gc_cfg, "model_embedding")
|
if key in paths_section:
|
||||||
self.gigachat_model_generation = self._get_required_from_section(gc_cfg, "model_generation")
|
p = Path(paths_section[key])
|
||||||
self.gigachat_timeout = int(self._get_required_from_section(gc_cfg, "timeout"))
|
if not p.is_absolute():
|
||||||
self.gigachat_temperature = float(self._get_required_from_section(gc_cfg, "temperature"))
|
paths_section[key] = str(profile_path / p)
|
||||||
|
# log_filename оставляем как есть
|
||||||
|
|
||||||
# ----- 6. Кэш эмбеддингов -----
|
# Добавляем profile_dir в merged для передачи в модель
|
||||||
emb_cfg = self._get_required_section("embedding")
|
merged['profile_dir'] = str(profile_path)
|
||||||
self.embedding_cache_size = int(self._get_required_from_section(emb_cfg, "cache_size"))
|
|
||||||
self.embedding_model = self._get_required_from_section(emb_cfg, "model")
|
|
||||||
self.embedding_timeout = int(self._get_required_from_section(emb_cfg, "timeout"))
|
|
||||||
self.embedding_verify_ssl = bool(self._get_required_from_section(emb_cfg, "verify_ssl_certs"))
|
|
||||||
|
|
||||||
# ----- 7. Разбиение документов на чанки -----
|
# Создаём экземпляр AppConfig (Pydantic выполнит валидацию)
|
||||||
chunk_cfg = self._get_required_section("chunking")
|
try:
|
||||||
self.chunking_enabled = bool(self._get_required_from_section(chunk_cfg, "enabled"))
|
config = AppConfig(**merged)
|
||||||
self.chunk_size_tokens = int(self._get_required_from_section(chunk_cfg, "chunk_size_tokens"))
|
except Exception as e:
|
||||||
self.overlap_tokens = int(self._get_required_from_section(chunk_cfg, "overlap_tokens"))
|
raise ValueError(f"Ошибка валидации конфигурации: {e}")
|
||||||
self.chunking_strategy = self._get_required_from_section(chunk_cfg, "strategy")
|
|
||||||
self.chunking_approx_chunk_chars = int(self._get_required_from_section(chunk_cfg, "approx_chunk_chars"))
|
|
||||||
self.chunking_approx_overlap_chars = int(self._get_required_from_section(chunk_cfg, "approx_overlap_chars"))
|
|
||||||
|
|
||||||
# ----- 8. Пути к файлам и каталогам -----
|
# Дополнительно загружаем промпты из файлов
|
||||||
paths = self._get_required_section("paths")
|
prompts_dir = config.prompts_dir_path
|
||||||
self.data_dir = (self.profile_dir / paths.get("data_dir")).resolve()
|
if prompts_dir and prompts_dir.exists():
|
||||||
self.temp_dir = (self.profile_dir / paths.get("temp_dir")).resolve()
|
prompts_content = {}
|
||||||
self.prompts_dir = (self.profile_dir / paths.get("prompts_dir")).resolve()
|
for prompt_name, filename in config.ai.prompts.items():
|
||||||
self.upload_dir = Path(paths.get("upload_dir")).resolve()
|
file_path = prompts_dir / filename
|
||||||
self.log_dir = Path(paths.get("log_dir")).resolve()
|
if file_path.exists():
|
||||||
log_filename = paths.get("log_filename")
|
try:
|
||||||
if not log_filename:
|
with open(file_path, 'r', encoding='utf-8') as f:
|
||||||
raise ValueError("В секции paths отсутствует log_filename")
|
prompts_content[prompt_name] = f.read()
|
||||||
self.log_file = self.log_dir / log_filename.format(name=self.name)
|
except Exception as e:
|
||||||
|
logger.warning(f"Не удалось загрузить промпт {prompt_name} из {file_path}: {e}")
|
||||||
# ----- 9. HTTP Upload -----
|
|
||||||
http_up = self._get_required_section("http_upload")
|
|
||||||
self.upload_base_url = http_up.get("base_url")
|
|
||||||
|
|
||||||
# ----- 10. Настройки AI (промпты, модель) -----
|
|
||||||
ai_cfg = self._get_required_section("ai")
|
|
||||||
self.ai_model = ai_cfg.get("model")
|
|
||||||
if not self.ai_model:
|
|
||||||
self.ai_model = self.gigachat_model_generation
|
|
||||||
self.ai_temperature = float(ai_cfg.get("temperature", self.gigachat_temperature))
|
|
||||||
self.ai_timeout = int(ai_cfg.get("timeout", self.gigachat_timeout))
|
|
||||||
|
|
||||||
prompts_cfg = ai_cfg.get("prompts")
|
|
||||||
if not prompts_cfg:
|
|
||||||
raise ValueError("В секции ai отсутствует подсекция prompts")
|
|
||||||
|
|
||||||
# промпты – обязательные файлы
|
|
||||||
self.system_prompt_file = self.prompts_dir / prompts_cfg.get("system")
|
|
||||||
self.synthesis_prompt_file = self.prompts_dir / prompts_cfg.get("synthesis")
|
|
||||||
self.quality_prompt_file = self.prompts_dir / prompts_cfg.get("quality")
|
|
||||||
self.rerank_prompt_file = self.prompts_dir / prompts_cfg.get("rerank")
|
|
||||||
self.expand_prompt_file = self.prompts_dir / prompts_cfg.get("expand")
|
|
||||||
self.intent_prompt_file = self.prompts_dir / prompts_cfg.get("intent")
|
|
||||||
self.metrics_prompt_file = self.prompts_dir / prompts_cfg.get("metrics")
|
|
||||||
self.summary_prompt_file = self.prompts_dir / prompts_cfg.get("summary")
|
|
||||||
self.consistency_prompt_file = self.prompts_dir / prompts_cfg.get("consistency")
|
|
||||||
self.critique_prompt_file = self.prompts_dir / prompts_cfg.get("critique")
|
|
||||||
self.spellcheck_prompt_file = self.prompts_dir / prompts_cfg.get("spellcheck")
|
|
||||||
self.generate_document_prompt_file = self.prompts_dir / prompts_cfg.get("generate_document")
|
|
||||||
|
|
||||||
# ----- 11. Параметры RAG -----
|
|
||||||
rag_cfg = self._get_required_section("rag")
|
|
||||||
self.rag_default_top_k = int(rag_cfg.get("default_top_k"))
|
|
||||||
self.rag_metrics_top_k = int(rag_cfg.get("metrics_top_k"))
|
|
||||||
self.rag_contradiction_top_k = int(rag_cfg.get("contradiction_top_k"))
|
|
||||||
self.max_context_tokens = int(rag_cfg.get("max_context_tokens"))
|
|
||||||
|
|
||||||
# ----- 12. Веб-скрапер -----
|
|
||||||
ws_cfg = self._get_required_section("web_scraper")
|
|
||||||
self.ws_max_pages = int(ws_cfg.get("max_pages"))
|
|
||||||
self.ws_max_depth = int(ws_cfg.get("max_depth"))
|
|
||||||
|
|
||||||
# ----- 13. Включение/отключение функций -----
|
|
||||||
feat = self._get_required_section("features")
|
|
||||||
self.file_processing = bool(feat.get("file_processing"))
|
|
||||||
self.surgical_replace = bool(feat.get("surgical_replace"))
|
|
||||||
self.voice_recognition = bool(feat.get("voice_recognition"))
|
|
||||||
self.vision = bool(feat.get("vision"))
|
|
||||||
self.archive_support = bool(feat.get("archive_support"))
|
|
||||||
self.ocr = bool(feat.get("ocr"))
|
|
||||||
self.allow_public_knowledge = bool(feat.get("allow_public_knowledge"))
|
|
||||||
self.max_file_size_mb = int(feat.get("max_file_size_mb"))
|
|
||||||
self.max_archive_files = int(feat.get("max_archive_files"))
|
|
||||||
self.enable_intent_classification = bool(feat.get("enable_intent_classification"))
|
|
||||||
self.enable_self_critique = bool(feat.get("enable_self_critique"))
|
|
||||||
self.surgical_keywords = feat.get("surgical_keywords")
|
|
||||||
if not isinstance(self.surgical_keywords, list):
|
|
||||||
raise ValueError("surgical_keywords должен быть списком строк")
|
|
||||||
self.mention_keyword = feat.get("mention_keyword", self.name) # если не указано, используем имя бота
|
|
||||||
|
|
||||||
# ----- 14. Очистка старых файлов -----
|
|
||||||
clean_cfg = self._get_required_section("cleanup")
|
|
||||||
self.temp_cleanup_days = int(clean_cfg.get("temp_days"))
|
|
||||||
self.upload_cleanup_days = int(clean_cfg.get("upload_days"))
|
|
||||||
self.cleanup_interval = int(clean_cfg.get("interval_seconds"))
|
|
||||||
self.user_data_cleanup_days = int(clean_cfg.get("user_data_days"))
|
|
||||||
|
|
||||||
# ----- 15. Таймауты HTTP -----
|
|
||||||
http_cfg = self._get_required_section("http")
|
|
||||||
self.http_download_timeout = int(http_cfg.get("download_timeout"))
|
|
||||||
self.http_upload_timeout = int(http_cfg.get("upload_timeout"))
|
|
||||||
|
|
||||||
# ----- 16. Права на создаваемые файлы -----
|
|
||||||
fp_cfg = self._get_required_section("file_permissions")
|
|
||||||
self.file_mode = int(fp_cfg.get("mode"), 8) # восьмеричное
|
|
||||||
|
|
||||||
# ----- 17. Администраторы -----
|
|
||||||
self.admin_jids = self._get_required("admin_jids")
|
|
||||||
if not isinstance(self.admin_jids, list):
|
|
||||||
raise ValueError("admin_jids должен быть списком JID")
|
|
||||||
|
|
||||||
# ----- 18. SaluteSpeech (из .env) -----
|
|
||||||
self.salute_speech_auth = os.getenv("SALUTE_SPEECH_AUTH", "")
|
|
||||||
if not self.salute_speech_auth:
|
|
||||||
raise ValueError("Переменная окружения SALUTE_SPEECH_AUTH не задана")
|
|
||||||
|
|
||||||
# ----- 19. Log level -----
|
|
||||||
log_level_str = self._get_required("log_level")
|
|
||||||
self.log_level = getattr(logging, log_level_str.upper(), logging.INFO)
|
|
||||||
|
|
||||||
# ----- 20. Настройки функций (intent, expand, metrics, summary, consistency, critique, rerank) -----
|
|
||||||
# Все они читаются из соответствующих секций, но сами секции могут отсутствовать?
|
|
||||||
# Поскольку мы требуем всё явно, то каждая секция должна быть.
|
|
||||||
self.intent_temperature = float(self._get_required_from_section(self._get_required_section("intent"), "temperature"))
|
|
||||||
self.expand_temperature = float(self._get_required_from_section(self._get_required_section("expand"), "temperature"))
|
|
||||||
self.metrics_temperature = float(self._get_required_from_section(self._get_required_section("metrics"), "temperature"))
|
|
||||||
self.summary_temperature = float(self._get_required_from_section(self._get_required_section("summary"), "temperature"))
|
|
||||||
self.summary_max_chars = int(self._get_required_from_section(self._get_required_section("summary"), "max_chars"))
|
|
||||||
self.consistency_temperature = float(self._get_required_from_section(self._get_required_section("consistency"), "temperature"))
|
|
||||||
self.consistency_max_fragments = int(self._get_required_from_section(self._get_required_section("consistency"), "max_fragments"))
|
|
||||||
self.critique_temperature = float(self._get_required_from_section(self._get_required_section("critique"), "temperature"))
|
|
||||||
self.rerank_temperature = float(self._get_required_from_section(self._get_required_section("rerank"), "temperature"))
|
|
||||||
self.rerank_min_length = int(self._get_required_from_section(self._get_required_section("rerank"), "min_length"))
|
|
||||||
self.generate_document_temperature = float(self._get_required_from_section(self._get_required_section("generate_document"), "temperature"))
|
|
||||||
|
|
||||||
# ----- 21. Совместимость с функциями: создаём вложенные объекты -----
|
|
||||||
self.intent = {"temperature": self.intent_temperature}
|
|
||||||
self.expand = {"temperature": self.expand_temperature}
|
|
||||||
self.metrics = {"temperature": self.metrics_temperature}
|
|
||||||
self.summary = {"temperature": self.summary_temperature, "max_chars": self.summary_max_chars}
|
|
||||||
self.consistency = {"temperature": self.consistency_temperature, "max_fragments": self.consistency_max_fragments}
|
|
||||||
self.critique = {"temperature": self.critique_temperature}
|
|
||||||
self.rerank = {"temperature": self.rerank_temperature, "min_length": self.rerank_min_length}
|
|
||||||
|
|
||||||
# Создаём директории
|
|
||||||
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.temp_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.prompts_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.upload_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
||||||
|
|
||||||
# ------------------------------------------------------------------
|
|
||||||
# Вспомогательные методы
|
|
||||||
# ------------------------------------------------------------------
|
|
||||||
def _merge_configs(self, core: dict, bot: dict) -> dict:
|
|
||||||
"""Рекурсивное слияние: bot имеет приоритет над core."""
|
|
||||||
merged = core.copy()
|
|
||||||
for key, value in bot.items():
|
|
||||||
if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
|
|
||||||
merged[key] = self._merge_configs(merged[key], value)
|
|
||||||
else:
|
else:
|
||||||
merged[key] = value
|
logger.warning(f"Файл промпта {filename} не найден в {prompts_dir}")
|
||||||
return merged
|
config.prompts_content = prompts_content
|
||||||
|
else:
|
||||||
|
logger.warning(f"Каталог промптов не существует: {prompts_dir}")
|
||||||
|
|
||||||
def _get_required(self, key: str) -> Any:
|
return config
|
||||||
"""Возвращает значение ключа из объединённого конфига. Если нет – ошибка."""
|
|
||||||
if key not in self.raw:
|
|
||||||
raise KeyError(f"В конфигурации (bot.conf или core.conf) отсутствует обязательный ключ: '{key}'")
|
|
||||||
return self.raw[key]
|
|
||||||
|
|
||||||
def _get_required_section(self, section: str) -> dict:
|
|
||||||
"""Возвращает словарь-секцию. Если секции нет – ошибка."""
|
|
||||||
if section not in self.raw:
|
|
||||||
raise KeyError(f"В конфигурации отсутствует обязательная секция: '{section}'")
|
|
||||||
if not isinstance(self.raw[section], dict):
|
|
||||||
raise TypeError(f"Секция '{section}' должна быть словарём")
|
|
||||||
return self.raw[section]
|
|
||||||
|
|
||||||
def _get_required_from_section(self, section_dict: dict, key: str) -> Any:
|
def deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
"""Возвращает значение ключа внутри секции. Если нет – ошибка."""
|
"""
|
||||||
if key not in section_dict:
|
Рекурсивное слияние словарей: override имеет приоритет над base.
|
||||||
raise KeyError(f"В секции отсутствует обязательный ключ: '{key}'")
|
"""
|
||||||
return section_dict[key]
|
result = base.copy()
|
||||||
|
for key, value in override.items():
|
||||||
def _get_optional_from_section(self, section_dict: dict, key: str, default=None):
|
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
|
||||||
"""Возвращает значение или default, если ключ отсутствует (не вызывает ошибку)."""
|
result[key] = deep_merge(result[key], value)
|
||||||
return section_dict.get(key, default)
|
else:
|
||||||
|
result[key] = value
|
||||||
def _get_env(self, env_var: str) -> str:
|
return result
|
||||||
"""Читает переменную окружения. Если нет – ошибка."""
|
|
||||||
value = os.getenv(env_var)
|
|
||||||
if value is None or value.strip() == "":
|
|
||||||
raise ValueError(
|
|
||||||
f"Переменная окружения {env_var} не задана или пуста. "
|
|
||||||
f"Бот не может запуститься без неё."
|
|
||||||
)
|
|
||||||
return value.strip()
|
|
||||||
Reference in New Issue
Block a user