From f8f6a216f5cb5a764aba557068ae3580414061e6 Mon Sep 17 00:00:00 2001 From: Danil Iarosh Date: Fri, 18 Sep 2026 10:37:21 +0300 Subject: [PATCH] feat: add OTA updates and zero-based mapping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Добавлена проверка и безопасная установка OTA-обновлений Windows с откатом. - Нумерация линий и блоков переведена на индексы с нуля. - Добавлена миграция старых JSON-схем с сохранением физической адресации. - Обновлены генератор, сборка, документация и smoke-проверки. - Подготовлена версия 3.3.0. --- LEDMapper.spec | 19 ++- README.md | 14 +- __init__.py | 2 + main.py | 4 + main_window.py | 148 ++++++++++++++++- models.py | 66 ++++++-- panel_mapping_generator.py | 4 +- qa_smoke.py | 226 +++++++++++++++++++++---- update_service.py | 326 +++++++++++++++++++++++++++++++++++++ version.py | 3 + 10 files changed, 751 insertions(+), 61 deletions(-) create mode 100644 update_service.py create mode 100644 version.py diff --git a/LEDMapper.spec b/LEDMapper.spec index 0e9cf37..d66e9be 100644 --- a/LEDMapper.spec +++ b/LEDMapper.spec @@ -10,15 +10,15 @@ spec_dir = Path(SPECPATH) def git_build_ref(): try: - dirty = subprocess.run( - ['git', 'diff-index', '--quiet', 'HEAD', '--'], + dirty = subprocess.check_output( + ['git', 'status', '--porcelain', '--untracked-files=normal'], cwd=spec_dir, stderr=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - ).returncode != 0 - if dirty: + text=True, + ) + if dirty.strip(): return datetime.now().strftime('%H%M%S-%d%m%Y') - except OSError: + except (OSError, subprocess.CalledProcessError): return datetime.now().strftime('%H%M%S-%d%m%Y') try: tag = subprocess.check_output( @@ -46,13 +46,16 @@ def git_build_ref(): app_name = f'LEDMapper-{git_build_ref()}' +build_version_path = spec_dir.parent / 'build' / 'ledmapper-build-version.txt' +build_version_path.parent.mkdir(parents=True, exist_ok=True) +build_version_path.write_text(app_name.removeprefix('LEDMapper-'), encoding='utf-8') a = Analysis( [str(spec_dir / 'main.py')], pathex=[str(spec_dir), str(spec_dir.parent)], binaries=[], - datas=[(str(spec_dir / 'assets'), 'assets')], + datas=[(str(spec_dir / 'assets'), 'assets'), (str(build_version_path), '.')], hiddenimports=[ 'canvas', 'background_dialog', @@ -66,6 +69,8 @@ a = Analysis( 'settings_dialog', 'theme', 'unfolding', + 'update_service', + 'version', ], hookspath=[], hooksconfig={}, diff --git a/README.md b/README.md index bdaf070..93144a1 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,7 @@ LED Mapper — настольное приложение на Python и PySide6 - Отображение схемы в реальных координатах. - Выделение точек мышью и работа через таблицу выбранных LED. - До 4 физических линий светодиодных лент. +- Нумерация линий и блоков с `0`. - Блоки внутри линий, включая пустые блоки. - Индексация LED отдельно для каждой линии, начиная с `0`. - Индексация внутри каждого блока. @@ -25,6 +26,7 @@ LED Mapper — настольное приложение на Python и PySide6 - Сортировка таблицы выбранных точек по столбцам. - Генерация `vm_map.h`, `ordered_circles.json`, `validation_report.txt` и README для результата. - Сборка самостоятельного Windows EXE через PyInstaller. +- Проверка и установка Windows-обновлений из релизов Gitea. ## Структура приложения @@ -34,6 +36,8 @@ main_window.py Главное окно, меню, действия, canvas.py Холст схемы, выделение, кисть, сетка и связи models.py Модель карты, линий, блоков и алгоритмы маршрутов panel_mapping_generator.py Генерация файлов маппинга +update_service.py Проверка релизов и безопасная установка Windows EXE +version.py Базовая версия приложения при запуске из исходников icons.py Встроенные lucide-иконки unfolding.py Описание развертки линий и блоков qa_smoke.py Smoke-тесты приложения @@ -91,6 +95,8 @@ python -m venv .venv В режиме редактирования выбранные светодиоды можно скопировать через `Ctrl+C` и вставить через `Ctrl+V`. Вставленные LED остаются свободными, сохраняют взаимное расположение и пользовательские поля исходных точек. Первая копия смещается на один шаг сетки вправо и вниз, а следующие вставки образуют каскад. Вся группа вставляется одной операцией Undo и сразу становится текущим выделением. +В собранном Windows EXE приложение проверяет наличие нового стабильного релиза после запуска. В меню `Справка` можно запустить проверку вручную и, если новая версия найдена, выбрать `Установить обновление`. Загружается только артефакт вида `LEDMapper-vX.Y.Z.exe` из репозитория `Robotail/LED-mapper` на `git.robotail.pro`. После проверки размера и формата EXE приложение предлагает сохранить изменённую схему, закрывается, а отдельный процесс заменяет EXE по прежнему пути и запускает новую версию — существующие ярлыки продолжают работать. Версия релизной сборки хранится внутри EXE, поэтому переименование файла не вызывает повторную установку того же обновления. При неудачном запуске старая версия восстанавливается; причина записывается рядом с EXE в `update-error.log`. Для обновления нужны доступ к серверу релизов и право записи в каталог приложения. Из исходников можно проверить наличие версии, но автоустановка недоступна. + Если не назначен ни один светодиод, генерация не запускается. Если назначены не все светодиоды, приложение показывает предупреждение и предлагает либо сгенерировать неполный маппинг, либо продолжить редактирование. ## Горячие клавиши @@ -128,13 +134,13 @@ python -m venv .venv Устаревший `virtual_matrix_legacy.ino.txt` больше не создаётся. Если файл с этим точным именем остался в выходной папке от прежней версии, следующая успешная генерация удалит его. Другие `.ino`-файлы не затрагиваются. -JSON схемы хранит `coordinate_unit` и `grid_step_mm`. При первом открытии старого файла без единицы координат приложение определяет её один раз и затем сохраняет явно. Сохранение выполняется через временный файл и атомарную замену, поэтому исходный JSON не повреждается при ошибке записи. +JSON схемы хранит `coordinate_unit`, `grid_step_mm` и `index_base: 0`. При первом открытии старого файла без единицы координат приложение определяет её один раз и затем сохраняет явно. Старые схемы с 1-based ID линий и блоков при загрузке переводятся на 0-based; сохранение записывает маркер `index_base: 0`, чтобы повторная загрузка не сдвигала ID. Сохранение выполняется через временный файл и атомарную замену, поэтому исходный JSON не повреждается при ошибке записи. Если у схемы есть подложка, при сохранении её исходное PNG- или JPEG-изображение встраивается в секцию `background.embedded_image` внутри JSON вместе с именем, MIME-типом и контрольной суммой. После этого схема не зависит от внешнего файла, а размеры и параметры подложки сохраняются при редактировании и переносе JSON. Для уже встроенного и не заменённого изображения диалог `Подложка` показывает в поле пути значение `Встроенная`. Чтобы заменить его, нажмите `Выбрать` и укажите новый файл; новое изображение будет встроено при следующем сохранении схемы. В `vm_map.h` используются физические адреса из GUI: -- `strip = line_id - 1` +- `strip = line_id` - `led_id = line_index` Неназначенные точки не попадают в рабочий `VM_MAPPINGS`, но исходные точки могут оставаться в служебных метаданных. @@ -178,10 +184,10 @@ JSON схемы хранит `coordinate_unit` и `grid_step_mm`. При пер .\.venv\Scripts\python.exe -m PyInstaller --noconfirm --clean app\LEDMapper.spec ``` -Готовый файл появится здесь: +В чистой сборке из релизного тега файл будет называться: ```text -dist/LEDMapper.exe +dist/LEDMapper-vX.Y.Z.exe ``` `build/` и `dist/` не коммитятся, потому что это воспроизводимые артефакты сборки. diff --git a/__init__.py b/__init__.py index cd9b1a3..2a58672 100644 --- a/__init__.py +++ b/__init__.py @@ -1 +1,3 @@ """LED Mapper Editor application package.""" + +from .version import __version__ diff --git a/main.py b/main.py index 1105e70..e4b76b6 100644 --- a/main.py +++ b/main.py @@ -3,6 +3,7 @@ from __future__ import annotations import sys from pathlib import Path +from PySide6.QtCore import QTimer from PySide6.QtWidgets import QApplication APP_DIR = Path(__file__).resolve().parent @@ -14,11 +15,13 @@ try: from .main_window import MainWindow from .settings import AppSettings from .theme import apply_app_theme, effective_theme + from .update_service import signal_update_ready except ImportError: from icons import lucide_icon from main_window import MainWindow from settings import AppSettings from theme import apply_app_theme, effective_theme + from update_service import signal_update_ready def main() -> int: @@ -31,6 +34,7 @@ def main() -> int: window = MainWindow() window.setWindowIcon(app_icon) window.show() + QTimer.singleShot(500, signal_update_ready) return app.exec() diff --git a/main_window.py b/main_window.py index 1616790..b48b8ee 100644 --- a/main_window.py +++ b/main_window.py @@ -2,8 +2,9 @@ from __future__ import annotations from copy import deepcopy from pathlib import Path import sys +from threading import Event, Thread -from PySide6.QtCore import QEvent, QSignalBlocker, QSize, Qt, QTimer +from PySide6.QtCore import QEvent, QObject, QSignalBlocker, QSize, Qt, QTimer, Signal from PySide6.QtGui import QAction, QColor, QImageReader, QKeySequence from PySide6.QtWidgets import ( QButtonGroup, @@ -61,6 +62,10 @@ try: from .theme import apply_app_theme, effective_theme from .unfolding import MAX_LINES from .mapping_renderer import render_mapping_image + from .update_service import ( + UpdateCancelled, UpdateError, UpdateInfo, check_for_update, current_executable, current_version, + download_update, launch_update, + ) except ImportError: from background_dialog import BackgroundDialog from canvas import LedCanvas @@ -80,6 +85,10 @@ except ImportError: from theme import apply_app_theme, effective_theme from unfolding import MAX_LINES from mapping_renderer import render_mapping_image + from update_service import ( + UpdateCancelled, UpdateError, UpdateInfo, check_for_update, current_executable, current_version, + download_update, launch_update, + ) ROOT_DIR = Path(__file__).resolve().parents[1] @@ -191,6 +200,13 @@ class GenerationStrategyDialog(QDialog): self.ok_button.setEnabled(True) +class UpdateSignals(QObject): + checked = Signal(object) + downloaded = Signal(object) + progress = Signal(int) + failed = Signal(str) + + class MainWindow(QMainWindow): def __init__(self) -> None: super().__init__() @@ -231,6 +247,16 @@ class MainWindow(QMainWindow): self.menu_buttons_widget: QWidget | None = None self.right_panel: QFrame | None = None self.right_panel_buttons: dict[str, tuple[QToolButton, str]] = {} + self._update_info: UpdateInfo | None = None + self._update_busy = False + self._update_manual_check = False + self._update_cancel = Event() + self._installing_update = False + self._update_signals = UpdateSignals(self) + self._update_signals.checked.connect(self._on_update_checked) + self._update_signals.downloaded.connect(self._on_update_downloaded) + self._update_signals.progress.connect(self._on_update_progress) + self._update_signals.failed.connect(self._on_update_failed) self.canvas = LedCanvas() self._apply_canvas_settings() @@ -321,6 +347,8 @@ class MainWindow(QMainWindow): application.focusChanged.connect(self._on_application_focus_changed) application.installEventFilter(self) self._refresh_ui_state() + if getattr(sys, "frozen", False): + QTimer.singleShot(1200, self.check_updates) def _build_actions(self) -> None: def add_action( @@ -359,6 +387,9 @@ class MainWindow(QMainWindow): "Экспортировать текущую схему маппинга в изображение (PNG, JPEG)" ) add_action("settings", "Параметры", self.open_settings, "settings", QKeySequence("Ctrl+,")) + add_action("check_updates", "Проверить обновления", self.check_updates) + add_action("install_update", "Установить обновление", self.install_update) + self.actions["install_update"].setEnabled(False) add_action( "generate_script", "Сгенерировать скрипт", @@ -529,6 +560,10 @@ class MainWindow(QMainWindow): tools_menu.addSeparator() tools_menu.addAction(self.actions["brush"]) + help_menu = menu_bar.addMenu("Справка") + help_menu.addAction(self.actions["check_updates"]) + help_menu.addAction(self.actions["install_update"]) + # Delete old widget if recreating if hasattr(self, "menu_buttons_widget") and self.menu_buttons_widget is not None: self.menu_buttons_widget.deleteLater() @@ -3059,8 +3094,117 @@ class MainWindow(QMainWindow): ) return answer == QMessageBox.StandardButton.Yes + def check_updates(self) -> None: + if self._update_busy: + return + self._update_manual_check = not bool(getattr(sys, "frozen", False)) or self.sender() is self.actions["check_updates"] + self._update_busy = True + self.actions["check_updates"].setEnabled(False) + self.actions["install_update"].setEnabled(False) + if self._update_manual_check: + self.status.setText("Проверка обновлений...") + + def worker() -> None: + try: + self._update_signals.checked.emit(check_for_update(current_version())) + except (UpdateError, OSError) as exc: + self._update_signals.failed.emit(str(exc)) + + Thread(target=worker, name="ledmapper-update-check", daemon=True).start() + + def _on_update_checked(self, info: UpdateInfo | None) -> None: + self._update_busy = False + self._update_info = info + self.actions["check_updates"].setEnabled(True) + self.actions["install_update"].setEnabled(info is not None and bool(getattr(sys, "frozen", False))) + if info is None: + if self._update_manual_check: + show_information(self, "Обновления", f"Установлена актуальная версия {current_version()}.") + return + self.actions["install_update"].setText(f"Установить обновление {info.tag}") + self.status.setText(f"Доступна версия {info.tag}. Меню «Справка» → «Установить обновление».") + if not getattr(sys, "frozen", False) and self._update_manual_check: + show_information(self, "Обновление", "Автоустановка доступна только в собранном EXE.") + + def install_update(self) -> None: + info = self._update_info + if info is None or self._update_busy: + return + try: + executable = current_executable() + except UpdateError as exc: + show_warning(self, "Обновление", str(exc)) + return + answer = ask_question(self, "Обновление LED Mapper", f"Скачать {info.asset_name} и перезапустить приложение?") + if answer != QMessageBox.StandardButton.Yes: + return + self._update_busy = True + self._update_cancel.clear() + self.actions["check_updates"].setEnabled(False) + self.actions["install_update"].setEnabled(False) + self.status.setText(f"Загрузка {info.asset_name}: 0 %") + + def worker() -> None: + try: + downloaded = download_update( + info, executable.parent, + progress=self._update_signals.progress.emit, + cancelled=self._update_cancel.is_set, + ) + if self._update_cancel.is_set(): + downloaded.unlink(missing_ok=True) + return + self._update_signals.downloaded.emit(downloaded) + except UpdateCancelled: + return + except (UpdateError, OSError) as exc: + self._update_signals.failed.emit(str(exc)) + + Thread(target=worker, name="ledmapper-update-download", daemon=True).start() + + def _on_update_progress(self, percent: int) -> None: + if self._update_info is not None: + self.status.setText(f"Загрузка {self._update_info.asset_name}: {percent} %") + + def _on_update_downloaded(self, downloaded: Path) -> None: + self._update_busy = False + self.actions["check_updates"].setEnabled(True) + self.actions["install_update"].setEnabled(True) + info = self._update_info + if info is None: + downloaded.unlink(missing_ok=True) + return + if self._dirty: + answer = ask_question(self, "Сохранить схему", "Перед перезапуском нужно сохранить изменения схемы. Сохранить сейчас?") + if answer != QMessageBox.StandardButton.Yes: + downloaded.unlink(missing_ok=True) + return + self.save_file() + if self._dirty: + downloaded.unlink(missing_ok=True) + return + try: + launch_update(downloaded, info) + except UpdateError as exc: + downloaded.unlink(missing_ok=True) + show_warning(self, "Обновление", str(exc)) + return + self._installing_update = True + self.close() + + def _on_update_failed(self, message: str) -> None: + self._update_busy = False + self.actions["check_updates"].setEnabled(True) + self.actions["install_update"].setEnabled( + self._update_info is not None and bool(getattr(sys, "frozen", False)) + ) + self.status.setText(f"Ошибка обновления: {message}") + if self._update_manual_check or self._update_info is not None: + show_warning(self, "Обновление", message) + def closeEvent(self, event) -> None: # noqa: ANN001 - if self._confirm_discard_changes("Закрыть приложение и потерять несохраненные изменения текущей карты?"): + if self._installing_update or self._confirm_discard_changes("Закрыть приложение и потерять несохраненные изменения текущей карты?"): + self._update_cancel.set() event.accept() else: event.ignore() diff --git a/models.py b/models.py index 295ed7f..e225b68 100644 --- a/models.py +++ b/models.py @@ -29,6 +29,7 @@ _DOCUMENT_KNOWN_FIELDS = { "circles", "coordinate_unit", "grid_step_mm", + "index_base", "leds", "lines", "unit", @@ -321,8 +322,8 @@ class MappingDocument: self.coordinate_unit = "mm" self.grid_step_mm = DEFAULT_GRID_STEP_MM self.extra_payload: dict[str, Any] = {} - self._next_line_id = 1 - self._next_block_id = 1 + self._next_line_id = 0 + self._next_block_id = 0 def clear(self) -> None: self.path = None @@ -334,8 +335,8 @@ class MappingDocument: self.coordinate_unit = "mm" self.grid_step_mm = DEFAULT_GRID_STEP_MM self.extra_payload.clear() - self._next_line_id = 1 - self._next_block_id = 1 + self._next_line_id = 0 + self._next_block_id = 0 def load(self, path: str | Path) -> None: self.clear() @@ -386,6 +387,7 @@ class MappingDocument: self._canonicalize_coordinate_unit() self._load_lines(data) + self._migrate_legacy_index_base(data) self.background = BackgroundImage.from_payload(data.get("background")) self.unfolding = UnfoldingDescription.from_payload(data.get("unfolding")) self.unfolding.max_lines = MAX_LINES @@ -424,6 +426,41 @@ class MappingDocument: self.lines.append(line) self._next_line_id = max(self._next_line_id, line_id + 1) + def _migrate_legacy_index_base(self, data: dict[str, Any]) -> None: + """Однократно переводит старые 1-based ID, сохраняя физические адреса лент.""" + marker = data.get("index_base") + if marker == 0: + return + if marker not in (None, 1): + raise ValueError("Unsupported mapping index_base; expected 0 or 1.") + identifiers = [line.id for line in self.lines] + identifiers.extend(point.line_id for point in self.points if point.line_id is not None) + if marker is None and (not identifiers or min(identifiers) == 0): + return + if identifiers and min(identifiers) < 1: + return + for line in self.lines: + old_id = line.id + line.id -= 1 + if re.fullmatch(r"(?:Линия|Line)\s+\d+", line.name): + line.name = f"Линия {line.id}" + if line.color == default_color(old_id): + line.color = default_color(line.id) + for block in line.blocks: + block.id -= 1 + if re.fullmatch(r"(?:Блок|Block)\s+\d+", block.name): + block.name = f"Блок {block.id}" + for point in self.points: + if point.line_id is not None: + point.line_id -= 1 + if point.block_id is not None: + point.block_id -= 1 + self._next_line_id = max((line.id for line in self.lines), default=-1) + 1 + self._next_block_id = max( + (block.id for line in self.lines for block in line.blocks), + default=-1, + ) + 1 + def _rebuild_lines_from_points(self) -> None: for line_id in sorted({point.line_id for point in self.points if point.line_id is not None}): line = LedLine(int(line_id), f"Line {line_id}", default_color(int(line_id))) @@ -445,7 +482,7 @@ class MappingDocument: if point.line_id == line_id ] if ids: - line.blocks.append(LedBlock(self._next_block_id, "Block 1", ids)) + line.blocks.append(LedBlock(self._next_block_id, f"Блок {self._next_block_id}", ids)) self._next_block_id += 1 for block_id in block_ids: ids = [ @@ -513,6 +550,7 @@ class MappingDocument: payload["unfolding"] = self.describe_unfolding().to_payload() payload["coordinate_unit"] = self._coordinate_unit_for_save() payload["grid_step_mm"] = self.grid_step_mm + payload["index_base"] = 0 payload["background"] = (self.background if background is None else background).to_payload() payload["lines"] = self.lines_to_payload() if lines_payload is None else lines_payload return payload @@ -655,7 +693,7 @@ class MappingDocument: point_id for point_id in point_ids if 0 <= point_id < len(self.points) ]) self._detach_points_from_blocks(clean_point_ids) - block = LedBlock(len(line.blocks) + 1, f"Блок {len(line.blocks) + 1}", clean_point_ids) + block = LedBlock(len(line.blocks), f"Блок {len(line.blocks)}", clean_point_ids) insert_index = len(line.blocks) if after_block_id is not None: for index, existing_block in enumerate(line.blocks): @@ -799,14 +837,14 @@ class MappingDocument: assigned = sum(1 for point in self.points if point.line_id is not None or point.block_id is not None) self.lines.clear() self._clear_points([point.uid for point in self.points]) - self._next_line_id = 1 - self._next_block_id = 1 + self._next_line_id = 0 + self._next_block_id = 0 return assigned def renumber_lines(self) -> None: self.order_lines_clockwise() id_map: dict[int, int] = {} - for new_id, line in enumerate(self.lines, start=1): + for new_id, line in enumerate(self.lines): old_id = line.id id_map[old_id] = new_id line.id = new_id @@ -818,7 +856,7 @@ class MappingDocument: if point.line_id in id_map: point.line_id = id_map[point.line_id] - self._next_line_id = len(self.lines) + 1 + self._next_line_id = len(self.lines) def _validate_line_limit(self) -> None: line_limit = self._line_limit() @@ -985,7 +1023,7 @@ class MappingDocument: start_id = start_ids[0] if start_ids else point_ids[0] groups = self.route_groups_from_points(point_ids, start_id) line.blocks.clear() - for block_number, group in enumerate(groups, start=1): + for block_number, group in enumerate(groups): line.blocks.append(LedBlock(block_number, f"Блок {block_number}", group)) self.reindex_all() return len(groups) @@ -997,7 +1035,7 @@ class MappingDocument: self._detach_points_from_blocks([point_id for group in groups for point_id in group]) line = self.get_line(line_id) for group_ids in groups: - block_number = len(line.blocks) + 1 + block_number = len(line.blocks) block = LedBlock(block_number, f"Блок {block_number}", unique_in_order(group_ids)) line.blocks.append(block) self.order_blocks_clockwise(line_id, reorder_leds=False) @@ -1107,7 +1145,7 @@ class MappingDocument: @staticmethod def renumber_blocks(line: LedLine) -> None: - for block_number, block in enumerate(line.blocks, start=1): + for block_number, block in enumerate(line.blocks): block.id = block_number if not block.name or re.match(r"^(Блок|Block)\s+\d+$", block.name): block.name = f"Блок {block_number}" @@ -1216,7 +1254,7 @@ def _correlation_sign(value: float) -> int: def default_color(line_id: int) -> str: colors = ["#e11d48", "#2563eb", "#16a34a", "#f59e0b", "#7c3aed", "#0891b2"] - return colors[(line_id - 1) % len(colors)] + return colors[line_id % len(colors)] def distance(a: LedPoint, b: LedPoint) -> float: diff --git a/panel_mapping_generator.py b/panel_mapping_generator.py index 24767e4..3814252 100644 --- a/panel_mapping_generator.py +++ b/panel_mapping_generator.py @@ -205,7 +205,7 @@ def normalize_document_points(document: MappingDocument) -> list[PanelLedPoint]: generated_id=index, source_id=point.source_id, input_index=point.uid, - strip=(point.line_id - 1) if point.line_id is not None else None, + strip=point.line_id, led_id=point.line_index, x_mm=point.x * scale, y_mm=point.y * scale, @@ -586,6 +586,8 @@ def write_readme(path: Path) -> None: "- `validation_report.txt` — результаты проверки маршрутов и ячеек.", "- `README.md` — описание набора файлов.", "", + "Физический адрес ленты в `vm_map.h`: `strip = line_id`; нумерация начинается с `0`.", + "", "Устаревший `virtual_matrix_legacy.ino.txt` больше не создаётся.", "", ] diff --git a/qa_smoke.py b/qa_smoke.py index b75e3eb..c9495f2 100644 --- a/qa_smoke.py +++ b/qa_smoke.py @@ -4,6 +4,7 @@ import base64 import hashlib import json import importlib +import io import gc import re import subprocess @@ -13,6 +14,7 @@ import time from copy import deepcopy from pathlib import Path from typing import Callable +from unittest.mock import patch if __package__ is None or __package__ == "": sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) @@ -29,6 +31,8 @@ from app.models import ( order_points_like_grid_svg, ) from app.panel_mapping_generator import generate_panel_mapping_from_document +from app.update_service import UpdateError, check_for_update, download_update +import app.update_service as updater from app.unfolding import MAX_LINES try: import app.canvas as canvas_module @@ -99,6 +103,7 @@ def main() -> None: test_utf16_le_json_without_bom_is_converted_to_utf8_on_load, test_line_limit_rejects_fifth_line, test_line_indexes_are_scoped_per_line, + test_legacy_index_base_migration_roundtrip, test_remove_selected_leds_blocks_and_lines, test_remove_blocks_across_lines_is_atomic, test_remove_block_preserves_line_numbers, @@ -139,6 +144,8 @@ def main() -> None: test_saved_coordinate_unit_is_reused, test_uid_point_id_mode_does_not_fallback_to_source_ids, test_panel_mapping_generation_from_document, + test_update_release_validation_and_download, + test_update_install_keeps_executable_path_and_bundled_version, test_ordered_json_meter_coordinates_reload_as_mm, test_legacy_cleanup_waits_for_success, test_standalone_panel_mapping_generation_without_legacy_ino, @@ -186,6 +193,7 @@ def main() -> None: test_canvas_rebuild_drops_deleted_selection_items, test_canvas_refresh_does_not_accumulate_connection_items, test_main_window_has_classic_menu_bar, + test_update_actions_are_available_in_help_menu, test_classic_menu_contains_expected_actions, test_side_panel_hides_removed_tools, test_undo_redo_roundtrip_for_line_creation, @@ -390,6 +398,42 @@ def test_line_indexes_are_scoped_per_line() -> None: assert sorted(indexes_by_line.values()) == [[0, 1], [0, 1, 2]] +def test_legacy_index_base_migration_roundtrip() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + source = Path(temp_dir) / "legacy.json" + saved = Path(temp_dir) / "migrated.json" + points = fixture_points() + for index in (0, 1): + points[index].update(line_id=1, block_id=1, line_index=index, block_index=index) + for index in (4, 5): + points[index].update(line_id=2, block_id=1, line_index=index - 4, block_index=index - 4) + source.write_text(json.dumps({ + "circles": points, + "unknown_field": {"retained": True}, + "lines": [ + {"id": 1, "name": "Линия 1", "blocks": [{"id": 1, "name": "Блок 1", "point_ids": [0, 1]}]}, + {"id": 2, "name": "Линия 2", "color": "#123456", "blocks": [{"id": 1, "name": "Блок 1", "point_ids": [4, 5]}]}, + ], + }), encoding="utf-8") + doc = MappingDocument() + doc.load(source) + assert [line.id for line in doc.lines] == [0, 1] + assert [line.name for line in doc.lines] == ["Линия 0", "Линия 1"] + assert doc.lines[1].color == "#123456" + assert [line.blocks[0].id for line in doc.lines] == [0, 0] + assert [doc.points[index].line_id for index in (0, 1, 4, 5)] == [0, 0, 1, 1] + assert [doc.points[index].block_id for index in (0, 1, 4, 5)] == [0, 0, 0, 0] + generated = generate_panel_mapping_from_document(doc, Path(temp_dir) / "generated") + ordered = json.loads(generated.generated_files["ordered_json"].read_text(encoding="utf-8")) + assert sorted({point["strip"] for point in ordered["circles"] if point["strip"] is not None}) == [0, 1] + doc.save(saved) + assert json.loads(saved.read_text(encoding="utf-8"))["index_base"] == 0 + again = MappingDocument() + again.load(saved) + assert again.to_payload() == doc.to_payload() + assert again.extra_payload["unknown_field"] == {"retained": True} + + def test_remove_selected_leds_blocks_and_lines() -> None: with temp_document_path() as path: doc = MappingDocument() @@ -467,8 +511,8 @@ def test_remove_block_preserves_line_numbers() -> None: removed = doc.remove_block(second_line.id, second_block.id) assert removed == 2 - assert [line.id for line in doc.lines] == [1, 2] - assert [line.name for line in doc.lines] == ["Линия 1", "Линия 2"] + assert [line.id for line in doc.lines] == [0, 1] + assert [line.name for line in doc.lines] == ["Линия 0", "Линия 1"] assert doc.get_line(second_line.id).blocks == [] @@ -581,8 +625,8 @@ def test_line_numbers_do_not_renumber_when_points_are_cleared() -> None: doc.clear_point_assignments([0, 1]) - assert [line.id for line in doc.lines] == [1, 2] - assert [line.name for line in doc.lines] == ["Линия 1", "Линия 2"] + assert [line.id for line in doc.lines] == [0, 1] + assert [line.name for line in doc.lines] == ["Линия 0", "Линия 1"] def test_auto_blocks_reindex_all_affected_lines() -> None: @@ -615,8 +659,8 @@ def test_legacy_rebuild_groups_by_line_and_block() -> None: doc.load(path) assert len(doc.lines) == 2 assert [len(line.blocks) for line in doc.lines] == [1, 1] - assert sorted(doc.points[point_id].line_id for point_id in doc.lines[0].blocks[0].point_ids) == [1, 1] - assert sorted(doc.points[point_id].line_id for point_id in doc.lines[1].blocks[0].point_ids) == [2, 2] + assert sorted(doc.points[point_id].line_id for point_id in doc.lines[0].blocks[0].point_ids) == [0, 0] + assert sorted(doc.points[point_id].line_id for point_id in doc.lines[1].blocks[0].point_ids) == [1, 1] def test_legacy_line_without_block_id_is_preserved() -> None: @@ -815,8 +859,8 @@ def test_remove_points_reindexes_uids_and_blocks() -> None: assert [point.source_id for point in doc.points[:4]] == [0, 2, 3, 5] assert first.point_ids == [0, 1] assert second.point_ids == [2, 3] - assert [doc.points[point_id].block_id for point_id in first.point_ids] == [1, 1] - assert [doc.points[point_id].block_id for point_id in second.point_ids] == [2, 2] + assert [doc.points[point_id].block_id for point_id in first.point_ids] == [0, 0] + assert [doc.points[point_id].block_id for point_id in second.point_ids] == [1, 1] def test_loaded_duplicate_memberships_are_deduplicated() -> None: @@ -870,21 +914,21 @@ def test_block_numbers_renumber_after_block_changes() -> None: first = doc.add_block(line.id, []) second = doc.add_block(line.id, []) third = doc.add_block(line.id, []) - assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 1", "Блок 2", "Блок 3"] + assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 0", "Блок 1", "Блок 2"] doc.reindex_all() - assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 1", "Блок 2", "Блок 3"] + assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 0", "Блок 1", "Блок 2"] doc.remove_block(line.id, second.id) - assert [(block.id, block.name) for block in doc.get_line(line.id).blocks] == [(1, "Блок 1"), (2, "Блок 2")] + assert [(block.id, block.name) for block in doc.get_line(line.id).blocks] == [(0, "Блок 0"), (1, "Блок 1")] assert doc.points[0].block_id is None - assert third.id == 2 + assert third.id == 1 fourth = doc.add_block(line.id, []) - assert fourth.id == 3 - assert fourth.name == "Блок 3" - assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 1", "Блок 2", "Блок 3"] + assert fourth.id == 2 + assert fourth.name == "Блок 2" + assert [block.name for block in doc.get_line(line.id).blocks] == ["Блок 0", "Блок 1", "Блок 2"] def test_block_numbers_are_local_to_each_line() -> None: @@ -898,14 +942,14 @@ def test_block_numbers_are_local_to_each_line() -> None: third = doc.add_block(first_line.id, []) other_line_block = doc.add_block(second_line.id, [0]) - assert [block.id for block in doc.get_line(first_line.id).blocks] == [1, 2, 3] - assert other_line_block.id == 1 + assert [block.id for block in doc.get_line(first_line.id).blocks] == [0, 1, 2] + assert other_line_block.id == 0 doc.remove_block(first_line.id, second.id) - assert [(block.id, block.name) for block in doc.get_line(first_line.id).blocks] == [(1, "Блок 1"), (2, "Блок 2")] + assert [(block.id, block.name) for block in doc.get_line(first_line.id).blocks] == [(0, "Блок 0"), (1, "Блок 1")] doc.remove_line(second_line.id) - assert [(block.id, block.name) for block in doc.get_line(first_line.id).blocks] == [(1, "Блок 1"), (2, "Блок 2")] - assert [first.name, third.name] == ["Блок 1", "Блок 2"] + assert [(block.id, block.name) for block in doc.get_line(first_line.id).blocks] == [(0, "Блок 0"), (1, "Блок 1")] + assert [first.name, third.name] == ["Блок 0", "Блок 1"] def test_reindex_preserves_block_point_order() -> None: @@ -968,15 +1012,15 @@ def test_add_block_can_insert_after_selected_block_and_renumber() -> None: inserted = doc.add_block(line.id, [2, 3], after_block_id=first.id) assert [(block.id, block.name, block.point_ids) for block in line.blocks] == [ - (1, "Блок 1", [0, 1]), - (2, "Блок 2", [2, 3]), - (3, "Блок 3", [4, 5]), - (4, "Блок 4", [8, 9]), + (0, "Блок 0", [0, 1]), + (1, "Блок 1", [2, 3]), + (2, "Блок 2", [4, 5]), + (3, "Блок 3", [8, 9]), ] - assert inserted.id == 2 - assert second.id == 3 - assert third.id == 4 - assert [doc.points[point_id].block_id for point_id in [0, 2, 4, 8]] == [1, 2, 3, 4] + assert inserted.id == 1 + assert second.id == 2 + assert third.id == 3 + assert [doc.points[point_id].block_id for point_id in [0, 2, 4, 8]] == [0, 1, 2, 3] assert [doc.points[point_id].line_index for point_id in [0, 1, 2, 3, 4, 5, 8, 9]] == list(range(8)) @@ -1244,6 +1288,95 @@ def test_panel_mapping_generation_from_document() -> None: assert ordered_payload["circles"][0]["strip"] == 0 +def test_update_release_validation_and_download() -> None: + class FakeResponse(io.BytesIO): + def __init__(self, data: bytes, url: str) -> None: + super().__init__(data) + self.url = url + + def geturl(self) -> str: + return self.url + + tag = "v3.2.1" + name = f"LEDMapper-{tag}.exe" + url = f"https://git.robotail.pro/Robotail/LED-mapper/releases/download/{tag}/{name}" + executable = bytearray(256) + executable[:2] = b"MZ" + executable[0x3C:0x40] = (128).to_bytes(4, "little") + executable[128:134] = b"PE\x00\x00\x64\x86" + release = { + "tag_name": tag, + "draft": False, + "prerelease": False, + "assets": [{"name": name, "size": len(executable), "browser_download_url": url}], + } + + def release_opener(_request, **_kwargs): + return FakeResponse(json.dumps(release).encode("utf-8"), "https://git.robotail.pro/api/v1/repos/Robotail/LED-mapper/releases/latest") + + assert check_for_update("v3.2.1", opener=release_opener) is None + info = check_for_update("3.2.0", opener=release_opener) + assert info is not None and info.asset_name == name and info.version == "3.2.1" + with tempfile.TemporaryDirectory() as temp_dir: + downloaded = download_update(info, Path(temp_dir), opener=lambda _request, **_kwargs: FakeResponse(bytes(executable), url)) + assert downloaded.read_bytes() == bytes(executable) + downloaded.unlink() + try: + download_update(info, Path(temp_dir), opener=lambda _request, **_kwargs: FakeResponse(b"bad", url)) + except UpdateError: + assert not list(Path(temp_dir).glob(".ledmapper-update-*")) + else: + raise AssertionError("A truncated update must be rejected and removed") + release["assets"][0]["browser_download_url"] = "https://untrusted.example/update.exe" + try: + check_for_update("3.2.0", opener=release_opener) + except UpdateError: + pass + else: + raise AssertionError("An update on another host must be rejected") + + +def test_update_install_keeps_executable_path_and_bundled_version() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + directory = Path(temp_dir) + old_executable = directory / "LEDMapper-v3.2.0.exe" + staged = directory / ".ledmapper-update-test.download" + payload = bytearray(256) + payload[:2] = b"MZ" + payload[0x3C:0x40] = (128).to_bytes(4, "little") + payload[128:134] = b"PE\x00\x00\x64\x86" + old_executable.write_bytes(payload) + staged.write_bytes(payload) + (directory / "ledmapper-build-version.txt").write_text("v3.2.1", encoding="utf-8") + info = updater.UpdateInfo( + "v3.2.1", "3.2.1", "LEDMapper-v3.2.1.exe", + "https://git.robotail.pro/Robotail/LED-mapper/releases/download/v3.2.1/LEDMapper-v3.2.1.exe", + len(payload), + ) + with patch.object(updater.sys, "frozen", True, create=True), \ + patch.object(updater.sys, "_MEIPASS", temp_dir, create=True), \ + patch.object(updater.sys, "executable", str(old_executable)), \ + patch.object(updater.subprocess, "Popen") as popen: + assert updater.current_version() == "v3.2.1" + updater.launch_update(staged, info, process_id=123) + arguments = popen.call_args.args[0] + assert arguments[8] == arguments[9] == str(old_executable) + helper = Path(arguments[5]) + script = helper.read_text(encoding="utf-8-sig") + assert "Wait-Process" in script and "$moveDeadline" in script + assert "Move-Item -LiteralPath $Backup" in script + quoted_path = str(helper).replace("'", "''") + parser_command = ( + "$errors = @(); " + f"[System.Management.Automation.Language.Parser]::ParseFile('{quoted_path}', [ref]$null, [ref]$errors) | Out-Null; " + "if ($errors.Count) { exit 1 }" + ) + subprocess.run( + ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", parser_command], + check=True, stdout=subprocess.DEVNULL, + ) + + def test_ordered_json_meter_coordinates_reload_as_mm() -> None: with tempfile.TemporaryDirectory() as temp_dir: doc = MappingDocument() @@ -2675,6 +2808,31 @@ def test_main_window_has_classic_menu_bar() -> None: window.close() +def test_update_actions_are_available_in_help_menu() -> None: + if MainWindow is None or QApplication is None: + return + app = QApplication.instance() or QApplication([]) + window = MainWindow() + app.processEvents() + top_actions = list(window.menuBar().actions()) + help_action = next(action for action in top_actions if action.text() == "Справка") + help_menu = help_action.menu() + assert help_menu is not None + assert [action.text() for action in help_menu.actions()] == ["Проверить обновления", "Установить обновление"] + assert window.actions["check_updates"].isEnabled() + assert not window.actions["install_update"].isEnabled() + info = updater.UpdateInfo( + "v3.2.1", "3.2.1", "LEDMapper-v3.2.1.exe", + "https://git.robotail.pro/Robotail/LED-mapper/releases/download/v3.2.1/LEDMapper-v3.2.1.exe", + 1024, + ) + with patch.object(sys, "frozen", True, create=True): + window._on_update_checked(info) + assert window.actions["install_update"].isEnabled() + assert "v3.2.1" in window.actions["install_update"].text() + window.close() + + def test_classic_menu_contains_expected_actions() -> None: if MainWindow is None or QApplication is None: return @@ -3938,14 +4096,14 @@ def test_brush_creates_new_block_after_selected_tree_block() -> None: line_after = window.document.get_line(line.id) assert [(block.id, block.point_ids) for block in line_after.blocks] == [ - (1, [0, 1]), - (2, [2, 3]), - (3, [4, 5]), + (0, [0, 1]), + (1, [2, 3]), + (2, [4, 5]), ] - assert [window.document.points[point_id].block_id for point_id in [0, 2, 4]] == [1, 2, 3] + assert [window.document.points[point_id].block_id for point_id in [0, 2, 4]] == [0, 1, 2] current = window.tree.currentItem() assert current is not None - assert current.data(0, Qt.UserRole) == ("block", line.id, 2) + assert current.data(0, Qt.UserRole) == ("block", line.id, 1) assert window.canvas.selected_point_ids() == [2, 3] window._dirty = False window.close() @@ -4125,6 +4283,8 @@ def test_pyinstaller_spec_build_ref_rules() -> None: (repo / "tracked.txt").write_text("clean\n", encoding="utf-8") subprocess.run(["git", "add", "tracked.txt"], cwd=repo, check=True) subprocess.run(["git", "commit", "-m", "initial"], cwd=repo, check=True, stdout=subprocess.DEVNULL) + subprocess.run(["git", "add", "runner.py"], cwd=repo, check=True) + subprocess.run(["git", "commit", "-m", "track runner"], cwd=repo, check=True, stdout=subprocess.DEVNULL) commit = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"], cwd=repo, text=True).strip() clean_ref = subprocess.check_output([sys.executable, str(runner)], cwd=repo, text=True).strip() diff --git a/update_service.py b/update_service.py new file mode 100644 index 0000000..9dfd936 --- /dev/null +++ b/update_service.py @@ -0,0 +1,326 @@ +"""Безопасная проверка и установка Windows-релизов LED Mapper.""" + +from __future__ import annotations + +import json +import os +import platform +import re +import subprocess +import sys +import tempfile +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO, Callable +from urllib.parse import urlparse +from urllib.request import Request, urlopen + +try: + from .version import __version__ +except ImportError: + from version import __version__ + +RELEASE_API_URL = "https://git.robotail.pro/api/v1/repos/Robotail/LED-mapper/releases/latest" +RELEASE_PAGE_URL = "https://git.robotail.pro/Robotail/LED-mapper/releases" +RELEASE_HOST = "git.robotail.pro" +READY_ENV = "LEDMAPPER_UPDATE_READY_FILE" +MAX_METADATA_BYTES = 1024 * 1024 +MAX_DOWNLOAD_BYTES = 200 * 1024 * 1024 +CHUNK_BYTES = 1024 * 1024 +VERSION_RE = re.compile(r"v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)\Z") +EXECUTABLE_RE = re.compile(r"LEDMapper-(v\d+\.\d+\.\d+)\.exe\Z", re.IGNORECASE) + + +class UpdateError(RuntimeError): + """Проверка, загрузка или установка обновления не выполнена.""" + + +class UpdateCancelled(UpdateError): + """Загрузка отменена пользователем.""" + + +@dataclass(frozen=True) +class UpdateInfo: + tag: str + version: str + asset_name: str + download_url: str + size: int + + +def parse_version(value: str) -> tuple[int, int, int]: + match = VERSION_RE.fullmatch(value.strip()) + if match is None: + raise UpdateError("Версия релиза должна иметь формат vX.Y.Z") + return tuple(int(part) for part in match.groups()) + + +def current_version() -> str: + if getattr(sys, "frozen", False): + bundle_dir = getattr(sys, "_MEIPASS", None) + if bundle_dir is not None: + try: + bundled_version = (Path(bundle_dir) / "ledmapper-build-version.txt").read_text(encoding="utf-8").strip() + if VERSION_RE.fullmatch(bundled_version): + return bundled_version + except OSError: + pass + match = EXECUTABLE_RE.fullmatch(Path(sys.executable).name) + if match is not None: + return match.group(1) + return __version__ + + +def _trusted_url(value: str) -> bool: + parsed = urlparse(value) + return ( + parsed.scheme == "https" + and parsed.hostname == RELEASE_HOST + and parsed.port in (None, 443) + and not parsed.username + and not parsed.password + and not parsed.fragment + ) + + +def check_for_update( + installed_version: str, + *, + opener: Callable[..., BinaryIO] | None = None, + timeout: float = 10.0, +) -> UpdateInfo | None: + """Возвращает только новый релиз с ровно одним подходящим Windows EXE.""" + installed = parse_version(installed_version) + if platform.system() != "Windows": + raise UpdateError("Автообновление доступно только в Windows") + request = Request( + RELEASE_API_URL, + headers={"Accept": "application/json", "User-Agent": "LEDMapper-Updater"}, + ) + try: + with (opener or urlopen)(request, timeout=timeout) as response: + if not _trusted_url(response.geturl()): + raise UpdateError("Сервер релизов перенаправил запрос на другой сайт") + payload = response.read(MAX_METADATA_BYTES + 1) + if len(payload) > MAX_METADATA_BYTES: + raise UpdateError("Ответ сервера релизов слишком велик") + release = json.loads(payload) + except UpdateError: + raise + except (OSError, ValueError, TypeError) as exc: + raise UpdateError(f"Не удалось проверить обновления: {exc}") from exc + if not isinstance(release, dict) or release.get("draft") or release.get("prerelease"): + raise UpdateError("Сервер не вернул опубликованный стабильный релиз") + tag = release.get("tag_name") + if not isinstance(tag, str) or not tag.startswith("v"): + raise UpdateError("У релиза отсутствует корректный тег") + remote = parse_version(tag) + if remote <= installed: + return None + expected_name = f"LEDMapper-{tag}.exe" + assets = release.get("assets") + if not isinstance(assets, list): + raise UpdateError("У релиза отсутствует список артефактов") + matching = [asset for asset in assets if isinstance(asset, dict) and asset.get("name") == expected_name] + if len(matching) != 1: + raise UpdateError(f"Не найден Windows-артефакт {expected_name}") + asset = matching[0] + download_url = asset.get("browser_download_url") + expected_path = f"/Robotail/LED-mapper/releases/download/{tag}/{expected_name}" + if not isinstance(download_url, str) or not _trusted_url(download_url) or urlparse(download_url).path != expected_path: + raise UpdateError("Релиз содержит неожиданный адрес артефакта") + size = asset.get("size") + if isinstance(size, bool) or not isinstance(size, int) or not 0 < size <= MAX_DOWNLOAD_BYTES: + raise UpdateError("Релиз содержит недопустимый размер артефакта") + return UpdateInfo(tag, tag.removeprefix("v"), expected_name, download_url, size) + + +def _check_executable(path: Path) -> None: + with path.open("rb") as source: + header = source.read(64) + if len(header) < 64 or header[:2] != b"MZ": + raise UpdateError("Скачанный файл не является Windows EXE") + pe_offset = int.from_bytes(header[0x3C:0x40], "little") + if not 64 <= pe_offset <= 1024 * 1024: + raise UpdateError("Некорректный заголовок Windows EXE") + source.seek(pe_offset) + if source.read(4) != b"PE\x00\x00": + raise UpdateError("Некорректный заголовок Windows EXE") + if source.read(2) != b"\x64\x86": + raise UpdateError("Обновление не предназначено для Windows x64") + + +def download_update( + info: UpdateInfo, + directory: Path, + *, + opener: Callable[..., BinaryIO] | None = None, + progress: Callable[[int], None] | None = None, + cancelled: Callable[[], bool] | None = None, + timeout: float = 30.0, +) -> Path: + """Скачивает артефакт в отдельный временный файл и проверяет размер/PE.""" + target_dir = directory.resolve() + if not target_dir.is_dir(): + raise UpdateError("Каталог приложения не найден") + if not _trusted_url(info.download_url) or urlparse(info.download_url).path != ( + f"/Robotail/LED-mapper/releases/download/{info.tag}/{info.asset_name}" + ): + raise UpdateError("Недопустимый адрес загрузки обновления") + path: Path | None = None + received = 0 + request = Request(info.download_url, headers={"User-Agent": "LEDMapper-Updater"}) + try: + with (opener or urlopen)(request, timeout=timeout) as response: + if not _trusted_url(response.geturl()): + raise UpdateError("Загрузка перенаправлена на другой сайт") + with tempfile.NamedTemporaryFile( + mode="wb", dir=target_dir, prefix=".ledmapper-update-", suffix=".download", delete=False + ) as output: + path = Path(output.name) + while True: + if cancelled is not None and cancelled(): + raise UpdateCancelled("Загрузка обновления отменена") + chunk = response.read(CHUNK_BYTES) + if not chunk: + break + received += len(chunk) + if received > MAX_DOWNLOAD_BYTES or received > info.size: + raise UpdateError("Артефакт превышает объявленный размер") + output.write(chunk) + if progress is not None: + progress(min(received * 100 // info.size, 100)) + output.flush() + os.fsync(output.fileno()) + if received != info.size: + raise UpdateError("Артефакт скачан не полностью") + _check_executable(path) + if progress is not None: + progress(100) + return path + except (UpdateError, OSError) as exc: + if path is not None: + path.unlink(missing_ok=True) + if isinstance(exc, UpdateError): + raise + raise UpdateError(f"Не удалось скачать обновление: {exc}") from exc + + +def current_executable() -> Path: + if not getattr(sys, "frozen", False): + raise UpdateError("Установка обновления доступна только в собранном EXE") + return Path(sys.executable).resolve() + + +def signal_update_ready() -> None: + marker = os.environ.pop(READY_ENV, None) + if marker is None: + return + try: + path = Path(marker).resolve() + if path.parent != current_executable().parent or not path.name.startswith(".ledmapper-ready-"): + return + path.write_text(str(os.getpid()), encoding="ascii") + except (OSError, UpdateError): + pass + + +def _write_windows_updater(path: Path) -> None: + script = r'''param( + [int]$ApplicationProcessId, + [string]$Source, + [string]$OldTarget, + [string]$NewTarget, + [string]$Backup, + [string]$ReadyFile +) +$ErrorActionPreference = "Stop" +$applicationDirectory = Split-Path -Parent $OldTarget +$errorLog = Join-Path $applicationDirectory "update-error.log" +$restarted = $null +$oldMoved = $false +$newMoved = $false +try { + Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue + # Родительский процесс PyInstaller может ещё удерживать EXE при очистке. + $moveDeadline = (Get-Date).AddSeconds(30) + while (-not $oldMoved) { + try { + Move-Item -LiteralPath $OldTarget -Destination $Backup -ErrorAction Stop + $oldMoved = $true + } catch { + if ((Get-Date) -ge $moveDeadline) { throw } + Start-Sleep -Milliseconds 250 + } + } + if (Test-Path -LiteralPath $NewTarget) { throw "Целевой EXE уже существует" } + Move-Item -LiteralPath $Source -Destination $NewTarget -ErrorAction Stop + $newMoved = $true + $env:LEDMAPPER_UPDATE_READY_FILE = $ReadyFile + $env:PYINSTALLER_RESET_ENVIRONMENT = "1" + $restarted = Start-Process -FilePath $NewTarget -WorkingDirectory $applicationDirectory -PassThru + $deadline = (Get-Date).AddSeconds(60) + while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) { + if ($restarted.HasExited) { throw "Новая версия завершилась до открытия окна" } + if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" } + Start-Sleep -Milliseconds 250 + } + if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" } + Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue + Remove-Item -LiteralPath $Backup -Force -ErrorAction SilentlyContinue + Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue +} catch { + $failure = $_.ToString() + if ($restarted -and -not $restarted.HasExited) { + & taskkill.exe /PID $restarted.Id /T /F *> $null + } + if ($newMoved -and (Test-Path -LiteralPath $NewTarget)) { + Remove-Item -LiteralPath $NewTarget -Force -ErrorAction SilentlyContinue + } + if ($oldMoved -and (Test-Path -LiteralPath $Backup)) { + Move-Item -LiteralPath $Backup -Destination $OldTarget -Force -ErrorAction SilentlyContinue + } + Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue + $failure | Out-File -LiteralPath $errorLog -Encoding utf8 + Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue + exit 1 +} +''' + path.write_text(script, encoding="utf-8-sig", newline="\r\n") + + +def launch_update(downloaded: Path, info: UpdateInfo, *, process_id: int | None = None) -> None: + """Запускает отдельный helper; замену EXE он делает после выхода приложения.""" + old_target = current_executable() + source = downloaded.resolve() + if platform.system() != "Windows": + raise UpdateError("Установка обновления доступна только в Windows") + if source.parent != old_target.parent or not source.is_file() or not source.name.startswith(".ledmapper-update-"): + raise UpdateError("Скачанный файл должен находиться рядом с приложением") + if not EXECUTABLE_RE.fullmatch(info.asset_name): + raise UpdateError("Недопустимое имя Windows-артефакта") + new_target = old_target + if source.stat().st_size != info.size: + raise UpdateError("Размер скачанного артефакта изменился") + _check_executable(source) + token = uuid.uuid4().hex + script_path = old_target.parent / f".ledmapper-updater-{token}.ps1" + backup_path = old_target.parent / f".ledmapper-backup-{token}.exe" + ready_path = old_target.parent / f".ledmapper-ready-{token}" + try: + _write_windows_updater(script_path) + subprocess.Popen( + [ + "powershell.exe", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", + str(script_path), str(process_id or os.getpid()), str(source), str(old_target), + str(new_target), str(backup_path), str(ready_path), + ], + close_fds=True, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except (OSError, UpdateError) as exc: + script_path.unlink(missing_ok=True) + if isinstance(exc, UpdateError): + raise + raise UpdateError(f"Не удалось запустить установщик: {exc}") from exc diff --git a/version.py b/version.py new file mode 100644 index 0000000..c009da0 --- /dev/null +++ b/version.py @@ -0,0 +1,3 @@ +"""Текущая версия приложения; обновляется при публикации релиза.""" + +__version__ = "3.3.0"