mt5-manager/mt5_control.py

'''
File: mt5_control.py
Description: Логика управления запуском и остановкой терминалов MetaTrader 5
'''

__version__ = '0.5.1'

import copy
import os
import sqlite3
import subprocess
import time
from pathlib import Path

import psutil
from config import instance_directory


class MT5_Control:
    # Сколько секунд с последней записи в states считается "свежими" данными.
    # Сервис пишет раз в 2-3 секунды, порог даёт запас на редкие пропуски.
    FRESHNESS_THRESHOLD_SEC = 30

    STATE_QUERY = '''
        SELECT
            t.path AS terminal_path,
            t.company AS terminal_company,
            t.build AS terminal_build,
            s.updated_at, s.connected, s.ping_last,
            s.login, s.server, s.currency, s.leverage,
            s.trade_mode, s.trade_allowed,
            s.balance, s.equity, s.profit, s.margin, s.margin_free,
            s.margin_level, s.credit, s.positions,
            s.company AS account_company, s.holder_name,
            s.error_code, s.error_text
        FROM terminals t
        LEFT JOIN states s ON s.terminal_id = t.id
        WHERE t.folder = ?
    '''

    def __init__(self, terminals: dict, mt5_folder: str, database: str,
                 mt5_exe: str = 'terminal64.exe'):
        '''
        Инициализирует экземпляр класса MT5_Control.

        Args:
            terminals (dict): Словарь с информацией о терминалах.
            mt5_folder (str): Путь к папке с экземплярами терминалов.
            database (str): Путь к общей базе SQLite с состоянием терминалов.
            mt5_exe (str): Имя исполняемого файла терминала (по умолчанию 'terminal64.exe').
        '''
        # Путь к папке с экземплярами терминалов
        self.mt5_folder = mt5_folder

        # Путь к общей базе SQLite
        self.database = database

        # Имя исполняемого файла терминала
        self.mt5_exe = mt5_exe
        self.configure(terminals, mt5_folder, database)

    @staticmethod
    def normalized_path(path: str) -> str:
        return os.path.normcase(os.path.abspath(path))

    def configure(self, terminals: dict, mt5_folder: str, database: str) -> None:
        self.mt5_folder = mt5_folder
        self.database = database
        # Пароли остаются в ConfigStore; словари карточек содержат только
        # разрешённые для выдачи поля конфигурации и состояние процесса.
        self.terminal_config = {
            folder: {**{key: value for key, value in record.items()
                        if key in ('name', 'login', 'server')},
                     'path': str(instance_directory(mt5_folder, folder))}
            for folder, record in terminals.items()
        }

        # Список имён экземпляров
        self.instances_folders = list(terminals.keys())

        # Словарь соответствия имя -> путь
        self.instances_paths = {folder: self.instance_path(
            folder) for folder in self.instances_folders}

        # Словарь соответствия путь -> имя
        self.paths_instances = {self.normalized_path(self.instance_path(
            folder)): folder for folder in self.instances_folders}

        # Глубокая копия: instances хранит pid/status, не должен делить
        # вложенные словари с конфигурацией
        self.instances = copy.deepcopy(self.terminal_config)

    def instance_path(self, folder: str) -> str:
        '''
        Формирует полный путь по имени экземпляра терминала

        Args:
            folder (str): Имя экземпляра терминала.

        Returns:
            path (str): Полный путь к запускаемому файлу экземпляра терминала.
        '''
        return str(instance_directory(self.mt5_folder, folder) / self.mt5_exe)

    def read_state(self, folder: str) -> sqlite3.Row | None:
        '''
        Читает последнее известное состояние терминала из общей базы.

        Args:
            folder (str): Имя экземпляра терминала.

        Returns:
            row (sqlite3.Row | None): Строка состояния, или None, если
            терминал ещё не регистрировался в базе (сервис ни разу не
            запускался) либо база недоступна.
        '''
        database = Path(self.database).resolve()
        try:
            # Создание файла и таблиц — задача MQL5-сервиса. Ошибка пути
            # менеджера не должна создавать вторую пустую базу.
            con = sqlite3.connect(database.as_uri() + '?mode=ro', uri=True, timeout=3)
            con.row_factory = sqlite3.Row
            try:
                return con.execute(self.STATE_QUERY, (folder,)).fetchone()
            finally:
                con.close()
        except sqlite3.OperationalError as e:
            # База ещё не создана сервисом, временно занята, либо у неё
            # неожиданно не совпадает схема со STATE_QUERY (например, таблица
            # уже существовала до того, как в неё добавили колонки, а
            # CREATE TABLE IF NOT EXISTS не выполняет миграцию) - в любом из
            # этих случаев продолжаем без данных, но печатаем причину, чтобы
            # рассинхронизацию схемы не пришлось второй раз откапывать вручную
            message = str(e)
            if 'no such table:' in message:
                hint = (' В этой базе ещё нет схемы сервиса. Проверьте его запуск, '
                        'журнал и совпадение database с Common/Files/InpDatabase.')
            elif not database.is_file():
                hint = (' Файл базы ещё не создан сервисом или неверно задан database. '
                        'Менеджер не создаёт его при чтении.')
            else:
                hint = ''
            print(f'read_state({folder!r}): {message}; database={database}.{hint}')
            return None

    def instance_info(self, folder: str) -> dict:
        '''
        Получает информацию о терминале по имени экземпляра терминала.

        Статус процесса определяется через psutil, данные счёта и терминала -
        из общей базы SQLite, которую пишет сервис ServiceStateWriter.
        Совмещение даёт три состояния вместо простого running/stopped:
        процесс жив и данные свежие - "running"; процесс жив, но данные
        устарели или их ещё нет (терминал грузится, сервис не запущен или
        завис) - "starting"; процесса нет - "stopped".

        Args:
            folder (str): Имя экземпляра терминала.

        Returns:
            info (dict): Информация о терминале или обнаруженных ошибках.
        '''
        # Если терминала с таким именем нет в конфигурации
        if folder not in self.instances:
            return {
                'last_error': {'code': -1, 'description': 'Manager: Terminal not found'},
                'status': 'not found',
            }

        info = self.instances[folder]

        # Если для терминала нет запущенного процесса
        if not info.get('pid'):
            info['last_error'] = {'code': 0, 'description': 'Stopped'}
            info['status'] = 'stopped'
            return info

        row = self.read_state(folder)
        if row is not None:
            expected = os.path.dirname(self.instance_path(folder))
            if not row['terminal_path'] or self.normalized_path(row['terminal_path']) != self.normalized_path(expected):
                row = None
        fresh = (
            row is not None
            and row['updated_at'] is not None
            and (time.time() - row['updated_at']) < self.FRESHNESS_THRESHOLD_SEC
        )

        if row is not None:
            info['terminal'] = {
                'path': row['terminal_path'],
                'company': row['terminal_company'],
                'build': row['terminal_build'],
                'connected': bool(row['connected']),
                'ping_last': row['ping_last'],
            }

        if not fresh:
            # Терминал грузится, сервис ещё не написал данные, либо завис
            info['last_error'] = {'code': 0, 'description': 'Starting'}
            info['status'] = 'starting'
            return info

        info['account'] = {
            'login': row['login'],
            'server': row['server'],
            'currency': row['currency'],
            'leverage': row['leverage'],
            'trade_mode': row['trade_mode'],
            'trade_allowed': bool(row['trade_allowed']),
            'balance': row['balance'],
            'equity': row['equity'],
            'profit': row['profit'],
            'margin': row['margin'],
            'margin_free': row['margin_free'],
            'margin_level': row['margin_level'],
            'credit': row['credit'],
            'positions': row['positions'],
            'company': row['account_company'],
            'holder_name': row['holder_name'],
        }

        # Секунды unix: сервис пишет TimeGMT(), миллисекундная точность не нужна
        info['last_update'] = row['updated_at']
        info['last_error'] = {'code': row['error_code'], 'description': row['error_text']}
        info['status'] = 'running'

        return info

    def find_instances(self) -> None:
        '''Получение информации о статусе экземпляров
        терминала MetaTrader 5 из заданных папок.

        Returns:
            instances (dict): Информация об экземплярах терминала.
        '''

        # Формируем словарь для всех экземпляров по списку имён
        for folder in self.instances_folders:
            self.instances[folder]['pid'] = 0
            self.instances[folder]['status'] = 'stopped'

        # Перебираем все запущенные процессы
        for proc in psutil.process_iter(['pid', 'name', 'exe']):
            try:
                # Получаем путь к запускаемому файлу процесса
                exe_path = self.normalized_path(proc.info['exe']) if proc.info['exe'] else ''

                # Если он есть среди путей экземпляров терминала
                if exe_path and exe_path in self.paths_instances:
                    # Определяем имя экземпляра для данного пути
                    folder = self.paths_instances[exe_path]

                    # Берём нужный экземпляр
                    instance = self.instances[folder]

                    # Запоминаем идентификатор и статус процесса
                    instance['pid'] = proc.info['pid']
                    instance['status'] = 'starting'

            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue

    def load_instances(self) -> dict:
        self.find_instances()
        return self.instances

    def start_mt5(self, folder: str) -> dict:
        '''
        Запуск экземпляра терминала

        Args:
            folder (str): Имя экземпляра терминала.

        Returns:
            instance (dict): Информация о запущенном терминале.
        '''

        # Получаем информацию об экземплярах терминала
        instances = self.load_instances()

        # Если имя экземпляра есть среди имеющихся
        if folder in instances:
            # Берём нужный экземпляр
            instance = instances[folder]

            # Если для него нет запущенного процесса, то
            if not instance['pid']:
                # Запускаем новый процесс
                process = subprocess.Popen(
                    [self.instances_paths[folder], '/portable'])

                # Идентификатор запущенного процесса
                pid = process.pid

                # Запоминаем идентификатор и статус процесса
                instance['pid'] = pid
                instance['status'] = 'starting'

            # Возвращаем результат - успешный запуск
            return {folder: instance}

        # Возвращаем результат - имя не найдено
        return {folder: {'status': 'not found'}}

    def stop_mt5(self, folder: str) -> dict:
        '''
        Остановка терминала

        Args:
            folder (str): Имя экземпляра терминала.

        Returns:
            instance (dict): Информация о запущенном терминале.
        '''

        # Получаем информацию об экземплярах терминала
        instances = self.load_instances()

        # Если имя экземпляра есть среди имеющихся
        if folder in instances:
            # Берём нужный экземпляр
            instance = instances[folder]

            # Берём идентификатор его процесса
            pid = instance['pid']

            # Если ранее терминал был запущен
            if pid:
                try:
                    # Получаем объект процесса терминала
                    p = psutil.Process(pid)

                    # Останавливаем процесс
                    p.terminate()
                    p.wait(10)
                except psutil.NoSuchProcess:
                    pass

            # Удаляем информацию о запущенном ранее процессе
            instance['pid'] = 0
            instance['status'] = 'stopped'

            # Возвращаем результат - успешная остановка
            return {folder: instance}

        # Возвращаем результат - процесс не найден
        return {folder: {'status': 'not found'}}