import argparse import os import shutil import subprocess import sys import tempfile import time from PySide6.QtCore import QObject, Qt, QThread, Signal, QTimer, QUrl from PySide6.QtGui import QDesktopServices from PySide6.QtWidgets import QApplication, QLabel, QProgressBar, QVBoxLayout, QWidget, QPushButton, QHBoxLayout def _write_log(log_path, message): try: os.makedirs(os.path.dirname(log_path), exist_ok=True) with open(log_path, "a", encoding="utf-8") as f: f.write(message.rstrip() + "\n") except Exception: pass def _is_pid_running(pid): if pid <= 0: return False try: completed = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}"], capture_output=True, text=True, timeout=5, check=False, ) return str(pid) in (completed.stdout or "") except Exception: return False def _mirror_tree(src_dir, dst_dir, skip_names=None): skip_set = {name.lower() for name in (skip_names or [])} os.makedirs(dst_dir, exist_ok=True) for root, dirs, files in os.walk(src_dir): rel = os.path.relpath(root, src_dir) target_root = dst_dir if rel == "." else os.path.join(dst_dir, rel) os.makedirs(target_root, exist_ok=True) for file_name in files: if file_name.lower() in skip_set: continue source_file = os.path.join(root, file_name) target_file = os.path.join(target_root, file_name) os.makedirs(os.path.dirname(target_file), exist_ok=True) shutil.copy2(source_file, target_file) class UpdateWorker(QObject): status = Signal(int, str) failed = Signal(str) done = Signal() def __init__(self, app_dir, source_dir, exe_name, target_pid, version, work_dir=""): super().__init__() self.app_dir = app_dir self.source_dir = source_dir self.exe_name = exe_name self.target_pid = int(target_pid or 0) self.version = version or "" self.work_dir = work_dir or "" self.log_path = os.path.join(app_dir, "update_error.log") def run(self): backup_dir = os.path.join(tempfile.gettempdir(), f"anabasis_backup_{int(time.time())}") skip_names = {"anabasisupdater.exe"} try: self.status.emit(1, "Ожидание завершения приложения...") wait_loops = 0 while _is_pid_running(self.target_pid): time.sleep(1) wait_loops += 1 if wait_loops >= 180: self.status.emit(1, "Принудительное завершение зависшего процесса...") subprocess.run( ["taskkill", "/PID", str(self.target_pid), "/T", "/F"], capture_output=True, text=True, timeout=10, check=False, ) time.sleep(2) if _is_pid_running(self.target_pid): raise RuntimeError(f"Процесс {self.target_pid} не завершился.") break self.status.emit(2, "Создание резервной копии...") _mirror_tree(self.app_dir, backup_dir, skip_names=skip_names) self.status.emit(3, "Применение обновления...") _mirror_tree(self.source_dir, self.app_dir, skip_names=skip_names) self.status.emit(4, "Запуск обновленного приложения...") app_exe = os.path.join(self.app_dir, self.exe_name) creation_flags = 0 if hasattr(subprocess, "DETACHED_PROCESS"): creation_flags |= subprocess.DETACHED_PROCESS if hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"): creation_flags |= subprocess.CREATE_NEW_PROCESS_GROUP subprocess.Popen([app_exe], cwd=self.app_dir, creationflags=creation_flags) _write_log(self.log_path, f"Update success to version {self.version or 'unknown'}") self.status.emit(5, "Очистка временных файлов...") try: shutil.rmtree(backup_dir, ignore_errors=True) if self.work_dir and os.path.isdir(self.work_dir): shutil.rmtree(self.work_dir, ignore_errors=True) except Exception: pass self.done.emit() except Exception as exc: _write_log(self.log_path, f"Update failed: {exc}") try: self.status.emit(4, "Восстановление из резервной копии...") _mirror_tree(backup_dir, self.app_dir, skip_names=skip_names) except Exception as rollback_exc: _write_log(self.log_path, f"Rollback failed: {rollback_exc}") self.failed.emit(str(exc)) class UpdaterWindow(QWidget): def __init__(self, app_dir, source_dir, exe_name, target_pid, version, work_dir=""): super().__init__() self.setWindowTitle("Anabasis Updater") self.setMinimumWidth(480) self.log_path = os.path.join(app_dir, "update_error.log") self.label = QLabel("Подготовка обновления...") self.label.setWordWrap(True) self.progress = QProgressBar() self.progress.setRange(0, 5) self.progress.setValue(0) self.open_log_btn = QPushButton("Открыть лог") self.open_log_btn.setEnabled(False) self.open_log_btn.clicked.connect(self.open_log) self.close_btn = QPushButton("Закрыть") self.close_btn.setEnabled(False) self.close_btn.clicked.connect(self.close) layout = QVBoxLayout(self) layout.addWidget(self.label) layout.addWidget(self.progress) actions = QHBoxLayout() actions.addStretch(1) actions.addWidget(self.open_log_btn) actions.addWidget(self.close_btn) layout.addLayout(actions) self.thread = QThread(self) self.worker = UpdateWorker(app_dir, source_dir, exe_name, target_pid, version, work_dir=work_dir) self.worker.moveToThread(self.thread) self.thread.started.connect(self.worker.run) self.worker.status.connect(self.on_status) self.worker.failed.connect(self.on_failed) self.worker.done.connect(self.on_done) self.worker.done.connect(self.thread.quit) self.worker.failed.connect(self.thread.quit) self.thread.start() def on_status(self, step, text): self.label.setText(text) self.progress.setValue(max(0, min(5, int(step)))) def on_done(self): self.label.setText("Обновление успешно применено. Запускаю приложение...") self.progress.setValue(5) QTimer.singleShot(900, self.close) def on_failed(self, error_text): self.label.setText( "Не удалось применить обновление.\n" f"Причина: {error_text}\n" "Подробности сохранены в update_error.log." ) self.open_log_btn.setEnabled(True) self.close_btn.setEnabled(True) def open_log(self): if os.path.exists(self.log_path): QDesktopServices.openUrl(QUrl.fromLocalFile(self.log_path)) def parse_args(): parser = argparse.ArgumentParser() parser.add_argument("--app-dir", required=True) parser.add_argument("--source-dir", required=True) parser.add_argument("--exe-name", required=True) parser.add_argument("--target-pid", required=True) parser.add_argument("--version", default="") parser.add_argument("--work-dir", default="") return parser.parse_args() def main(): args = parse_args() app = QApplication(sys.argv) app.setStyle("Fusion") window = UpdaterWindow( app_dir=args.app_dir, source_dir=args.source_dir, exe_name=args.exe_name, target_pid=args.target_pid, version=args.version, work_dir=args.work_dir, ) window.show() return app.exec() if __name__ == "__main__": sys.exit(main())