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 |