| '''
|
| 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)
|
