262 lines
11 KiB
Python
262 lines
11 KiB
Python
|
import os
|
|||
|
|
import json
|
|||
|
|
import logging
|
|||
|
|
import requests
|
|||
|
|
import shutil
|
|||
|
|
import tempfile
|
|||
|
|
import zipfile
|
|||
|
|
from pathlib import Path
|
|||
|
|
import sys
|
|||
|
|
import datetime
|
|||
|
|
from PySide6.QtWidgets import QMessageBox, QProgressDialog
|
|||
|
|
from PySide6.QtCore import Qt
|
|||
|
|
|
|||
|
|
# 版本检查服务器地址
|
|||
|
|
VERSION_CHECK_URL = "http://your-server.com/api/version_check"
|
|||
|
|
UPDATE_DOWNLOAD_URL = "http://your-server.com/api/download_update"
|
|||
|
|
|
|||
|
|
class VersionManager:
|
|||
|
|
"""版本管理工具类"""
|
|||
|
|
|
|||
|
|
VERSION = "1.0.0"
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def get_current_version():
|
|||
|
|
"""获取当前版本号"""
|
|||
|
|
return VersionManager.VERSION
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def check_for_updates(parent_widget=None):
|
|||
|
|
"""
|
|||
|
|
检查是否有更新可用
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
parent_widget: 父窗口，用于显示消息框
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
dict: 包含更新信息的字典，如果没有更新或检查失败则为None
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
current_version = VersionManager.get_current_version()
|
|||
|
|
logging.info(f"正在检查更新，当前版本: {current_version}")
|
|||
|
|
|
|||
|
|
# 从服务器获取最新版本信息
|
|||
|
|
response = requests.get(VERSION_CHECK_URL, params={"current_version": current_version}, timeout=5)
|
|||
|
|
response.raise_for_status() # 如果HTTP请求返回了不成功的状态码，将引发HTTPError异常
|
|||
|
|
|
|||
|
|
data = response.json()
|
|||
|
|
if not data.get("success"):
|
|||
|
|
logging.warning(f"版本检查失败: {data.get('message')}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
if data.get("update_available"):
|
|||
|
|
logging.info(f"发现新版本: {data.get('latest_version')}")
|
|||
|
|
return {
|
|||
|
|
"latest_version": data.get("latest_version"),
|
|||
|
|
"update_url": data.get("update_url"),
|
|||
|
|
"release_notes": data.get("release_notes"),
|
|||
|
|
"is_mandatory": data.get("is_mandatory", False)
|
|||
|
|
}
|
|||
|
|
else:
|
|||
|
|
logging.info("当前已是最新版本")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
except requests.RequestException as e:
|
|||
|
|
logging.error(f"检查更新时网络错误: {e}")
|
|||
|
|
if parent_widget:
|
|||
|
|
QMessageBox.warning(parent_widget, "更新检查失败", f"无法连接到更新服务器: {e}")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
logging.error(f"检查更新时发生错误: {e}")
|
|||
|
|
if parent_widget:
|
|||
|
|
QMessageBox.warning(parent_widget, "更新检查失败", f"检查更新时发生错误: {e}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def download_and_install_update(update_info, parent_widget=None):
|
|||
|
|
"""
|
|||
|
|
下载并安装更新
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
update_info: 包含更新信息的字典
|
|||
|
|
parent_widget: 父窗口，用于显示进度对话框
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
bool: 更新是否成功
|
|||
|
|
"""
|
|||
|
|
if not update_info or not update_info.get("update_url"):
|
|||
|
|
logging.error("无效的更新信息")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
update_url = update_info.get("update_url")
|
|||
|
|
version = update_info.get("latest_version")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# 创建进度对话框
|
|||
|
|
progress = None
|
|||
|
|
if parent_widget:
|
|||
|
|
progress = QProgressDialog("正在下载更新...", "取消", 0, 100, parent_widget)
|
|||
|
|
progress.setWindowTitle(f"下载更新 v{version}")
|
|||
|
|
progress.setWindowModality(Qt.WindowModal)
|
|||
|
|
progress.setAutoClose(True)
|
|||
|
|
progress.setMinimumDuration(0)
|
|||
|
|
progress.setValue(0)
|
|||
|
|
progress.show()
|
|||
|
|
|
|||
|
|
# 下载更新文件
|
|||
|
|
logging.info(f"开始下载更新: {update_url}")
|
|||
|
|
|
|||
|
|
# 使用临时目录
|
|||
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|||
|
|
# 下载文件
|
|||
|
|
update_file = os.path.join(temp_dir, "update.zip")
|
|||
|
|
|
|||
|
|
# 使用流式下载以显示进度
|
|||
|
|
with requests.get(update_url, stream=True) as r:
|
|||
|
|
r.raise_for_status()
|
|||
|
|
total_size = int(r.headers.get('content-length', 0))
|
|||
|
|
downloaded = 0
|
|||
|
|
|
|||
|
|
with open(update_file, 'wb') as f:
|
|||
|
|
for chunk in r.iter_content(chunk_size=8192):
|
|||
|
|
if progress and progress.wasCanceled():
|
|||
|
|
logging.info("用户取消了下载")
|
|||
|
|
return False
|
|||
|
|
if chunk:
|
|||
|
|
f.write(chunk)
|
|||
|
|
downloaded += len(chunk)
|
|||
|
|
if total_size and progress:
|
|||
|
|
percent = int((downloaded / total_size) * 100)
|
|||
|
|
progress.setValue(percent)
|
|||
|
|
|
|||
|
|
if progress:
|
|||
|
|
progress.setValue(100)
|
|||
|
|
progress.setLabelText("正在安装更新...")
|
|||
|
|
|
|||
|
|
# 解压更新文件
|
|||
|
|
logging.info("正在解压更新文件")
|
|||
|
|
extract_dir = os.path.join(temp_dir, "extracted")
|
|||
|
|
os.makedirs(extract_dir, exist_ok=True)
|
|||
|
|
|
|||
|
|
with zipfile.ZipFile(update_file, 'r') as zip_ref:
|
|||
|
|
zip_ref.extractall(extract_dir)
|
|||
|
|
|
|||
|
|
# 获取应用程序根目录
|
|||
|
|
app_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|||
|
|
|
|||
|
|
# 备份当前版本
|
|||
|
|
current_version = VersionManager.get_current_version()
|
|||
|
|
backup_dir = os.path.join(app_dir, f"backup_v{current_version}")
|
|||
|
|
logging.info(f"备份当前版本到: {backup_dir}")
|
|||
|
|
if os.path.exists(backup_dir):
|
|||
|
|
shutil.rmtree(backup_dir)
|
|||
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|||
|
|
|
|||
|
|
# 复制当前文件到备份目录
|
|||
|
|
for item in os.listdir(app_dir):
|
|||
|
|
if not item.startswith("backup_v"):
|
|||
|
|
s = os.path.join(app_dir, item)
|
|||
|
|
d = os.path.join(backup_dir, item)
|
|||
|
|
if os.path.isdir(s):
|
|||
|
|
shutil.copytree(s, d, dirs_exist_ok=True)
|
|||
|
|
else:
|
|||
|
|
shutil.copy2(s, d)
|
|||
|
|
|
|||
|
|
# 复制更新文件到应用目录
|
|||
|
|
logging.info("正在安装更新文件")
|
|||
|
|
for item in os.listdir(extract_dir):
|
|||
|
|
s = os.path.join(extract_dir, item)
|
|||
|
|
d = os.path.join(app_dir, item)
|
|||
|
|
if os.path.isdir(s):
|
|||
|
|
shutil.copytree(s, d, dirs_exist_ok=True)
|
|||
|
|
else:
|
|||
|
|
shutil.copy2(s, d)
|
|||
|
|
|
|||
|
|
# 更新版本文件
|
|||
|
|
version_file = os.path.join(app_dir, "version.json")
|
|||
|
|
with open(version_file, 'w') as f:
|
|||
|
|
json.dump({
|
|||
|
|
"version": version,
|
|||
|
|
"updated_at": str(datetime.datetime.now()),
|
|||
|
|
"release_notes": update_info.get("release_notes", "")
|
|||
|
|
}, f, indent=4)
|
|||
|
|
|
|||
|
|
logging.info(f"更新完成，新版本: {version}")
|
|||
|
|
|
|||
|
|
if parent_widget:
|
|||
|
|
QMessageBox.information(parent_widget, "更新完成",
|
|||
|
|
f"已成功更新到版本 {version}。\n应用程序需要重启才能应用更新。")
|
|||
|
|
|
|||
|
|
# 提示用户重启应用
|
|||
|
|
if parent_widget and QMessageBox.question(
|
|||
|
|
parent_widget, "重启应用",
|
|||
|
|
"需要重启应用程序以完成更新。是否现在重启？",
|
|||
|
|
QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes:
|
|||
|
|
|
|||
|
|
logging.info("用户选择重启应用")
|
|||
|
|
# 重启应用
|
|||
|
|
python = sys.executable
|
|||
|
|
os.execl(python, python, *sys.argv)
|
|||
|
|
|
|||
|
|
return True
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logging.error(f"下载或安装更新时发生错误: {e}", exc_info=True)
|
|||
|
|
if parent_widget:
|
|||
|
|
QMessageBox.critical(parent_widget, "更新失败", f"下载或安装更新时发生错误: {e}")
|
|||
|
|
return False
|
|||
|
|
finally:
|
|||
|
|
if progress:
|
|||
|
|
progress.close()
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def check_and_prompt_update(parent_widget=None):
|
|||
|
|
"""检查更新并提示用户"""
|
|||
|
|
# 这里只是一个示例，实际应该连接到服务器检查更新
|
|||
|
|
logging.info(f"检查更新，当前版本: {VersionManager.VERSION}")
|
|||
|
|
return False # 返回是否有更新
|
|||
|
|
|
|||
|
|
@staticmethod
|
|||
|
|
def check_and_prompt_update(parent_widget):
|
|||
|
|
"""
|
|||
|
|
检查更新并提示用户
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
parent_widget: 父窗口
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
bool: 是否有可用更新
|
|||
|
|
"""
|
|||
|
|
update_info = VersionManager.check_for_updates(parent_widget)
|
|||
|
|
if not update_info:
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
# 获取更新信息
|
|||
|
|
latest_version = update_info.get("latest_version", "未知")
|
|||
|
|
release_notes = update_info.get("release_notes", "无更新说明")
|
|||
|
|
is_mandatory = update_info.get("is_mandatory", False)
|
|||
|
|
|
|||
|
|
message = f"发现新版本: {latest_version}\n\n"
|
|||
|
|
message += "更新内容:\n"
|
|||
|
|
message += release_notes
|
|||
|
|
|
|||
|
|
if is_mandatory:
|
|||
|
|
message += "\n\n这是一个必须安装的更新。"
|
|||
|
|
result = QMessageBox.information(parent_widget, "发现必要更新", message,
|
|||
|
|
QMessageBox.Ok | QMessageBox.Cancel)
|
|||
|
|
if result == QMessageBox.Ok:
|
|||
|
|
return VersionManager.download_and_install_update(update_info, parent_widget)
|
|||
|
|
else:
|
|||
|
|
# 如果是必要更新但用户取消，则退出应用
|
|||
|
|
if parent_widget:
|
|||
|
|
QMessageBox.critical(parent_widget, "更新取消",
|
|||
|
|
"这是一个必要更新，应用程序将退出。")
|
|||
|
|
sys.exit(0)
|
|||
|
|
else:
|
|||
|
|
message += "\n\n是否现在更新？"
|
|||
|
|
result = QMessageBox.question(parent_widget, "发现新版本", message,
|
|||
|
|
QMessageBox.Yes | QMessageBox.No)
|
|||
|
|
if result == QMessageBox.Yes:
|
|||
|
|
return VersionManager.download_and_install_update(update_info, parent_widget)
|
|||
|
|
|
|||
|
|
return False
|