"""启动阶段的强制升级进度窗口。""" from __future__ import annotations import os from pathlib import Path from PySide6.QtCore import QObject, QThread, Signal, Slot from PySide6.QtWidgets import ( QDialog, QHBoxLayout, QLabel, QProgressBar, QPushButton, QVBoxLayout, ) from .. import appconfig, diagnostics, update_installer, updater_entry def _format_bytes(value): value = max(0, int(value or 0)) if value >= 1024 * 1024: return "%.1f MB" % (value / (1024 * 1024)) if value >= 1024: return "%.1f KB" % (value / 1024) return "%d B" % value class UpdatePreparationWorker(QObject): stage_changed = Signal(str) progress_changed = Signal(int, int) succeeded = Signal(object) failed = Signal(str) finished = Signal() def __init__(self, result, install_root, prepare=None): super().__init__() self.result = result self.install_root = install_root self.prepare = prepare or update_installer.prepare_update self._cancelled = False @Slot() def run(self): try: metadata = update_installer.metadata_from_update_info(self.result) staged = self.prepare( metadata, self.install_root, cancelled=lambda: self._cancelled, progress=self.progress_changed.emit, stage_callback=self.stage_changed.emit, ) if self._cancelled: raise update_installer.UpdateCancelled("已取消下载新版") self.succeeded.emit(staged) except update_installer.UpdateCancelled: self.failed.emit("已取消下载新版") except update_installer.UpdateInstallError as exc: self.failed.emit(str(exc)) except Exception as exc: try: diagnostics.write_diagnostic_log( "准备自动更新失败", level="ERROR", step="startup_update_prepare", exc=exc, ) except Exception: pass self.failed.emit("准备新版失败，请稍后重试") finally: self.finished.emit() def cancel(self): self._cancelled = True class ForcedUpdateDialog(QDialog): def __init__( self, result, *, parent=None, install_root=None, worker_factory=None, updater_launcher=None, ): super().__init__(parent) self.result = result self.install_root = Path(install_root or appconfig.app_base_dir()).resolve() self.worker_factory = worker_factory or UpdatePreparationWorker self.updater_launcher = updater_launcher or updater_entry.copy_and_launch_updater self.thread = None self.worker = None self.exit_requested = False self.update_started = False self._build_ui() self._validate_available_update() def _build_ui(self): self.setWindowTitle("必须升级") self.setModal(True) self.setMinimumWidth(460) layout = QVBoxLayout(self) title = QLabel("当前版本已不能继续使用，请先升级到新版。") title.setWordWrap(True) layout.addWidget(title) online_version = self.result.latest_version or self.result.min_supported_version or "未知" details = "当前版本：%s\n线上版本：%s" % ( self.result.current_version, online_version, ) if self.result.message: details += "\n升级说明：%s" % self.result.message self.details_label = QLabel(details) self.details_label.setWordWrap(True) layout.addWidget(self.details_label) self.stage_label = QLabel("发现必须升级的新版本") layout.addWidget(self.stage_label) self.progress_bar = QProgressBar() self.progress_bar.setRange(0, 100) self.progress_bar.setValue(0) layout.addWidget(self.progress_bar) self.bytes_label = QLabel("等待开始") layout.addWidget(self.bytes_label) buttons = QHBoxLayout() buttons.addStretch(1) self.action_button = QPushButton("立即升级") self.exit_button = QPushButton("退出程序") self.action_button.clicked.connect(self.start_update) self.exit_button.clicked.connect(self.request_exit) buttons.addWidget(self.action_button) buttons.addWidget(self.exit_button) layout.addLayout(buttons) def _validate_available_update(self): if self.result.automatic_update_error: self.stage_label.setText(self.result.automatic_update_error) self.action_button.setText("等待修复版本") self.action_button.setEnabled(False) return try: update_installer.metadata_from_update_info(self.result) except update_installer.UpdateInstallError as exc: self._show_failure(str(exc)) @Slot() def start_update(self): if self.thread is not None: return self.exit_requested = False self.action_button.setEnabled(False) self.exit_button.setText("取消并退出") self.stage_label.setText("正在准备新版") self.worker = self.worker_factory(self.result, self.install_root) self.thread = QThread(self) self.worker.moveToThread(self.thread) self.thread.started.connect(self.worker.run) self.worker.stage_changed.connect(self.stage_label.setText) self.worker.progress_changed.connect(self._update_progress) self.worker.succeeded.connect(self._on_prepared) self.worker.failed.connect(self._on_failed) self.worker.finished.connect(self.thread.quit) self.worker.finished.connect(self.worker.deleteLater) self.thread.finished.connect(self._on_thread_finished) self.thread.start() @Slot(int, int) def _update_progress(self, downloaded, total): percent = int(downloaded * 100 / total) if total else 0 self.progress_bar.setValue(max(0, min(100, percent))) self.bytes_label.setText("已下载 %s / %s" % (_format_bytes(downloaded), _format_bytes(total))) @Slot(object) def _on_prepared(self, staged): self.stage_label.setText("新版准备完成，正在重启") self.progress_bar.setValue(100) try: plan_path = updater_entry.create_plan( self.install_root, staged.staging_dir, staged.version, os.getpid(), staged.sha256, ) updater_path = self.install_root / "cmshopee-updater.exe" self.updater_launcher(updater_path, plan_path) self.update_started = True except Exception as exc: try: diagnostics.write_diagnostic_log( "启动独立更新器失败", level="ERROR", step="startup_update_launch", exc=exc, ) except Exception: pass self._show_failure("启动更新程序失败，请重试或联系管理员") @Slot(str) def _on_failed(self, message): if self.exit_requested: self.reject() return self._show_failure(message) @Slot() def _on_thread_finished(self): thread = self.thread self.thread = None self.worker = None if thread is not None: thread.deleteLater() if self.update_started: self.accept() def _show_failure(self, message): self.stage_label.setText("升级失败：%s" % message) self.action_button.setText("重试") self.action_button.setEnabled(True) self.exit_button.setText("退出程序") @Slot() def request_exit(self): if self.worker is not None: self.exit_requested = True self.action_button.setEnabled(False) self.exit_button.setEnabled(False) self.stage_label.setText("正在取消下载并清理临时文件") self.worker.cancel() return self.reject() def reject(self): if self.worker is not None: self.request_exit() return super().reject()