''' File: config.py Description: Класс работы с конфигурацией приложения ''' __version__ = '0.4.0' from pydantic_settings import BaseSettings, JsonConfigSettingsSource, SettingsConfigDict from pydantic import BaseModel import os import json import re import tempfile from pathlib import Path from threading import RLock def validate_folder(folder: str) -> str: '''Имя экземпляра — один компонент пути Windows внутри базовой папки.''' if (not isinstance(folder, str) or not folder or len(folder) > 80 or folder != folder.strip() or folder.endswith('.') or re.search(r'[<>:"/\\|?*\x00-\x1f]', folder) or folder in ('.', '..') or re.fullmatch(r'(CON|PRN|AUX|NUL|COM[0-9]|LPT[0-9])(?:\..*)?', folder, re.I)): raise ValueError('Укажите допустимое имя папки Windows длиной до 80 символов.') return folder def instance_directory(mt5_folder: str | Path, folder: str) -> Path: return Path(mt5_folder) / validate_folder(folder) class ConfigData(BaseModel): # Словарь с информацией о терминалах terminals: dict # Путь к папке с экземплярами терминалов mt5_folder: str # Имя исполняемого файла терминала mt5_exe: str # Путь к общей базе SQLite (папка Common/Files, зависит от профиля Windows) database: str # Относительные пути считаются от папки config.json. terminal_template: str = 'terminal-template' mt5_installation: str = '' service_ex5: str = '' startup_timeout_sec: int = 180 disk_warning_mb: int = 1024 model_config = SettingsConfigDict() class Config(ConfigData, BaseSettings): @classmethod def settings_customise_sources( cls, settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings, ): # Получаем путь к файлу из переменной окружения config_file = os.getenv("MT5_MANAGER_CONFIG_FILE", "config.json") return ( init_settings, JsonConfigSettingsSource(settings_cls, json_file=config_file), env_settings, dotenv_settings, file_secret_settings, ) class ConfigStore: '''Читает настройки и атомарно заменяет JSON при сохранении.''' def __init__(self, path: str | Path | None = None): self.path = Path(path or os.getenv('MT5_MANAGER_CONFIG_FILE', 'config.json')).resolve() self.lock = RLock() self.data = self._validate(json.loads(self.path.read_text(encoding='utf-8-sig'))) def resolve(self, value: str) -> Path: path = Path(value) return (self.path.parent / path).resolve() if not path.is_absolute() else path.resolve() def _validate(self, data: dict) -> dict: # Обычная модель проверяет именно этот документ. BaseSettings здесь # мог бы дополнить пустой terminals настройками из другого JSON. config = ConfigData.model_validate(data) if not 10 <= config.startup_timeout_sec <= 900: raise ValueError('Время ожидания должно быть от 10 до 900 секунд.') if config.disk_warning_mb < 0: raise ValueError('Запас свободного места не может быть отрицательным.') result = config.model_dump() root = self.resolve(config.mt5_folder) result['mt5_folder'] = str(root) result['database'] = str(self.resolve(config.database)) result['terminal_template'] = str(self.resolve(config.terminal_template)) if config.mt5_installation: result['mt5_installation'] = str(self.resolve(config.mt5_installation)) if config.service_ex5: result['service_ex5'] = str(self.resolve(config.service_ex5)) seen = set() for folder, record in result['terminals'].items(): validate_folder(folder) if not isinstance(record, dict): raise ValueError('Настройки экземпляра должны быть объектом JSON.') key = folder.casefold() if key in seen: raise ValueError('Имена экземпляров различаются только регистром.') seen.add(key) record.setdefault('name', folder) record.setdefault('login', '') record.setdefault('server', '') record.setdefault('password', '') # Старые конфиги читаются, но path больше не определяет расположение. # При следующем сохранении устаревшее поле исчезнет из JSON. record.pop('path', None) return result def save(self, data: dict) -> None: with self.lock: validated = self._validate(data) temporary = None try: with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=self.path.parent, delete=False, prefix=self.path.name + '.', suffix='.tmp') as out: temporary = Path(out.name) json.dump(validated, out, ensure_ascii=False, indent=2) out.write('\n') out.flush() os.fsync(out.fileno()) os.replace(temporary, self.path) self.data = validated finally: if temporary: temporary.unlink(missing_ok=True)