170 lines
5.6 KiB
Python
170 lines
5.6 KiB
Python
"""
|
||
软件自动更新模块
|
||
- 检查版本：GET https://api.yunzer.cn/api/softwareupgrade/check?code=niumasortware
|
||
- 下载新版 exe，替换当前程序后重启
|
||
"""
|
||
import os
|
||
import sys
|
||
import subprocess
|
||
import tempfile
|
||
from PyQt6.QtCore import QThread, pyqtSignal, QObject
|
||
from PyQt6.QtWidgets import QProgressDialog, QMessageBox, QApplication
|
||
import urllib.request
|
||
|
||
|
||
UPDATE_CHECK_URL = "https://api.yunzer.cn/api/softwareupgrade/check?code=niumasortware"
|
||
|
||
|
||
def _current_version() -> str:
|
||
"""从 main 模块取当前版本号，避免循环导入。"""
|
||
try:
|
||
import importlib
|
||
m = importlib.import_module("__main__")
|
||
return getattr(m, "__VERSION__", "0.0.0")
|
||
except Exception:
|
||
return "0.0.0"
|
||
|
||
|
||
def _is_newer(latest: str, current: str) -> bool:
|
||
try:
|
||
def _parse(v):
|
||
return tuple(int(x) for x in v.strip().lstrip("v").split("."))
|
||
return _parse(latest) > _parse(current)
|
||
except Exception:
|
||
return latest != current
|
||
|
||
|
||
class _CheckWorker(QThread):
|
||
result = pyqtSignal(dict) # 成功：返回 data 字段
|
||
error = pyqtSignal(str)
|
||
|
||
def run(self):
|
||
try:
|
||
import json
|
||
import urllib.request
|
||
with urllib.request.urlopen(UPDATE_CHECK_URL, timeout=10) as resp:
|
||
body = json.loads(resp.read().decode())
|
||
if body.get("code") == 200:
|
||
self.result.emit(body["data"])
|
||
else:
|
||
self.error.emit(body.get("msg", "接口返回异常"))
|
||
except Exception as e:
|
||
self.error.emit(str(e))
|
||
|
||
|
||
class _DownloadWorker(QThread):
|
||
progress = pyqtSignal(int) # 0-100
|
||
finished = pyqtSignal(str) # 下载完成，返回临时文件路径
|
||
error = pyqtSignal(str)
|
||
|
||
def __init__(self, url: str):
|
||
super().__init__()
|
||
self._url = url
|
||
|
||
def run(self):
|
||
try:
|
||
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".exe")
|
||
tmp.close()
|
||
dst = tmp.name
|
||
|
||
def _reporthook(count, block_size, total_size):
|
||
if total_size > 0:
|
||
pct = min(100, int(count * block_size * 100 / total_size))
|
||
self.progress.emit(pct)
|
||
|
||
urllib.request.urlretrieve(self._url, dst, _reporthook)
|
||
self.progress.emit(100)
|
||
self.finished.emit(dst)
|
||
except Exception as e:
|
||
self.error.emit(str(e))
|
||
|
||
|
||
def _replace_and_restart(new_exe: str):
|
||
"""
|
||
写一个批处理脚本：等待当前进程退出 → 覆盖 exe → 启动新版。
|
||
仅在打包为 exe 时执行真正替换；开发环境直接启动下载文件。
|
||
"""
|
||
current_exe = sys.executable if getattr(sys, "frozen", False) else None
|
||
|
||
if current_exe:
|
||
bat = tempfile.NamedTemporaryFile(delete=False, suffix=".bat", mode="w", encoding="gbk")
|
||
bat.write(f"""@echo off
|
||
ping 127.0.0.1 -n 3 >nul
|
||
move /y "{new_exe}" "{current_exe}"
|
||
start "" "{current_exe}"
|
||
del "%~f0"
|
||
""")
|
||
bat.close()
|
||
subprocess.Popen(["cmd", "/c", bat.name], creationflags=subprocess.CREATE_NO_WINDOW)
|
||
else:
|
||
# 开发环境：直接运行下载的 exe
|
||
subprocess.Popen([new_exe])
|
||
|
||
QApplication.quit()
|
||
|
||
|
||
class Updater(QObject):
|
||
"""对外接口：调用 check() 即可。"""
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self._parent_widget = parent
|
||
|
||
def check(self, silent_if_latest: bool = False):
|
||
self._silent = silent_if_latest
|
||
self._worker = _CheckWorker()
|
||
self._worker.result.connect(self._on_check_result)
|
||
self._worker.error.connect(self._on_check_error)
|
||
self._worker.start()
|
||
|
||
def _on_check_result(self, data: dict):
|
||
current = _current_version()
|
||
latest = data.get("latestVersion", "")
|
||
url = data.get("downloadUrl", "")
|
||
|
||
if not _is_newer(latest, current):
|
||
if not self._silent:
|
||
QMessageBox.information(self._parent_widget, "检查更新", f"当前已是最新版本 v{current}")
|
||
return
|
||
|
||
notes = data.get("releaseNotes", "") or ""
|
||
msg = f"发现新版本 v{latest}（当前 v{current}）"
|
||
if notes:
|
||
msg += f"\n\n更新内容：\n{notes}"
|
||
msg += "\n\n是否立即下载更新？"
|
||
|
||
ret = QMessageBox.question(
|
||
self._parent_widget, "发现新版本", msg,
|
||
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
|
||
QMessageBox.StandardButton.Yes,
|
||
)
|
||
if ret != QMessageBox.StandardButton.Yes:
|
||
return
|
||
|
||
self._download(url)
|
||
|
||
def _on_check_error(self, err: str):
|
||
if not self._silent:
|
||
QMessageBox.warning(self._parent_widget, "检查更新失败", f"无法连接更新服务器：\n{err}")
|
||
|
||
def _download(self, url: str):
|
||
self._progress = QProgressDialog("正在下载更新...", "取消", 0, 100, self._parent_widget)
|
||
self._progress.setWindowTitle("下载更新")
|
||
self._progress.setMinimumDuration(0)
|
||
self._progress.setValue(0)
|
||
|
||
self._dl_worker = _DownloadWorker(url)
|
||
self._dl_worker.progress.connect(self._progress.setValue)
|
||
self._dl_worker.finished.connect(self._on_download_done)
|
||
self._dl_worker.error.connect(self._on_download_error)
|
||
self._progress.canceled.connect(self._dl_worker.terminate)
|
||
self._dl_worker.start()
|
||
|
||
def _on_download_done(self, path: str):
|
||
self._progress.close()
|
||
_replace_and_restart(path)
|
||
|
||
def _on_download_error(self, err: str):
|
||
self._progress.close()
|
||
QMessageBox.critical(self._parent_widget, "下载失败", f"下载更新失败：\n{err}")
|