mt5-manager/terminal_manager.py

'''Создание экземпляров и настройка счёта без Python API MetaTrader5.'''

__version__ = '0.9.0'

import copy
import ctypes
import errno
import hashlib
import json
import os
import re
import shutil
import subprocess
import tempfile
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import psutil
from pydantic import ValidationError

from config import ConfigStore, instance_directory, validate_folder
from deploy_service import (compile_service, service_build_paths,
                            services_ini_update, write_services_ini)
from mt5_control import MT5_Control
from project_updates import ProjectUpdates


PROJECT_DIR = Path(__file__).resolve().parent
SERVICE_REL = Path('MQL5/Services/Shared Projects/mt5-manager/ServiceStateWriter.ex5')
UPLOAD_NAMES = {'terminal': 'terminal64.exe', 'editor': 'MetaEditor64.exe'}
MAX_SERVERS_SIZE = 16 * 1024 * 1024
GLOBAL_ACTIONS = ('check_updates', 'update_project', 'deploy_service', 'move_settings')


def account_values(record: dict, *, complete: bool = False) -> tuple:
    login = str(record.get('login') or '').strip()
    server = str(record.get('server') or '').strip()
    password = record.get('password') or ''
    if not isinstance(password, str) or any(c in password + server for c in '\r\n\x00'):
        raise ValueError('В реквизитах счёта недопустимы переводы строк.')
    if not login and not server and not password:
        return '', '', ''
    if not login.isascii() or not login.isdecimal() or int(login) <= 0 or not server:
        raise ValueError('Укажите положительный номер счёта и имя сервера либо очистите реквизиты.')
    if complete and not password:
        raise ValueError('Для первого подключения укажите пароль счёта.')
    return int(login), server, password


def graceful_close(process: psutil.Process, timeout: int = 30) -> None:
    '''WM_CLOSE даёт терминалу возможность сохранить параметры счёта.'''
    if os.name != 'nt':
        raise RuntimeError('Штатное закрытие MetaTrader реализовано для Windows.')
    from ctypes import wintypes
    user32 = ctypes.WinDLL('user32', use_last_error=True)
    callback_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)
    user32.EnumWindows.argtypes = [callback_type, wintypes.LPARAM]
    user32.GetWindowThreadProcessId.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.DWORD)]
    user32.PostMessageW.argtypes = [wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM]
    user32.GetWindow.argtypes = [wintypes.HWND, wintypes.UINT]
    user32.GetWindow.restype = wintypes.HWND
    user32.IsWindowVisible.argtypes = [wintypes.HWND]
    user32.GetWindowRect.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.RECT)]
    candidates = []

    @callback_type
    def close_window(hwnd, _):
        pid = wintypes.DWORD()
        user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
        if (pid.value == process.pid and not user32.GetWindow(hwnd, 4)
                and user32.IsWindowVisible(hwnd)):
            rect = wintypes.RECT()
            user32.GetWindowRect(hwnd, ctypes.byref(rect))
            candidates.append(((rect.right - rect.left) * (rect.bottom - rect.top), hwnd))
        return True

    user32.EnumWindows(close_window, 0)
    if not candidates:
        raise RuntimeError('Не найдено окно для штатного закрытия терминала.')
    # Только главное видимое окно, а не скрытые служебные окна процесса.
    hwnd = max(candidates, key=lambda item: item[0])[1]
    if not user32.PostMessageW(hwnd, 0x0010, 0, 0):  # WM_CLOSE
        raise RuntimeError('Не удалось запросить штатное закрытие терминала.')
    try:
        process.wait(timeout)
    except psutil.TimeoutExpired:
        raise RuntimeError('Терминал не закрылся штатно. Проверьте его диалоговые окна.') from None


class TerminalManager:
    '''Один рабочий поток последовательно выполняет длительные операции.

    Короткие запросы состояния продолжают обслуживаться во время ожидания.
    Пароли не копируются в задачи, карточки и ответы настроек.
    '''

    def __init__(self, store: ConfigStore):
        self.store = store
        self.lock = threading.RLock()
        self.pool = ThreadPoolExecutor(max_workers=1, thread_name_prefix='terminals')
        self.jobs = {}
        self.closing = threading.Event()
        self.verified_accounts = {}
        self.updates = ProjectUpdates(PROJECT_DIR)
        self.control = MT5_Control(store.data['terminals'], store.data['mt5_folder'],
                                   store.data['database'], store.data['mt5_exe'])

    def close(self):
        self.closing.set()
        self.pool.shutdown(wait=True, cancel_futures=True)

    def reload_control(self):
        data = self.store.data
        self.control.configure(data['terminals'], data['mt5_folder'], data['database'])

    def busy(self):
        return any(job['status'] in ('queued', 'running') for job in self.jobs.values())

    def snapshot(self):
        with self.lock:
            self.control.load_instances()
            return copy.deepcopy(self.control.instances)

    def info(self, folder):
        with self.lock:
            self.control.find_instances()
            return copy.deepcopy(self.control.instance_info(folder))

    def suggest_name(self):
        with self.lock:
            root = Path(self.store.data['mt5_folder'])
            names = {name.casefold() for name in self.store.data['terminals']}
            if root.exists():
                names.update(child.name.casefold() for child in root.iterdir())
            index = 1
            while f'MetaTrader5.{index}'.casefold() in names:
                index += 1
            return f'MetaTrader5.{index}'

    def template_source(self, kind):
        return Path(self.store.data['terminal_template']) / UPLOAD_NAMES[kind]

    def service_source(self):
        return service_build_paths(PROJECT_DIR, self.store.data)[2]

    def prepare_service(self):
        if not compile_service(*service_build_paths(PROJECT_DIR, self.store.data)):
            raise RuntimeError('Не удалось скомпилировать ServiceStateWriter.mq5. '
                               'Проверьте MetaEditor64.exe в папке исходной установки MT5 и журнал компиляции.')
        return self.service_source()

    def settings(self):
        with self.lock:
            data = copy.deepcopy(self.store.data)
            for folder, record in data['terminals'].items():
                record['password_set'] = bool(record.pop('password', ''))
                # Вычисленный путь нужен только для отображения в интерфейсе.
                record['path'] = str(instance_directory(data['mt5_folder'], folder))
                record['running'] = bool(self.processes(folder))
            files = {}
            for kind, relative in UPLOAD_NAMES.items():
                path = self.template_source(kind)
                files[kind] = {'name': Path(relative).name, 'exists': path.is_file()}
                if path.is_file():
                    stat = path.stat()
                    files[kind].update(size=stat.st_size, modified=int(stat.st_mtime))
            return {'config': data, 'files': files, 'suggested_name': self.suggest_name(),
                    'base_folder_lock': self.base_folder_lock()}

    def base_folder_lock(self):
        if PROJECT_DIR.resolve().is_relative_to(Path(self.store.data['mt5_folder'])):
            return ('Сейчас изменение базовой папки невозможно: менеджер запущен из папки внутри неё. '
                    'Для переноса терминалов разместите проект менеджера вне базовой папки и запустите его оттуда.')
        return ''

    def settings_candidate(self, values):
        data = copy.deepcopy(self.store.data)
        for key in ('mt5_folder', 'terminal_template', 'startup_timeout_sec', 'disk_warning_mb'):
            if key in values:
                data[key] = values[key]
        return self.store._validate(data)

    def settings_move_plan(self, values, *, during_job=False):
        with self.lock:
            if self.busy() and not during_job:
                raise ValueError('Дождитесь завершения операций с терминалами.')
            data = self.settings_candidate(values)
            source = Path(self.store.data['mt5_folder'])
            target = Path(data['mt5_folder'])
            folders, missing, running = [], [], []
            if source != target:
                if self.base_folder_lock():
                    raise ValueError(self.base_folder_lock())
                if target.exists() and not target.is_dir():
                    raise ValueError('По новому пути базовой папки находится файл.')
                for folder in data['terminals']:
                    old = instance_directory(source, folder)
                    new = instance_directory(target, folder)
                    if old.is_symlink() or old.is_junction() or old.resolve() != old:
                        raise ValueError(f'{folder}: перенос ссылки вместо папки терминала не поддерживается.')
                    if not old.exists():
                        missing.append(folder)
                        continue
                    if not old.is_dir():
                        raise ValueError(f'{folder}: по пути экземпляра находится файл.')
                    if target.is_relative_to(old):
                        raise ValueError('Новая базовая папка не может находиться внутри перемещаемого экземпляра.')
                    if new.exists() or new.is_symlink() or new.is_junction():
                        raise ValueError(f'Папка назначения уже занята: {new}. Объединение папок не выполняется.')
                    for path in (PROJECT_DIR, self.store.path, Path(data['database'])):
                        if path.resolve().is_relative_to(old):
                            raise ValueError(f'{folder}: внутри находится проект, конфигурация или база менеджера. '
                                             'Перед переносом разместите менеджер и его данные вне папок экземпляров.')
                    folders.append(folder)
                    if self.processes(folder):
                        running.append(folder)
            identity = json.dumps([str(source), str(target), folders, missing])
            return {'required': bool(folders), 'source': str(source), 'destination': str(target),
                    'folders': folders, 'missing': missing, 'running': running,
                    'token': hashlib.sha256(identity.encode()).hexdigest()}

    def save_settings(self, values, *, move_token='', stop_running=False):
        with self.lock:
            plan = self.settings_move_plan(values)
            if plan['required']:
                if move_token != plan['token']:
                    raise ValueError('Подтвердите перенос папок терминалов при изменении базовой папки.')
                if plan['running'] and not stop_running:
                    raise ValueError('Перед переносом нужно остановить все терминалы. Подтвердите их остановку.')
                return self.submit('move_settings', '',
                                   {'values': values, 'token': move_token, 'stop_running': stop_running})
            data = self.settings_candidate(values)
            old_root = self.store.data['mt5_folder']
            self.store.save(data)
            self.reload_control()
            if data['mt5_folder'] != old_root:
                self.verified_accounts.clear()

    @staticmethod
    def remove_move_temporary(path, root):
        # Удаляем только собственную временную папку в проверенном родителе.
        if (path.parent != root or path.resolve() != path or path.is_symlink() or path.is_junction()
                or not path.name.startswith('.mt5-manager-move-')):
            raise ValueError(f'Небезопасный путь временной папки переноса: {path}')
        if path.exists():
            shutil.rmtree(path)

    def move_settings(self, record, job_id):
        # Повторная проверка после очереди: пути могли измениться вне менеджера.
        with self.lock:
            plan = self.settings_move_plan(record['values'], during_job=True)
            data = self.settings_candidate(record['values'])
            source = Path(self.store.data['mt5_folder'])
        target = Path(data['mt5_folder'])
        folders, missing = plan['folders'], plan['missing']
        if plan['token'] != record['token']:
            raise ValueError('Список папок изменился. Повторите сохранение и подтверждение переноса.')
        for folder in data['terminals']:
            if self.processes(folder):
                if not record['stop_running']:
                    raise ValueError('Терминал был запущен после подтверждения. Повторите сохранение настроек.')
                self.stage(job_id, 'Штатная остановка: ' + folder)
                self.stop(folder, graceful=True)
        if any(self.processes(folder) for folder in data['terminals']):
            raise RuntimeError('Не все терминалы остановлены. Базовая папка не изменена.')
        moved = []
        current = None
        try:
            if target.resolve() != target:
                raise ValueError('Путь новой базовой папки изменился после подтверждения.')
            target.mkdir(parents=True, exist_ok=True)
            for folder in folders:
                old, new = instance_directory(source, folder), instance_directory(target, folder)
                # Проверяем абсолютные пути непосредственно перед перемещением.
                if (old.resolve() != old or old.is_symlink() or old.is_junction()
                        or not old.is_dir() or new.exists() or new.is_symlink() or new.is_junction()
                        or target.is_relative_to(old) or self.processes(folder)):
                    raise ValueError(f'{folder}: папки или состояние терминала изменились. Перенос отменён.')
                self.stage(job_id, 'Перенос: ' + folder)
                try:
                    old.rename(new)
                    moved.append([old, new, None, True])
                except OSError as error:
                    if error.errno != errno.EXDEV and getattr(error, 'winerror', None) != 17:
                        raise
                    # Между дисками сначала копируем; оригинал держим до сохранения JSON.
                    temporary = target / ('.mt5-manager-move-' + uuid.uuid4().hex)
                    current = temporary
                    shutil.copytree(old, temporary, symlinks=True)
                    backup = source / ('.mt5-manager-move-' + uuid.uuid4().hex)
                    old.rename(backup)
                    moved.append([old, new, backup, False])
                    temporary.rename(new)
                    moved[-1][3] = True
                    current = None
                template = Path(data['terminal_template'])
                if template.is_relative_to(old):
                    data['terminal_template'] = str(new / template.relative_to(old))
            self.stage(job_id, 'Сохранение новой базовой папки')
            with self.lock:
                self.store.save(data)
        except Exception as error:
            failures = []
            for old, new, backup, published in reversed(moved):
                try:
                    if backup is None:
                        new.rename(old)
                    else:
                        backup.rename(old)
                        if published:
                            cleanup = target / ('.mt5-manager-move-' + uuid.uuid4().hex)
                            new.rename(cleanup)
                            self.remove_move_temporary(cleanup, target)
                except (OSError, ValueError):
                    failures.append(str(old) + ' ↔ ' + str(backup or new))
            if current:
                try:
                    self.remove_move_temporary(current, target)
                except (OSError, ValueError):
                    failures.append(str(current))
            if failures:
                raise RuntimeError('Перенос не завершён, конфигурация осталась прежней. '
                                   'Проверьте оставшиеся папки: ' + '; '.join(failures)) from error
            raise RuntimeError('Перенос отменён, папки возвращены на место, конфигурация не изменена. '
                               'Терминалы остаются остановленными. Причина: ' + str(error)) from error
        with self.lock:
            self.reload_control()
            self.verified_accounts.clear()
            self.jobs[job_id]['result'] = {'moved': folders, 'missing': missing, 'mt5_folder': str(target)}
        for old, new, backup, published in moved:
            if backup:
                try:
                    self.remove_move_temporary(backup, source)
                except (OSError, ValueError):
                    self.jobs[job_id]['warnings'].append('Перенос выполнен, но осталась резервная папка: ' + str(backup))
        if missing:
            self.jobs[job_id]['warnings'].append('Не найдены в исходной папке: ' + ', '.join(missing))

    def upload_path(self, kind):
        if kind not in UPLOAD_NAMES:
            raise ValueError('Неизвестный тип файла.')
        return Path(self.store.data['terminal_template']) / UPLOAD_NAMES[kind]

    def install_upload(self, kind, source):
        with self.lock:
            if self.busy():
                raise ValueError('Дождитесь завершения операций с терминалами.')
            target = self.upload_path(kind)
            if target.exists():
                raise ValueError('Файл уже существует в исходной установке MT5. Загрузка нужна только для отсутствующих EXE.')
            with open(source, 'rb') as file:
                if kind in ('terminal', 'editor') and file.read(2) != b'MZ':
                    raise ValueError('Файл не похож на исполняемый файл Windows.')
            target.parent.mkdir(parents=True, exist_ok=True)
            temporary = target.with_name(target.name + '.' + uuid.uuid4().hex + '.tmp')
            try:
                shutil.copyfile(source, temporary)
                os.replace(temporary, target)
            finally:
                temporary.unlink(missing_ok=True)

    def submit(self, action, folder, record=None):
        with self.lock:
            if self.closing.is_set():
                raise ValueError('Приложение завершает работу.')
            project_actions = GLOBAL_ACTIONS
            if action not in ('create', 'start', 'stop', 'account', 'test_connection', 'delete', *project_actions):
                raise ValueError('Неизвестная операция.')
            project_action = action in project_actions
            if project_action:
                if self.busy():
                    raise ValueError('Дождитесь завершения текущих операций.')
                folder = ''
            elif any(j['action'] in project_actions and j['status'] in ('queued', 'running')
                     for j in self.jobs.values()):
                raise ValueError('Дождитесь завершения обновления проекта или сервиса.')
            if action == 'create':
                validate_folder(folder)
                account_values(record or {}, complete=True)
                servers = (record or {}).get('_servers_dat')
                if servers is not None and (not isinstance(servers, bytes) or not 0 < len(servers) <= MAX_SERVERS_SIZE):
                    raise ValueError('Выберите непустой servers.dat размером не более 16 МиБ либо создайте экземпляр без файла.')
            elif not project_action and folder not in self.store.data['terminals']:
                raise ValueError('Экземпляр не найден.')
            if any(j['folder'] == folder and j['status'] in ('queued', 'running')
                   for j in self.jobs.values()):
                raise ValueError('Операция с этим экземпляром уже выполняется.')
            if action == 'account':
                self.account_record(folder, record or {})
                self.validate_instance_edit(folder, record or {})
            if action == 'test_connection':
                self.connection_record(folder, record)
            if action == 'delete':
                self.deletion_directory(folder)
            job_id = uuid.uuid4().hex
            job = {'id': job_id, 'folder': folder, 'action': action,
                   'status': 'queued', 'stage': 'Ожидание', 'warnings': []}
            self.jobs[job_id] = job
            finished = [key for key, value in self.jobs.items()
                        if value['status'] in ('done', 'error')]
            for key in finished[:-50]:
                del self.jobs[key]
            self.pool.submit(self._run, job_id, record)
            return copy.deepcopy(job)

    def job(self, job_id):
        with self.lock:
            return copy.deepcopy(self.jobs[job_id])

    def stage(self, job_id, text):
        with self.lock:
            self.jobs[job_id]['stage'] = text

    def _run(self, job_id, record):
        job = self.jobs[job_id]
        with self.lock:
            job['status'] = 'running'
        try:
            if job['action'] == 'move_settings':
                self.move_settings(record, job_id)
            elif job['action'] in ('check_updates', 'update_project', 'deploy_service'):
                action = job['action']
                if action == 'check_updates':
                    self.stage(job_id, 'Проверка удалённого репозитория')
                    result = self.updates.check()
                elif action == 'update_project':
                    self.stage(job_id, 'Загрузка обновления проекта')
                    result = self.updates.update()
                    if result['updated']:
                        job['warnings'].append('Код проекта обновлён. Перезапустите менеджер для загрузки новой версии Python.')
                else:
                    self.stage(job_id, 'Компиляция и распространение сервиса')
                    result = self.updates.deploy(copy.deepcopy(self.store.data), self.processes)
                    if result['restart_required']:
                        job['warnings'].append('Для загрузки нового сервиса перезапустите: ' + ', '.join(result['restart_required']))
                    job['result'] = result
                    if result['failed']:
                        details = ' '.join(f'{folder}: {message}'
                                           for folder, message in result.get('errors', {}).items())
                        raise RuntimeError('Не удалось завершить развёртывание для: '
                                           + ', '.join(result['failed'])
                                           + f'. Успешно обработано: {len(result["copied"])}. '
                                           + details)
                job['result'] = result
            elif job['action'] == 'create':
                self.create_mt5(job['folder'], record or {}, job_id)
            elif job['action'] == 'stop':
                self.stage(job_id, 'Остановка')
                self.stop(job['folder'])
            elif job['action'] == 'account':
                folder = self.edit_instance(job['folder'], record or {}, job_id)
                job['result'] = {'folder': folder}
            elif job['action'] == 'test_connection':
                candidate = self.connection_record(job['folder'], record)
                self.test_connection(job['folder'], job_id, candidate)
                job['result'] = {'connected': True}
            elif job['action'] == 'delete':
                self.delete_instance(job['folder'], job_id)
            else:
                self.start(job['folder'], job_id)
            with self.lock:
                job.update(status='done', stage='Готово')
        except Exception as error:
            # Известные сообщения не содержат реквизитов. Не отдаём traceback
            # или представление исходного запроса в браузер.
            message = str(error) if isinstance(error, (ValueError, RuntimeError, OSError)) and not isinstance(error, ValidationError) else 'Внутренняя ошибка операции.'
            with self.lock:
                job.update(status='error', stage='Ошибка', error=message)

    def deletion_directory(self, folder):
        directory = instance_directory(Path(self.store.data['mt5_folder']).resolve(), folder)
        if self.processes(folder):
            raise ValueError('Терминал запущен. Остановите его перед удалением экземпляра.')
        if directory.is_symlink() or directory.is_junction() or directory.resolve() != directory:
            raise ValueError('Удаление ссылки или папки за пределами базовой папки запрещено.')
        if directory.exists() and not directory.is_dir():
            raise ValueError('По пути экземпляра находится файл вместо папки.')
        protected = ((PROJECT_DIR, 'В этой папке находится работающий проект менеджера.'),
                     (self.store.path, 'В этой папке находится конфигурация менеджера.'),
                     (Path(self.store.data['database']), 'В этой папке находится общая база менеджера.'),
                     (Path(self.store.data['terminal_template']),
                      'Эта папка используется как исходная установка MT5. Сначала выберите другую установку в настройках.'))
        for path, message in protected:
            if path.resolve().is_relative_to(directory):
                raise ValueError('Нельзя удалить экземпляр. ' + message)
        return directory

    def delete_instance(self, folder, job_id):
        self.stage(job_id, 'Подготовка удаления экземпляра')
        with self.lock:
            directory = self.deletion_directory(folder)
            root = directory.parent
            pending = root / ('.mt5-manager-delete-' + uuid.uuid4().hex)
            moved = False
            try:
                if directory.exists():
                    directory.rename(pending)
                    moved = True
                data = copy.deepcopy(self.store.data)
                del data['terminals'][folder]
                self.store.save(data)
            except Exception as error:
                if moved:
                    try:
                        pending.rename(directory)
                    except OSError:
                        raise RuntimeError('Запись экземпляра не удалена, но вернуть имя папки не удалось. '
                                           f'Проверьте папку: {pending}.') from error
                raise
            self.verified_accounts.pop(folder, None)
            self.reload_control()
            self.jobs[job_id]['result'] = {'folder': folder, 'deleted': True}
        if not moved:
            self.jobs[job_id]['warnings'].append('Папка экземпляра отсутствовала; удалена запись из конфигурации.')
            return
        self.stage(job_id, 'Удаление файлов экземпляра')
        try:
            # Окончательная проверка непосредственно перед рекурсивным удалением.
            if (pending.is_symlink() or pending.is_junction() or pending.resolve() != pending
                    or pending.parent != root or not pending.name.startswith('.mt5-manager-delete-')):
                raise ValueError('Путь папки изменился; удаление файлов остановлено.')
            shutil.rmtree(pending)
        except (OSError, ValueError):
            with self.lock:
                self.jobs[job_id]['result']['remaining_path'] = str(pending)
                self.jobs[job_id]['warnings'].append(
                    'Запись экземпляра удалена, но не все файлы удалось удалить. '
                    f'Оставшиеся файлы: {pending}.')

    def account_record(self, folder, changes):
        result = copy.deepcopy(self.store.data['terminals'][folder])
        result['name'] = str(changes.get('name', result.get('name', folder)))
        if not result['name'] or result['name'] == folder:
            result['name'] = changes.get('folder') or folder
        result['login'] = changes.get('login', '')
        result['server'] = changes.get('server', '')
        if changes.get('clear_account'):
            result.update(login='', server='', password='')
        elif changes.get('password'):
            result['password'] = changes['password']
        login, server, password = account_values(result)
        result.update(login=login, server=server, password=password)
        return result

    def validate_instance_edit(self, folder, changes):
        name = changes.get('folder')
        name = folder if name is None else validate_folder(name)
        servers = changes.get('_servers_dat')
        if servers is not None and (not isinstance(servers, bytes) or not 0 < len(servers) <= MAX_SERVERS_SIZE):
            raise ValueError('Выберите непустой servers.dat размером не более 16 МиБ.')
        if servers is not None and self.processes(folder):
            raise ValueError('Закройте терминал перед заменой servers.dat. Сохранение настроек не останавливает терминал.')
        if name != folder:
            if self.processes(folder):
                raise ValueError('Терминал запущен, переименование папки пока недоступно. Закройте терминал, чтобы переименовать папку.')
            data = self.store.data
            if any(key != folder and key.casefold() == name.casefold() for key in data['terminals']):
                raise ValueError('Экземпляр с таким именем уже существует.')
            root = Path(data['mt5_folder']).resolve()
            source, target = root / folder, root / name
            if not source.is_dir() or source.resolve() != source:
                raise ValueError('Не найдена обычная папка экземпляра внутри базовой папки.')
            if target.exists() and not target.samefile(source):
                raise ValueError('Папка с таким именем уже существует.')
        return name

    def edit_instance(self, folder, changes, job_id):
        # Сохранение настроек не запускает, не останавливает и не авторизует MT5.
        servers = changes.get('_servers_dat')
        with self.lock:
            name = self.validate_instance_edit(folder, changes)
            data = copy.deepcopy(self.store.data)
            updated = self.account_record(folder, changes)
            source = instance_directory(data['mt5_folder'], folder)
            target = instance_directory(data['mt5_folder'], name)
            renamed = False
            replaced = False
            previous = None
            temporary = None
            server_path = target / 'config/servers.dat'
            try:
                if name != folder:
                    self.stage(job_id, 'Переименование папки экземпляра')
                    source.rename(target)
                    renamed = True
                    template = Path(data['terminal_template'])
                    if template.is_relative_to(source):
                        data['terminal_template'] = str(target / template.relative_to(source))
                if servers is not None:
                    self.stage(job_id, 'Обновление servers.dat')
                    server_path.parent.mkdir(parents=True, exist_ok=True)
                    previous = server_path.read_bytes() if server_path.exists() else None
                    temporary = server_path.with_name('servers.' + uuid.uuid4().hex + '.tmp')
                    temporary.write_bytes(servers)
                    os.replace(temporary, server_path)
                    replaced = True
                data['terminals'] = {name if key == folder else key: updated if key == folder else value
                                     for key, value in data['terminals'].items()}
                self.store.save(data)
            except Exception:
                if replaced:
                    if previous is None:
                        server_path.unlink()
                    else:
                        temporary.write_bytes(previous)
                        os.replace(temporary, server_path)
                if renamed:
                    target.rename(source)
                raise
            finally:
                if temporary:
                    temporary.unlink(missing_ok=True)
            if renamed or servers is not None:
                self.verified_accounts.pop(folder, None)
            self.reload_control()
            return name

    def connection_record(self, folder, changes):
        if changes is None:
            candidate = copy.deepcopy(self.store.data['terminals'][folder])
        else:
            if changes.get('folder') not in (None, folder):
                raise ValueError('Сначала сохраните новое имя папки, затем проверьте подключение.')
            candidate = self.account_record(folder, changes)
        if not account_values(candidate)[0]:
            raise ValueError('Для проверки подключения укажите логин и торговый сервер.')
        return candidate

    def processes(self, folder):
        expected = self.control.normalized_path(self.control.instance_path(folder))
        result = []
        for process in psutil.process_iter(['exe']):
            try:
                if process.info['exe'] and self.control.normalized_path(process.info['exe']) == expected:
                    result.append(process)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        return result

    def stop(self, folder, *, graceful=False):
        for process in self.processes(folder):
            try:
                if graceful:
                    graceful_close(process)
                else:
                    process.terminate()
                    process.wait(10)
            except psutil.NoSuchProcess:
                pass

    def launch(self, folder, config=None):
        args = [self.control.instance_path(folder), '/portable']
        if config:
            # MT5 разбирает ключ сам: при пробелах кавычки должны окружать
            # значение после двоеточия, а не весь аргумент /config:путь.
            command = subprocess.list2cmdline(args) + f' /config:"{config}"'
            return subprocess.Popen(command, cwd=str(Path(args[0]).parent))
        return subprocess.Popen(args, cwd=str(Path(args[0]).parent))

    @staticmethod
    def prepare_account_defaults(directory, account):
        '''Сохраняет выбор счёта и KeepPrivate в основной конфигурации.

        Параметры /config не переносят выбор счёта в common.ini, поэтому
        отдельно задаём логин и сервер для последующего обычного запуска.
        Пароль передаётся только временным файлом.
        Вызывается только при остановленном терминале.
        '''
        path = directory / 'config/common.ini'
        path.parent.mkdir(parents=True, exist_ok=True)
        text = path.read_text(encoding='utf-16') if path.exists() else ''
        section = re.search(r'(?ims)^\[Common\][^\r\n]*\r?\n(.*?)(?=^\[|\Z)', text)
        body = section.group(1) if section else ''
        body = re.sub(r'(?im)^[ \t]*Password[ \t]*=.*\n?', '', body)
        for key, value in {'Login': account[0], 'Server': account[1], 'KeepPrivate': 1}.items():
            pattern = rf'(?im)^[ \t]*{key}[ \t]*=.*$'
            line = f'{key}={value}'
            if re.search(pattern, body):
                body = re.sub(pattern, lambda _: line, body)
            else:
                body = line + '\n' + body
        if section:
            text = text[:section.start(1)] + body + text[section.end(1):]
        else:
            text = '[Common]\n' + body + text
        temporary = path.with_name('common.' + uuid.uuid4().hex + '.tmp')
        try:
            temporary.write_text(text, encoding='utf-16')
            os.replace(temporary, path)
        finally:
            temporary.unlink(missing_ok=True)

    def wait_state(self, folder, since, account=None):
        deadline = time.monotonic() + self.store.data['startup_timeout_sec']
        directory = self.control.normalized_path(str(Path(self.control.instance_path(folder)).parent))
        while time.monotonic() < deadline:
            if self.closing.is_set():
                raise RuntimeError('Операция прервана завершением приложения.')
            row = self.control.read_state(folder)
            fresh = (row is not None and row['updated_at'] is not None
                     and row['updated_at'] >= since
                     and time.time() - row['updated_at'] < self.control.FRESHNESS_THRESHOLD_SEC
                     and self.control.normalized_path(row['terminal_path']) == directory)
            if fresh and self.processes(folder):
                if account is None or (row['connected'] and row['login'] == account[0]
                                       and row['server'] == account[1]):
                    return row
            # При LiveUpdate процесс может смениться; не считаем этот разрыв
            # немедленной ошибкой и ждём до общего срока.
            self.closing.wait(1)
        if account:
            raise RuntimeError('Не подтверждено подключение к заданному счёту. Проверьте реквизиты и наличие сервера в servers.dat.')
        raise RuntimeError('Нет свежих данных сервиса. Проверьте запуск терминала, журнал сервиса '
                           f'и совпадение его базы с database={self.store.data["database"]}.')

    def test_connection(self, folder, job_id, record):
        was_running = bool(self.processes(folder))
        try:
            self.start(folder, job_id, record)
        finally:
            self.stage(job_id, 'Восстановление состояния терминала')
            try:
                if was_running:
                    if not self.processes(folder):
                        self.launch(folder)
                elif self.processes(folder):
                    try:
                        self.stop(folder, graceful=True)
                    except Exception:
                        # При неудачном штатном закрытии всё равно завершаем
                        # процесс, который был запущен только для проверки.
                        self.stop(folder)
                if bool(self.processes(folder)) != was_running:
                    raise RuntimeError('Состояние процесса не восстановлено.')
            except Exception:
                state = 'запущенное' if was_running else 'остановленное'
                raise RuntimeError(f'После проверки не удалось вернуть терминал в {state} состояние. '
                                   'Проверьте его состояние вручную.') from None

    def start(self, folder, job_id, record=None):
        if record is None:
            record = self.store.data['terminals'][folder]
        account = account_values(record)
        desired = account if account[0] else None
        complete = bool(desired and desired[2])
        # После перезапуска менеджера один раз подтверждаем полные реквизиты
        # заново: пароль невозможно проверить по login/server в states.
        needs_auth = complete and self.verified_accounts.get(folder) != account
        if not needs_auth:
            self.stage(job_id, 'Запуск и проверка состояния')
            since = time.time()
            if not self.processes(folder):
                self.launch(folder)
            row = self.wait_state(folder, since)
            if desired is None or (row['connected'] and row['login'] == desired[0]
                                   and row['server'] == desired[1]):
                return
            if not complete:
                raise RuntimeError('Терминал не подключился к заданному счёту. Укажите пароль в настройках экземпляра.')
        self.stage(job_id, 'Настройка подключения к счёту')
        if self.processes(folder):
            self.stop(folder, graceful=True)
        directory = Path(self.control.instance_path(folder)).parent
        self.prepare_account_defaults(directory, account)
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(mode='w', encoding='utf-16',
                                             dir=directory, prefix='manager-login-',
                                             suffix='.ini', delete=False) as out:
                temporary = Path(out.name)
                out.write(f'[Common]\nLogin={account[0]}\nServer={account[1]}\n'
                          f'Password={account[2]}\nKeepPrivate=1\n')
            since = time.time()
            self.launch(folder, temporary)
            self.wait_state(folder, since, account)
            self.stage(job_id, 'Сохранение параметров счёта')
            self.stop(folder, graceful=True)
        except Exception:
            self.stop(folder)
            raise
        finally:
            if temporary:
                temporary.unlink(missing_ok=True)
        self.stage(job_id, 'Повторный запуск без файла авторизации')
        since = time.time()
        self.launch(folder)
        self.wait_state(folder, since, account)
        self.verified_accounts[folder] = account

    def create_mt5(self, folder, record, job_id):
        self.stage(job_id, 'Проверка и копирование файлов')
        with self.lock:
            data = self.store.data
            root = Path(data['mt5_folder']).resolve()
            root.mkdir(parents=True, exist_ok=True)
            target = instance_directory(root, folder)
            occupied = {name.casefold() for name in data['terminals']}
            occupied.update(child.name.casefold() for child in root.iterdir())
            if folder.casefold() in occupied:
                raise ValueError('Экземпляр или папка с таким именем уже существует.')
            sources = {UPLOAD_NAMES[kind]: self.template_source(kind)
                       for kind in ('terminal', 'editor')}
        for source in sources.values():
            if not source.is_file() or not source.stat().st_size:
                raise ValueError(f'Не найден исходный файл: {source.name}. Загрузите его в настройках.')
        # MetaEditor может работать долго: не блокируем запросы состояния.
        self.stage(job_id, 'Подготовка сервиса')
        sources[str(SERVICE_REL)] = self.prepare_service()
        for source in sources.values():
            if not source.is_file() or not source.stat().st_size:
                raise ValueError(f'Не найден исходный файл: {source.name}. Загрузите его в настройках.')
        servers = record.get('_servers_dat')
        required = sum(path.stat().st_size for path in sources.values()) + (len(servers) if servers is not None else 0)
        free = shutil.disk_usage(root).free
        if free < required:
            raise ValueError('На диске недостаточно места даже для копирования исходных файлов.')
        if free - required < data['disk_warning_mb'] * 1024 * 1024:
            with self.lock:
                self.jobs[job_id]['warnings'].append('После копирования останется меньше заданного запаса места. История и журналы будут занимать дополнительное место.')
        self.stage(job_id, 'Копирование файлов')
        # Только mkdir без exist_ok: чужую папку нельзя принять за свою
        # и затем удалить при откате.
        target.mkdir()
        registered = False
        try:
            for relative, source in sources.items():
                destination = target / relative
                destination.parent.mkdir(parents=True, exist_ok=True)
                shutil.copyfile(source, destination)
            (target / 'config').mkdir(parents=True, exist_ok=True)
            if servers is not None:
                (target / 'config/servers.dat').write_bytes(servers)
            if data['mt5_exe'] != 'terminal64.exe':
                raise ValueError('Создание экземпляров поддерживает имя terminal64.exe.')
            ini_path, ini_text, _ = services_ini_update(target, PROJECT_DIR / 'static/services.ini', data['database'])
            write_services_ini(ini_path, ini_text)
            login, server, password = account_values(record, complete=True)
            with self.lock:
                updated = copy.deepcopy(self.store.data)
                updated['terminals'][folder] = {'name': record.get('name') or folder,
                                               'login': login,
                                               'server': server, 'password': password}
                self.store.save(updated)
                registered = True
                self.reload_control()
            self.start(folder, job_id)
        except Exception as error:
            self.stage(job_id, 'Откат создания')
            try:
                if registered:
                    self.stop(folder)
                # Первый запуск может оставить отдельный процесс компилятора.
                for process in psutil.process_iter(['exe']):
                    try:
                        executable = process.info['exe']
                        if executable and Path(executable).resolve().is_relative_to(target.resolve()):
                            process.terminate()
                            process.wait(10)
                    except (psutil.NoSuchProcess, psutil.AccessDenied):
                        continue
                # Сверяем окончательный путь перед рекурсивным удалением.
                if target.is_symlink() or target.resolve() != target or target.resolve().parent != root:
                    raise RuntimeError('Путь папки изменился; автоматическое удаление остановлено.')
                shutil.rmtree(target)
                if registered:
                    with self.lock:
                        updated = copy.deepcopy(self.store.data)
                        updated['terminals'].pop(folder, None)
                        self.store.save(updated)
                        self.reload_control()
                self.verified_accounts.pop(folder, None)
            except Exception:
                raise RuntimeError('Создание не завершено, автоматический откат не удался. Проверьте папку экземпляра и конфигурацию вручную.') from error
            raise