mt5-manager/terminal_manager.py

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

__version__ = '0.4.2'

import copy
import ctypes
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, reference_terminal, 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',
                'servers': 'config/servers.dat', 'service': 'ServiceStateWriter.ex5'}


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 reference_terminal(self):
        return reference_terminal(PROJECT_DIR, self.store.data['mt5_installation'])

    def template_source(self, kind):
        relative = UPLOAD_NAMES[kind]
        uploaded = Path(self.store.data['terminal_template']) / relative
        reference = self.reference_terminal()
        return uploaded if uploaded.is_file() or reference is None else reference / relative

    def service_source(self):
        configured = self.store.data['service_ex5']
        if configured:
            return Path(configured)
        uploaded = Path(self.store.data['terminal_template']) / UPLOAD_NAMES['service']
        return uploaded if uploaded.is_file() else service_build_paths(PROJECT_DIR, self.store.data)[2]

    def prepare_service(self):
        # Готовый EX5, указанный пользователем, не перекомпилируем.
        uploaded = Path(self.store.data['terminal_template']) / UPLOAD_NAMES['service']
        if not self.store.data['service_ex5'] and not uploaded.is_file():
            if not compile_service(*service_build_paths(PROJECT_DIR, self.store.data)):
                raise RuntimeError('Не удалось скомпилировать ServiceStateWriter.mq5. '
                                   'Соберите сервис в MetaEditor или загрузите готовый EX5 в настройках.')
        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))
            files = {}
            for kind, relative in UPLOAD_NAMES.items():
                path = self.service_source() if kind == 'service' else 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()}

    def save_settings(self, values):
        with self.lock:
            if self.busy():
                raise ValueError('Дождитесь завершения операций с терминалами.')
            data = copy.deepcopy(self.store.data)
            old_root = data['mt5_folder']
            for key in ('mt5_folder', 'terminal_template', 'mt5_installation', 'service_ex5',
                        'startup_timeout_sec', 'disk_warning_mb'):
                if key in values:
                    data[key] = values[key]
            self.store.save(data)
            self.reload_control()
            if self.store.data['mt5_folder'] != old_root:
                self.verified_accounts.clear()

    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)
            # Папка-образец отделена от установленных экземпляров.
            for folder in self.store.data['terminals']:
                directory = instance_directory(self.store.data['mt5_folder'], folder).resolve()
                protected = [directory / 'terminal64.exe', directory / 'MetaEditor64.exe',
                             directory / 'config/servers.dat', directory / SERVICE_REL]
                if target.resolve() in protected:
                    raise ValueError('Загрузка не должна заменять рабочие файлы установленного терминала.')
            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 = ('check_updates', 'update_project', 'deploy_service')
            if action not in ('create', 'start', 'stop', 'account', *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)
            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 {})
            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'] 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'])
            else:
                if job['action'] == 'account':
                    with self.lock:
                        updated = self.account_record(job['folder'], record or {})
                        data = copy.deepcopy(self.store.data)
                        data['terminals'][job['folder']] = updated
                        self.store.save(data)
                        self.reload_control()
                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 account_record(self, folder, changes):
        result = copy.deepcopy(self.store.data['terminals'][folder])
        result['name'] = str(changes.get('name', result.get('name', 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 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 start(self, folder, job_id):
        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', 'servers')}
        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}. Загрузите его в настройках.')
        required = sum(path.stat().st_size for path in sources.values())
        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)
            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