'''Создание экземпляров и настройка счёта без 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