mt5-manager/config.py

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