mt5-manager/project_updates.py

'''Проверка Git и сборка сервиса для существующих экземпляров.'''

__version__ = '0.3.0'

import hashlib
import os
import shutil
import subprocess
import time
import uuid
from pathlib import Path

from config import instance_directory
from deploy_service import (SERVICE_EX5_REL, SERVICES_INI_REL, compile_service,
                            service_build_paths, services_ini_update, write_services_ini)


class ProjectUpdates:
    def __init__(self, project):
        self.project = Path(project).resolve()
        self.checked_at = None

    def git(self, *args):
        try:
            result = subprocess.run(['git', *args], cwd=self.project, capture_output=True,
                                    text=True, encoding='utf-8', errors='replace', timeout=120)
        except (OSError, subprocess.TimeoutExpired):
            raise RuntimeError('Git недоступен или операция превысила время ожидания.') from None
        if result.returncode:
            # stderr может содержать адрес с реквизитами доступа к репозиторию.
            raise RuntimeError(f'Не удалось выполнить git {args[0]} (код {result.returncode}). '
                               'Проверьте репозиторий и доступ к нему из командной строки.')
        return result.stdout.strip()

    def repository(self):
        root = Path(self.git('rev-parse', '--show-toplevel')).resolve()
        if root != self.project:
            raise RuntimeError('Папка проекта должна быть корнем отдельного Git-репозитория.')
        branch = self.git('symbolic-ref', '--short', 'HEAD')
        upstream = self.git('rev-parse', '--abbrev-ref', '--symbolic-full-name', '@{upstream}')
        remote = self.git('config', f'branch.{branch}.remote')
        if remote == '.':
            raise RuntimeError('Для ветки нужен upstream в удалённом репозитории.')
        ahead, behind = map(int, self.git('rev-list', '--left-right', '--count', 'HEAD...@{upstream}').split())
        return {'branch': branch, 'upstream': upstream, 'remote': remote,
                'revision': self.git('rev-parse', '--short', 'HEAD'),
                'dirty': bool(self.git('status', '--porcelain')),
                'ahead': ahead, 'behind': behind, 'checked_at': self.checked_at}

    def check(self):
        info = self.repository()
        self.git('fetch', '--', info['remote'])
        self.checked_at = int(time.time())
        return self.repository()

    def update(self):
        if os.getenv('MT5_MANAGER_RELOAD') == '1':
            raise RuntimeError('Перезапустите менеджер без --reload перед обновлением кода.')
        # Включаем неотслеживаемые файлы: не прячем работу пользователя в stash.
        if self.repository()['dirty']:
            raise RuntimeError('В проекте есть локальные изменения. Сохраните их в Git перед обновлением.')
        info = self.check()
        if info['ahead']:
            raise RuntimeError('В локальной ветке есть свои коммиты. Объедините изменения вручную.')
        if not info['behind']:
            return {'updated': False, 'revision': info['revision']}
        # Повторная проверка после сетевой операции защищает от обычной правки
        # файлов во время fetch; конфликты дополнительно проверит сам Git.
        if self.repository()['dirty']:
            raise RuntimeError('Во время проверки появились локальные изменения. Обновление отменено.')
        revision = self.git('rev-parse', '@{upstream}^{commit}')
        self.git('merge', '--ff-only', revision)
        return {'updated': True, 'revision': self.git('rev-parse', '--short', 'HEAD')}

    @staticmethod
    def digest(path):
        return hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None

    def service_status(self, data):
        editor, source, ex5 = service_build_paths(self.project, data)
        available = source.is_file() and editor.is_file()
        error = ('Не найден исходник ServiceStateWriter.mq5.' if not source.is_file() else
                 'Для сборки загрузите MetaEditor64.exe в настройках или укажите папку установки MT5.'
                 if not editor.is_file() else '')
        needs_compile = (not ex5.is_file() or not ex5.stat().st_size or
                         (source.is_file() and source.stat().st_mtime > ex5.stat().st_mtime))
        expected = self.digest(ex5)
        pending = []
        for folder in data['terminals']:
            directory = instance_directory(data['mt5_folder'], folder)
            destination = directory / SERVICE_EX5_REL
            try:
                _, _, config_pending = services_ini_update(directory, self.project / SERVICES_INI_REL,
                                                           data['database'])
            except (OSError, ValueError):
                config_pending = True
            if expected is None or self.digest(destination) != expected or config_pending:
                pending.append(folder)
        return {'available': available, 'error': error, 'needs_compile': needs_compile,
                'pending': pending, 'terminals': len(data['terminals'])}

    def status(self, data):
        try:
            repository = self.repository()
        except RuntimeError as error:
            repository = {'error': str(error)}
        return {'git': repository, 'service': self.service_status(data),
                'reload_enabled': os.getenv('MT5_MANAGER_RELOAD') == '1'}

    def deploy(self, data, running):
        editor, mq5, source = service_build_paths(self.project, data)
        if not mq5.is_file():
            raise RuntimeError('Не найден исходник ServiceStateWriter.mq5.')
        if not compile_service(editor, mq5, source, force=True):
            raise RuntimeError('Сборка сервиса не подтверждена. Проверьте MetaEditor; '
                               'старый EX5 не распространялся.')
        if not source.is_file() or not source.stat().st_size:
            raise RuntimeError('MetaEditor не создал непустой EX5. Распространение отменено.')
        copied, failed, restart = [], [], []
        errors = {}
        for folder in data['terminals']:
            directory = instance_directory(data['mt5_folder'], folder).resolve()
            destination = directory / SERVICE_EX5_REL
            temporary = None
            stage = 'Проверка установки терминала'
            try:
                executable = directory / data['mt5_exe']
                if not executable.is_file():
                    raise FileNotFoundError(f'Не найден исполняемый файл терминала: {executable}')
                if destination.resolve() != source.resolve():
                    stage = 'Создание папки сервиса'
                    destination.parent.mkdir(parents=True, exist_ok=True)
                    temporary = destination.with_name(destination.name + '.' + uuid.uuid4().hex + '.tmp')
                    stage = 'Копирование EX5 во временный файл'
                    shutil.copyfile(source, temporary)
                    stage = 'Замена EX5 в папке терминала'
                    os.replace(temporary, destination)
                stage = 'Настройка автозапуска сервиса'
                target, text, changed = services_ini_update(directory, self.project / SERVICES_INI_REL,
                                                            data['database'])
                if changed:
                    if running(folder):
                        raise ValueError('Остановите терминал и повторите развёртывание для обновления services.ini.')
                    write_services_ini(target, text)
                copied.append(folder)
                if running(folder):
                    restart.append(folder)
            except (OSError, ValueError) as error:
                failed.append(folder)
                errors[folder] = f'{stage}: {error}. Путь сервиса: {destination}'
                print(f'Ошибка развёртывания {folder}: {errors[folder]}', flush=True)
            finally:
                if temporary:
                    try:
                        temporary.unlink(missing_ok=True)
                    except OSError as error:
                        # Не подменяем исходную ошибку ошибкой удаления временного файла.
                        cleanup_error = f'Не удалось удалить временный файл {temporary}: {error}'
                        if folder not in failed:
                            failed.append(folder)
                        errors[folder] = (errors.get(folder, '') + ' ' + cleanup_error).strip()
                        print(cleanup_error, flush=True)
        return {'copied': copied, 'failed': failed, 'errors': errors, 'restart_required': restart}