diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml
index 4077495..a4ef339 100644
--- a/.gitea/workflows/ci.yml
+++ b/.gitea/workflows/ci.yml
@@ -28,8 +28,8 @@ jobs:
- name: Validate syntax
run: |
- python -m py_compile app_version.py main.py build.py tests/test_auth_relogin_smoke.py
+ python -m py_compile app_version.py main.py build.py updater_gui.py tests/test_auth_relogin_smoke.py tests/test_auto_update_service.py tests/test_chat_actions.py tests/test_token_store.py
- name: Run tests
run: |
- python -m unittest tests/test_auth_relogin_smoke.py
+ python -m unittest discover -s tests -p "test_*.py" -v
diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml
index 45da9c3..12d587d 100644
--- a/.gitea/workflows/release.yml
+++ b/.gitea/workflows/release.yml
@@ -70,8 +70,8 @@ jobs:
if: env.CONTINUE == 'true'
shell: powershell
run: |
- python -m py_compile app_version.py main.py build.py tests/test_auth_relogin_smoke.py
- python -m unittest tests/test_auth_relogin_smoke.py
+ python -m py_compile app_version.py main.py build.py updater_gui.py tests/test_auth_relogin_smoke.py tests/test_auto_update_service.py tests/test_chat_actions.py tests/test_token_store.py
+ python -m unittest discover -s tests -p "test_*.py" -v
- name: Build release zip
if: env.CONTINUE == 'true'
diff --git a/build.py b/build.py
index 9ffe9df..b30ddb6 100644
--- a/build.py
+++ b/build.py
@@ -6,12 +6,14 @@ from app_version import APP_VERSION
# --- Конфигурация ---
APP_NAME = "AnabasisManager"
+UPDATER_NAME = "AnabasisUpdater"
VERSION = APP_VERSION # Единая версия приложения
MAIN_SCRIPT = "main.py"
+UPDATER_SCRIPT = "updater_gui.py"
ICON_PATH = "icon.ico"
DIST_DIR = os.path.join("dist", APP_NAME)
ARCHIVE_NAME = f"{APP_NAME}-{VERSION}" # Формат Название-Версия
-SAFE_CLEAN_ROOT_FILES = {"main.py", "requirements.txt", "build.py"}
+SAFE_CLEAN_ROOT_FILES = {"main.py", "updater_gui.py", "requirements.txt", "build.py"}
REMOVE_LIST = [
"Qt6Pdf.dll", "Qt6PdfQuick.dll", "Qt6PdfWidgets.dll",
"Qt6VirtualKeyboard.dll", "Qt6Positioning.dll",
@@ -58,6 +60,29 @@ def run_build():
sys.exit(1)
+def run_updater_build():
+ print(f"\n--- 1.2 Сборка {UPDATER_NAME} ---")
+ command = [
+ "pyinstaller",
+ "--noconfirm",
+ "--onefile",
+ "--windowed",
+ f"--name={UPDATER_NAME}",
+ "--distpath", DIST_DIR,
+ "--workpath", os.path.join("build", "updater"),
+ "--specpath", os.path.join("build", "updater_spec"),
+ f"--icon={ICON_PATH}" if os.path.exists(ICON_PATH) else "",
+ UPDATER_SCRIPT,
+ ]
+ command = [arg for arg in command if arg]
+ try:
+ subprocess.check_call(command)
+ print(f"[OK] {UPDATER_NAME} собран.")
+ except subprocess.CalledProcessError as e:
+ print(f"[ERROR] Ошибка при сборке {UPDATER_NAME}: {e}")
+ sys.exit(1)
+
+
def run_cleanup():
print(f"\n--- 2. Оптимизация папки {APP_NAME} ---")
@@ -99,6 +124,7 @@ if __name__ == "__main__":
shutil.rmtree(folder)
run_build()
+ run_updater_build()
run_cleanup()
create_archive()
diff --git a/main.py b/main.py
index da50378..55dfc40 100644
--- a/main.py
+++ b/main.py
@@ -1,10 +1,22 @@
-import sys
import json
-import time
-import shutil
-import auth_webview
import os
+import shutil
+import sys
import threading
+import time
+
+from PySide6.QtCore import QProcess
+from PySide6.QtCore import QStandardPaths
+from PySide6.QtCore import Qt, QUrl, QDateTime, QTimer
+from PySide6.QtGui import QIcon, QAction, QActionGroup, QDesktopServices
+from PySide6.QtWidgets import (QApplication, QMainWindow, QLabel, QLineEdit,
+ QPushButton, QVBoxLayout, QWidget, QMessageBox,
+ QTextBrowser, QScrollArea, QCheckBox, QHBoxLayout,
+ QTabWidget, QDialog, QDialogButtonBox,
+ QProgressBar)
+from vk_api.exceptions import VkApiError
+
+import auth_webview
from app_version import APP_VERSION
from services import (
AutoUpdateService,
@@ -18,20 +30,14 @@ from services import (
)
from ui.dialogs import MultiLinkDialog as UIMultiLinkDialog
from ui.main_window import instructions_text
-from PySide6.QtWidgets import (QApplication, QMainWindow, QLabel, QLineEdit,
- QPushButton, QVBoxLayout, QWidget, QMessageBox,
- QTextBrowser, QScrollArea, QCheckBox, QHBoxLayout,
- QSizePolicy, QTabWidget, QDialog, QDialogButtonBox,
- QProgressBar)
-from PySide6.QtCore import Qt, QUrl, QDateTime, QTimer
-from PySide6.QtGui import QIcon, QAction, QDesktopServices
-from vk_api.exceptions import VkApiError
-from PySide6.QtCore import QStandardPaths
-from PySide6.QtCore import QProcess
# --- Управление токенами и настройками ---
-APP_DATA_DIR = os.path.join(QStandardPaths.writableLocation(QStandardPaths.AppDataLocation), "AnabasisVKChatManager")
+APP_DATA_DIR = os.path.join(
+ QStandardPaths.writableLocation(QStandardPaths.StandardLocation.AppDataLocation),
+ "AnabasisVKChatManager",
+)
TOKEN_FILE = os.path.join(APP_DATA_DIR, "token.json")
+SETTINGS_FILE = os.path.join(APP_DATA_DIR, "settings.json")
WEB_ENGINE_CACHE_DIR = os.path.join(APP_DATA_DIR, "web_engine_cache")
CACHE_CLEANUP_MARKER = os.path.join(APP_DATA_DIR, "pending_cache_cleanup")
LOG_FILE = os.path.join(APP_DATA_DIR, "app.log")
@@ -42,6 +48,7 @@ AUTH_RELOGIN_BACKOFF_SECONDS = 5.0
UPDATE_REPOSITORY = ""
# Full repository URL is preferred (supports GitHub/Gitea).
UPDATE_REPOSITORY_URL = "https://git.daemonlord.ru/benya/AnabasisChatRemove"
+UPDATE_CHANNEL_DEFAULT = "stable"
UPDATE_REQUEST_TIMEOUT = 8
@@ -79,6 +86,7 @@ class VkChatManager(QMainWindow):
self._auth_relogin_in_progress = False
self._last_auth_relogin_ts = 0.0
self.update_repository_url = detect_update_repository_url(UPDATE_REPOSITORY_URL, UPDATE_REPOSITORY)
+ self.update_channel = UPDATE_CHANNEL_DEFAULT
self.update_checker = None
self.update_thread = None
self._update_check_silent = False
@@ -90,6 +98,7 @@ class VkChatManager(QMainWindow):
self._cleanup_cache_if_needed()
self._ensure_log_dir()
+ self._load_settings()
self.init_ui()
self.load_saved_token_on_startup()
self.setup_token_timer()
@@ -114,7 +123,7 @@ class VkChatManager(QMainWindow):
layout.addWidget(self.token_input)
self.token_timer_label = QLabel("Срок действия токена: Н/Д")
- self.token_timer_label.setAlignment(Qt.AlignRight)
+ self.token_timer_label.setAlignment(Qt.AlignmentFlag.AlignRight)
layout.addWidget(self.token_timer_label)
self.auth_btn = QPushButton("Авторизоваться через VK")
@@ -176,7 +185,7 @@ class VkChatManager(QMainWindow):
self.add_user_btn.clicked.connect(self.add_user_to_chat)
layout.addWidget(self.add_user_btn)
self.status_label = QLabel("Статус: не авторизован")
- self.status_label.setAlignment(Qt.AlignCenter)
+ self.status_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
layout.addWidget(self.status_label)
layout.addStretch(1)
@@ -230,6 +239,26 @@ class VkChatManager(QMainWindow):
tools_menu.addAction(check_updates_action)
self.check_updates_action = check_updates_action
+ channel_menu = tools_menu.addMenu("Канал обновлений")
+ self.update_channel_group = QActionGroup(self)
+ self.update_channel_group.setExclusive(True)
+
+ stable_channel_action = QAction("Релизы (stable)", self)
+ stable_channel_action.setCheckable(True)
+ stable_channel_action.setChecked(self.update_channel == "stable")
+ stable_channel_action.triggered.connect(lambda checked: checked and self.set_update_channel("stable"))
+ channel_menu.addAction(stable_channel_action)
+ self.update_channel_group.addAction(stable_channel_action)
+ self.update_channel_stable_action = stable_channel_action
+
+ beta_channel_action = QAction("Бета (pre-release)", self)
+ beta_channel_action.setCheckable(True)
+ beta_channel_action.setChecked(self.update_channel == "beta")
+ beta_channel_action.triggered.connect(lambda checked: checked and self.set_update_channel("beta"))
+ channel_menu.addAction(beta_channel_action)
+ self.update_channel_group.addAction(beta_channel_action)
+ self.update_channel_beta_action = beta_channel_action
+
logout_action = QAction("Выйти и очистить", self)
logout_action.setStatusTip("Выйти, удалить токен и кэш")
logout_action.triggered.connect(self.logout_and_clear)
@@ -259,12 +288,12 @@ class VkChatManager(QMainWindow):
"Поддерживается проверка обновлений и автообновление Windows-сборки.
"
f"Репозиторий: {repo_html}"
)
- content.setTextFormat(Qt.RichText)
- content.setTextInteractionFlags(Qt.TextBrowserInteraction)
+ content.setTextFormat(Qt.TextFormat.RichText)
+ content.setTextInteractionFlags(Qt.TextInteractionFlag.TextBrowserInteraction)
content.setOpenExternalLinks(True)
content.setWordWrap(True)
- button_box = QDialogButtonBox(QDialogButtonBox.Ok, parent=dialog)
+ button_box = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok, parent=dialog)
button_box.accepted.connect(dialog.accept)
layout = QVBoxLayout(dialog)
@@ -281,8 +310,8 @@ class VkChatManager(QMainWindow):
scroll_area = QScrollArea()
scroll_area.setWidgetResizable(True)
- scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
- scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarAsNeeded)
+ scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
+ scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
tab_layout.addWidget(scroll_area)
@@ -300,15 +329,60 @@ class VkChatManager(QMainWindow):
if hasattr(self, "check_updates_action"):
self.check_updates_action.setEnabled(not in_progress)
+ def _normalize_update_channel(self, value):
+ channel = (value or "").strip().lower()
+ if channel in ("beta", "betas", "pre", "prerelease", "pre-release"):
+ return "beta"
+ return "stable"
+
+ def _load_settings(self):
+ self.update_channel = UPDATE_CHANNEL_DEFAULT
+ if not os.path.exists(SETTINGS_FILE):
+ return
+ try:
+ with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
+ settings = json.load(f)
+ self.update_channel = self._normalize_update_channel(settings.get("update_channel"))
+ except Exception as e:
+ self._log_event("settings_load", f"Ошибка загрузки настроек: {e}", level="WARN")
+
+ def _save_settings(self):
+ try:
+ os.makedirs(APP_DATA_DIR, exist_ok=True)
+ settings = {
+ "update_channel": self.update_channel,
+ }
+ with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
+ json.dump(settings, f, ensure_ascii=False, indent=2)
+ except Exception as e:
+ self._log_event("settings_save", f"Ошибка сохранения настроек: {e}", level="WARN")
+
+ def set_update_channel(self, channel):
+ normalized = self._normalize_update_channel(channel)
+ if normalized == self.update_channel:
+ return
+ self.update_channel = normalized
+ self._save_settings()
+ self.status_label.setText(
+ f"Статус: канал обновлений переключен на {'бета' if normalized == 'beta' else 'релизы'}."
+ )
+ self._log_event("update_channel", f"update_channel={self.update_channel}")
+
def check_for_updates(self, silent_no_updates=False):
if self.update_thread and self.update_thread.is_alive():
return
self._update_check_silent = silent_no_updates
self._set_update_action_state(True)
- self.status_label.setText("Статус: проверка обновлений...")
+ channel_label = "бета" if self.update_channel == "beta" else "релизы"
+ self.status_label.setText(f"Статус: проверка обновлений ({channel_label})...")
- self.update_checker = UpdateChecker(self.update_repository_url, APP_VERSION, request_timeout=UPDATE_REQUEST_TIMEOUT)
+ self.update_checker = UpdateChecker(
+ self.update_repository_url,
+ APP_VERSION,
+ request_timeout=UPDATE_REQUEST_TIMEOUT,
+ channel=self.update_channel,
+ )
self.update_checker.check_finished.connect(self._on_update_check_finished)
self.update_checker.check_failed.connect(self._on_update_check_failed)
self.update_thread = threading.Thread(target=self.update_checker.run, daemon=True)
@@ -324,17 +398,17 @@ class VkChatManager(QMainWindow):
self.status_label.setText(f"Статус: доступно обновление {latest_version}")
message_box = QMessageBox(self)
- message_box.setIcon(QMessageBox.Information)
+ message_box.setIcon(QMessageBox.Icon.Information)
message_box.setWindowTitle("Доступно обновление")
message_box.setText(
f"Текущая версия: {result.get('current_version')}\n"
f"Доступная версия: {latest_version}\n\n"
"Открыть страницу загрузки?"
)
- update_now_button = message_box.addButton("Обновить сейчас", QMessageBox.AcceptRole)
- download_button = message_box.addButton("Скачать", QMessageBox.AcceptRole)
- releases_button = message_box.addButton("Релизы", QMessageBox.ActionRole)
- cancel_button = message_box.addButton("Позже", QMessageBox.RejectRole)
+ update_now_button = message_box.addButton("Обновить сейчас", QMessageBox.ButtonRole.AcceptRole)
+ download_button = message_box.addButton("Скачать", QMessageBox.ButtonRole.AcceptRole)
+ releases_button = message_box.addButton("Релизы", QMessageBox.ButtonRole.ActionRole)
+ cancel_button = message_box.addButton("Позже", QMessageBox.ButtonRole.RejectRole)
message_box.setDefaultButton(update_now_button)
message_box.exec()
@@ -356,9 +430,10 @@ class VkChatManager(QMainWindow):
self._log_event("update_check", "Диалог обновления закрыт без действия.", level="WARN")
return
- self.status_label.setText(f"Статус: используется актуальная версия ({APP_VERSION}).")
+ channel_label = "бета" if self.update_channel == "beta" else "релизы"
+ self.status_label.setText(f"Статус: используется актуальная версия ({APP_VERSION}, канал: {channel_label}).")
if not self._update_check_silent:
- QMessageBox.information(self, "Обновления", "Установлена актуальная версия.")
+ QMessageBox.information(self, "Обновления", f"Установлена актуальная версия в канале {channel_label}.")
def _on_update_check_failed(self, error_text):
self._set_update_action_state(False)
@@ -453,7 +528,7 @@ class VkChatManager(QMainWindow):
self.status_label.setText(status_text)
if busy:
self._busy = True
- QApplication.setOverrideCursor(Qt.WaitCursor)
+ QApplication.setOverrideCursor(Qt.CursorShape.WaitCursor)
for widget in [
self.refresh_chats_btn, self.select_all_btn, self.deselect_all_btn,
self.vk_url_input, self.multi_link_btn, self.visible_messages_checkbox,
@@ -537,9 +612,9 @@ class VkChatManager(QMainWindow):
self,
"Подтверждение выхода",
"Вы уверены, что хотите выйти и удалить сохраненный токен и кэш?",
- QMessageBox.Yes | QMessageBox.No
+ QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No
)
- if confirm != QMessageBox.Yes:
+ if confirm != QMessageBox.StandardButton.Yes:
return
self._clear_auth_state(stop_timer=True, remove_token_file=True)
@@ -596,7 +671,7 @@ class VkChatManager(QMainWindow):
def _on_auth_process_error(self, process_error):
self._auth_process_error_text = f"Ошибка запуска авторизации: {process_error}"
# For failed starts Qt may not emit finished(), so release UI here.
- if self.auth_process and self.auth_process.state() == QProcess.NotRunning:
+ if self.auth_process and self.auth_process.state() == QProcess.ProcessState.NotRunning:
output_path = self.auth_output_path
self.auth_output_path = None
self.auth_process = None
@@ -656,7 +731,7 @@ class VkChatManager(QMainWindow):
self.handle_new_auth_token(token, expires_in)
def start_auth(self, keep_status_text=False):
- if self.auth_process and self.auth_process.state() != QProcess.NotRunning:
+ if self.auth_process and self.auth_process.state() != QProcess.ProcessState.NotRunning:
self._log_event("auth_process", "Авторизация уже запущена, повторный запуск пропущен.")
return
@@ -880,9 +955,9 @@ class VkChatManager(QMainWindow):
confirm_dialog = QMessageBox(self)
confirm_dialog.setWindowTitle("Подтверждение действия")
confirm_dialog.setText(msg)
- confirm_dialog.setIcon(QMessageBox.Question)
- yes_button = confirm_dialog.addButton("Да", QMessageBox.YesRole)
- no_button = confirm_dialog.addButton("Нет", QMessageBox.NoRole)
+ confirm_dialog.setIcon(QMessageBox.Icon.Question)
+ yes_button = confirm_dialog.addButton("Да", QMessageBox.ButtonRole.YesRole)
+ no_button = confirm_dialog.addButton("Нет", QMessageBox.ButtonRole.NoRole)
confirm_dialog.setDefaultButton(no_button) # "Нет" - кнопка по умолчанию
confirm_dialog.exec()
@@ -955,9 +1030,9 @@ class VkChatManager(QMainWindow):
confirm_dialog = QMessageBox(self)
confirm_dialog.setWindowTitle("Подтверждение прав")
confirm_dialog.setText(msg)
- confirm_dialog.setIcon(QMessageBox.Question)
- yes_button = confirm_dialog.addButton("Да", QMessageBox.YesRole)
- no_button = confirm_dialog.addButton("Нет", QMessageBox.NoRole)
+ confirm_dialog.setIcon(QMessageBox.Icon.Question)
+ yes_button = confirm_dialog.addButton("Да", QMessageBox.ButtonRole.YesRole)
+ no_button = confirm_dialog.addButton("Нет", QMessageBox.ButtonRole.NoRole)
confirm_dialog.setDefaultButton(no_button)
confirm_dialog.exec()
@@ -1129,18 +1204,18 @@ class VkChatManager(QMainWindow):
download_name=download_name,
)
app_exe = sys.executable
- script_path = AutoUpdateService.build_update_script(
- app_dir=os.path.dirname(app_exe),
+ AutoUpdateService.launch_gui_updater(
+ app_exe=app_exe,
source_dir=source_dir,
- exe_name=os.path.basename(app_exe),
+ work_dir=work_dir,
target_pid=os.getpid(),
+ version=latest_version,
)
- AutoUpdateService.launch_update_script(script_path, work_dir)
self._log_event("auto_update", f"Update {latest_version} started from {download_url}")
QMessageBox.information(
self,
"Обновление запущено",
- "Обновление скачано. Приложение будет перезапущено.",
+ "Обновление скачано. Открылось окно обновления.",
)
QApplication.instance().quit()
return True
diff --git a/services/auto_update_service.py b/services/auto_update_service.py
index d6c08eb..da73313 100644
--- a/services/auto_update_service.py
+++ b/services/auto_update_service.py
@@ -104,12 +104,26 @@ class AutoUpdateService:
"if %ERRORLEVEL% EQU 0 (",
" set /a WAIT_LOOPS+=1",
" if %WAIT_LOOPS% GEQ 180 (",
- " echo Timeout waiting for process %TARGET_PID% to exit >> \"%UPDATE_LOG%\"",
- " goto :backup",
+ " echo Timeout waiting for process %TARGET_PID%, attempting force stop >> \"%UPDATE_LOG%\"",
+ " taskkill /PID %TARGET_PID% /T /F >nul 2>&1",
+ " timeout /t 2 /nobreak >nul",
+ " tasklist /FI \"PID eq %TARGET_PID%\" | find \"%TARGET_PID%\" >nul",
+ " if %ERRORLEVEL% EQU 0 goto :pid_still_running",
+ " goto :wait_image_unlock",
" )",
" timeout /t 1 /nobreak >nul",
" goto :wait_for_exit",
")",
+ ":wait_image_unlock",
+ "set /a IMG_LOOPS=0",
+ ":check_image",
+ "tasklist /FI \"IMAGENAME eq %EXE_NAME%\" | find /I \"%EXE_NAME%\" >nul",
+ "if %ERRORLEVEL% EQU 0 (",
+ " set /a IMG_LOOPS+=1",
+ " if %IMG_LOOPS% GEQ 60 goto :image_still_running",
+ " timeout /t 1 /nobreak >nul",
+ " goto :check_image",
+ ")",
":backup",
"timeout /t 1 /nobreak >nul",
"mkdir \"%BACKUP_DIR%\" >nul 2>&1",
@@ -134,6 +148,12 @@ class AutoUpdateService:
":backup_error",
"echo Auto-update failed during backup. Code %RC% >> \"%UPDATE_LOG%\"",
"exit /b %RC%",
+ ":pid_still_running",
+ "echo Auto-update aborted: process %TARGET_PID% is still running after force stop. >> \"%UPDATE_LOG%\"",
+ "exit /b 4",
+ ":image_still_running",
+ "echo Auto-update aborted: %EXE_NAME% still running and file lock may remain. >> \"%UPDATE_LOG%\"",
+ "exit /b 5",
]
with open(script_path, "w", encoding="utf-8", newline="\r\n") as f:
f.write("\r\n".join(script_lines) + "\r\n")
@@ -152,6 +172,40 @@ class AutoUpdateService:
creationflags=creation_flags,
)
+ @staticmethod
+ def launch_gui_updater(app_exe, source_dir, work_dir, target_pid, version=""):
+ app_dir = os.path.dirname(app_exe)
+ exe_name = os.path.basename(app_exe)
+ updater_exe = os.path.join(app_dir, "AnabasisUpdater.exe")
+ if not os.path.exists(updater_exe):
+ raise RuntimeError("Файл AnabasisUpdater.exe не найден в папке приложения.")
+
+ creation_flags = 0
+ if hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
+ creation_flags |= subprocess.CREATE_NEW_PROCESS_GROUP
+ if hasattr(subprocess, "DETACHED_PROCESS"):
+ creation_flags |= subprocess.DETACHED_PROCESS
+
+ subprocess.Popen(
+ [
+ updater_exe,
+ "--app-dir",
+ app_dir,
+ "--source-dir",
+ source_dir,
+ "--exe-name",
+ exe_name,
+ "--target-pid",
+ str(target_pid),
+ "--version",
+ str(version or ""),
+ "--work-dir",
+ str(work_dir or ""),
+ ],
+ cwd=work_dir,
+ creationflags=creation_flags,
+ )
+
@classmethod
def prepare_update(cls, download_url, checksum_url, download_name):
work_dir = tempfile.mkdtemp(prefix="anabasis_update_")
diff --git a/services/update_service.py b/services/update_service.py
index bc81f67..d67703f 100644
--- a/services/update_service.py
+++ b/services/update_service.py
@@ -38,6 +38,75 @@ def _sanitize_repo_url(value):
return f"{parsed.scheme}://{parsed.netloc}{clean_path}"
+def _normalize_update_channel(value):
+ channel = (value or "").strip().lower()
+ if channel in ("beta", "betas", "pre", "prerelease", "pre-release"):
+ return "beta"
+ return "stable"
+
+
+def _select_release_from_list(releases):
+ for item in releases:
+ if not isinstance(item, dict):
+ continue
+ if item.get("draft"):
+ continue
+ tag_name = (item.get("tag_name") or item.get("name") or "").strip()
+ if not tag_name:
+ continue
+ return item
+ return None
+
+
+def _extract_release_payload(release_data, repository_url, current_version):
+ parsed = urlparse(repository_url)
+ base_url = f"{parsed.scheme}://{parsed.netloc}"
+ repo_path = parsed.path.strip("/")
+ releases_url = f"{base_url}/{repo_path}/releases"
+
+ latest_tag = release_data.get("tag_name") or release_data.get("name") or ""
+ latest_version = latest_tag.lstrip("vV").strip()
+ html_url = release_data.get("html_url") or releases_url
+ assets = release_data.get("assets") or []
+ download_url = ""
+ download_name = ""
+ checksum_url = ""
+ for asset in assets:
+ url = asset.get("browser_download_url", "")
+ if url.lower().endswith(".zip"):
+ download_url = url
+ download_name = asset.get("name", "")
+ break
+ if not download_url and assets:
+ download_url = assets[0].get("browser_download_url", "")
+ download_name = assets[0].get("name", "")
+
+ for asset in assets:
+ name = asset.get("name", "").lower()
+ if not name:
+ continue
+ is_checksum_asset = name.endswith(".sha256") or name.endswith(".sha256.txt") or name in ("checksums.txt", "sha256sums.txt")
+ if not is_checksum_asset:
+ continue
+ if download_name and (download_name.lower() in name or name in (f"{download_name.lower()}.sha256", f"{download_name.lower()}.sha256.txt")):
+ checksum_url = asset.get("browser_download_url", "")
+ break
+ if not checksum_url:
+ checksum_url = asset.get("browser_download_url", "")
+
+ return {
+ "repository_url": repository_url,
+ "latest_version": latest_version,
+ "current_version": current_version,
+ "latest_tag": latest_tag,
+ "release_url": html_url,
+ "download_url": download_url,
+ "download_name": download_name,
+ "checksum_url": checksum_url,
+ "has_update": _is_newer_version(latest_version, current_version),
+ }
+
+
def detect_update_repository_url(configured_url="", configured_repo=""):
env_url = _sanitize_repo_url(os.getenv("ANABASIS_UPDATE_URL", ""))
if env_url:
@@ -74,11 +143,12 @@ class UpdateChecker(QObject):
check_finished = Signal(dict)
check_failed = Signal(str)
- def __init__(self, repository_url, current_version, request_timeout=8):
+ def __init__(self, repository_url, current_version, request_timeout=8, channel="stable"):
super().__init__()
self.repository_url = repository_url
self.current_version = current_version
self.request_timeout = request_timeout
+ self.channel = _normalize_update_channel(channel)
def run(self):
if not self.repository_url:
@@ -92,10 +162,17 @@ class UpdateChecker(QObject):
self.check_failed.emit("Некорректный URL репозитория обновлений.")
return
+ use_beta_channel = self.channel == "beta"
if parsed.netloc.lower().endswith("github.com"):
- api_url = f"https://api.github.com/repos/{repo_path}/releases/latest"
+ if use_beta_channel:
+ api_url = f"https://api.github.com/repos/{repo_path}/releases"
+ else:
+ api_url = f"https://api.github.com/repos/{repo_path}/releases/latest"
else:
- api_url = f"{base_url}/api/v1/repos/{repo_path}/releases/latest"
+ if use_beta_channel:
+ api_url = f"{base_url}/api/v1/repos/{repo_path}/releases"
+ else:
+ api_url = f"{base_url}/api/v1/repos/{repo_path}/releases/latest"
releases_url = f"{base_url}/{repo_path}/releases"
request = urllib.request.Request(
api_url,
@@ -106,7 +183,7 @@ class UpdateChecker(QObject):
)
try:
with urllib.request.urlopen(request, timeout=self.request_timeout) as response:
- release_data = json.loads(response.read().decode("utf-8"))
+ response_data = json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as e:
self.check_failed.emit(f"Ошибка HTTP при проверке обновлений: {e.code}")
return
@@ -117,47 +194,20 @@ class UpdateChecker(QObject):
self.check_failed.emit(f"Не удалось проверить обновления: {e}")
return
- latest_tag = release_data.get("tag_name") or release_data.get("name") or ""
- latest_version = latest_tag.lstrip("vV").strip()
- html_url = release_data.get("html_url") or releases_url
- assets = release_data.get("assets") or []
- download_url = ""
- download_name = ""
- checksum_url = ""
- for asset in assets:
- url = asset.get("browser_download_url", "")
- if url.lower().endswith(".zip"):
- download_url = url
- download_name = asset.get("name", "")
- break
- if not download_url and assets:
- download_url = assets[0].get("browser_download_url", "")
- download_name = assets[0].get("name", "")
-
- for asset in assets:
- name = asset.get("name", "").lower()
- if not name:
- continue
- is_checksum_asset = name.endswith(".sha256") or name.endswith(".sha256.txt") or name in ("checksums.txt", "sha256sums.txt")
- if not is_checksum_asset:
- continue
- if download_name and (download_name.lower() in name or name in (f"{download_name.lower()}.sha256", f"{download_name.lower()}.sha256.txt")):
- checksum_url = asset.get("browser_download_url", "")
- break
- if not checksum_url:
- checksum_url = asset.get("browser_download_url", "")
-
- self.check_finished.emit(
- {
- "repository_url": self.repository_url,
- "latest_version": latest_version,
- "current_version": self.current_version,
- "latest_tag": latest_tag,
- "release_url": html_url,
- "download_url": download_url,
- "download_name": download_name,
- "checksum_url": checksum_url,
- "has_update": _is_newer_version(latest_version, self.current_version),
- }
- )
+ release_data = response_data
+ if use_beta_channel:
+ if not isinstance(response_data, list):
+ self.check_failed.emit("Сервер вернул некорректный ответ списка релизов.")
+ return
+ release_data = _select_release_from_list(response_data)
+ if not release_data:
+ self.check_failed.emit("В канале beta не найдено доступных релизов.")
+ return
+ elif not isinstance(response_data, dict):
+ self.check_failed.emit("Сервер вернул некорректный ответ релиза.")
+ return
+ payload = _extract_release_payload(release_data, self.repository_url, self.current_version)
+ payload["release_channel"] = self.channel
+ payload["releases_url"] = releases_url
+ self.check_finished.emit(payload)
diff --git a/tests/test_auth_relogin_smoke.py b/tests/test_auth_relogin_smoke.py
index 9dc0888..f43ac66 100644
--- a/tests/test_auth_relogin_smoke.py
+++ b/tests/test_auth_relogin_smoke.py
@@ -19,7 +19,7 @@ class AuthReloginSmokeTests(unittest.TestCase):
self.assertIn("process = QProcess(self)", self.main_source)
self.assertIn("process.start(program, args)", self.main_source)
self.assertIn("def _on_auth_process_finished(self, exit_code, _exit_status):", self.main_source)
- self.assertIn("if self.auth_process and self.auth_process.state() == QProcess.NotRunning:", self.main_source)
+ self.assertIn("if self.auth_process and self.auth_process.state() == QProcess.ProcessState.NotRunning:", self.main_source)
def test_force_relogin_has_backoff_and_event_log(self):
self.assertIn("AUTH_RELOGIN_BACKOFF_SECONDS = 5.0", self.main_source)
@@ -53,7 +53,7 @@ class AuthReloginSmokeTests(unittest.TestCase):
self.assertIn("class UpdateChecker(QObject):", self.update_source)
self.assertIn("def _start_auto_update(self, download_url, latest_version, checksum_url=\"\", download_name=\"\"):", self.main_source)
self.assertIn("AutoUpdateService.prepare_update", self.main_source)
- self.assertIn("AutoUpdateService.build_update_script", self.main_source)
+ self.assertIn("AutoUpdateService.launch_gui_updater", self.main_source)
if __name__ == "__main__":
diff --git a/tests/test_update_service.py b/tests/test_update_service.py
new file mode 100644
index 0000000..19ef71f
--- /dev/null
+++ b/tests/test_update_service.py
@@ -0,0 +1,51 @@
+import unittest
+import importlib.util
+from pathlib import Path
+
+MODULE_PATH = Path("services/update_service.py")
+SPEC = importlib.util.spec_from_file_location("update_service_under_test", MODULE_PATH)
+update_service = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(update_service)
+
+
+class UpdateServiceTests(unittest.TestCase):
+ def test_normalize_update_channel(self):
+ self.assertEqual(update_service._normalize_update_channel("stable"), "stable")
+ self.assertEqual(update_service._normalize_update_channel("beta"), "beta")
+ self.assertEqual(update_service._normalize_update_channel("pre-release"), "beta")
+ self.assertEqual(update_service._normalize_update_channel("unknown"), "stable")
+ self.assertEqual(update_service._normalize_update_channel(""), "stable")
+
+ def test_select_release_from_list_skips_drafts(self):
+ releases = [
+ {"tag_name": "v2.0.0", "draft": True},
+ {"tag_name": "", "draft": False},
+ {"tag_name": "v1.9.0-beta.1", "draft": False},
+ ]
+ selected = update_service._select_release_from_list(releases)
+ self.assertIsNotNone(selected)
+ self.assertEqual(selected["tag_name"], "v1.9.0-beta.1")
+
+ def test_extract_release_payload_uses_zip_and_checksum(self):
+ release_data = {
+ "tag_name": "v1.7.2",
+ "html_url": "https://example.com/release/v1.7.2",
+ "assets": [
+ {"name": "notes.txt", "browser_download_url": "https://example.com/notes.txt"},
+ {"name": "AnabasisManager-win64.zip", "browser_download_url": "https://example.com/app.zip"},
+ {"name": "AnabasisManager-win64.zip.sha256", "browser_download_url": "https://example.com/app.zip.sha256"},
+ ],
+ }
+ payload = update_service._extract_release_payload(
+ release_data=release_data,
+ repository_url="https://git.daemonlord.ru/benya/AnabasisChatRemove",
+ current_version="1.7.1",
+ )
+ self.assertEqual(payload["latest_version"], "1.7.2")
+ self.assertEqual(payload["download_url"], "https://example.com/app.zip")
+ self.assertEqual(payload["checksum_url"], "https://example.com/app.zip.sha256")
+ self.assertTrue(payload["has_update"])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/updater_gui.py b/updater_gui.py
new file mode 100644
index 0000000..faa8582
--- /dev/null
+++ b/updater_gui.py
@@ -0,0 +1,218 @@
+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())