[patch] 优化版本管理显示 与 更新提示(修改表名信息)

This commit is contained in:
2025-10-10 11:53:56 +08:00
parent bb5ef9ccbf
commit c819bdaa1c
5 changed files with 289 additions and 27 deletions

216
version_checker.py Normal file
View File

@@ -0,0 +1,216 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
版本检查器
定期检查后端是否有新版本并通知GUI更新
"""
import threading
import time
import logging
from typing import Callable, Optional
logger = logging.getLogger(__name__)
class VersionChecker:
"""
版本检查器
定期向后端发送版本检查请求,如果有新版本则触发更新回调
"""
def __init__(self, backend_client, update_callback: Optional[Callable] = None):
"""
初始化版本检查器
Args:
backend_client: 后端WebSocket客户端实例
update_callback: 发现新版本时的回调函数 callback(latest_version, download_url)
"""
self.backend_client = backend_client
self.update_callback = update_callback
self.is_running = False
self.check_interval = 600 # 10分钟检查一次
self.thread = None
# 从config获取当前版本
try:
from config import APP_VERSION
self.current_version = APP_VERSION
logger.info(f"✅ 版本检查器初始化成功,当前版本: v{self.current_version}")
except ImportError:
self.current_version = "1.0.0"
logger.warning("⚠️ 无法从config读取APP_VERSION使用默认版本: 1.0.0")
def start(self):
"""启动版本检查线程"""
if self.is_running:
logger.warning("版本检查器已在运行中")
return
self.is_running = True
self.thread = threading.Thread(target=self._check_loop, daemon=True)
self.thread.start()
logger.info(f"✅ 版本检查器已启动,每{self.check_interval}秒检查一次更新")
def stop(self):
"""停止版本检查"""
self.is_running = False
if self.thread:
self.thread.join(timeout=2)
logger.info("版本检查器已停止")
def _check_loop(self):
"""版本检查循环"""
# 等待5秒后首次检查避免启动时立即检查
time.sleep(5)
while self.is_running:
try:
self._perform_version_check()
except Exception as e:
logger.error(f"版本检查失败: {e}")
# 等待下次检查
time.sleep(self.check_interval)
def _perform_version_check(self):
"""执行版本检查"""
try:
if not self.backend_client or not self.backend_client.is_connected:
logger.debug("后端未连接,跳过版本检查")
return
# 向后端发送版本检查请求
message = {
'type': 'version_check',
'current_version': self.current_version,
'client_type': 'gui'
}
logger.debug(f"发送版本检查请求: v{self.current_version}")
self.backend_client.send_message(message)
except Exception as e:
logger.error(f"发送版本检查请求失败: {e}")
def handle_version_response(self, response: dict):
"""
处理后端返回的版本检查响应
Args:
response: 后端响应消息
{
"type": "version_response",
"has_update": true,
"latest_version": "1.5.0",
"current_version": "1.4.7",
"download_url": "https://example.com/download/v1.5.0",
"update_content": "修复了若干Bug"
}
"""
try:
has_update = response.get('has_update', None) # 改为 None便于后续判断
latest_version = response.get('latest_version', '')
current_version = response.get('current_version', self.current_version)
download_url = response.get('download_url', '')
update_content = response.get('update_content', '')
logger.info(f"📡 收到版本检查响应: 当前v{current_version}, 最新v{latest_version}")
# 如果后端没有返回 has_update 字段,自动通过版本比较判断
if has_update is None and latest_version:
# 自动比较版本号
compare_result = self.compare_versions(latest_version, current_version)
has_update = compare_result > 0 # latest > current 表示有更新
logger.info(f"🔍 后端未返回has_update字段通过版本比较判断: {has_update}")
if has_update and latest_version:
logger.info(f"🔔 发现新版本: v{latest_version}")
if update_content:
logger.info(f"📝 更新内容: {update_content[:100]}")
# 触发更新回调
if self.update_callback:
try:
self.update_callback(latest_version, download_url)
except Exception as e:
logger.error(f"执行更新回调失败: {e}")
else:
logger.debug(f"✅ 当前已是最新版本: v{current_version}")
except Exception as e:
logger.error(f"处理版本响应失败: {e}")
def compare_versions(self, version1: str, version2: str) -> int:
"""
比较两个版本号
Args:
version1: 版本1 (如 "1.4.7")
version2: 版本2 (如 "1.5.0")
Returns:
1: version1 > version2
0: version1 == version2
-1: version1 < version2
"""
try:
v1_parts = [int(x) for x in version1.split('.')]
v2_parts = [int(x) for x in version2.split('.')]
# 补齐长度
max_len = max(len(v1_parts), len(v2_parts))
v1_parts.extend([0] * (max_len - len(v1_parts)))
v2_parts.extend([0] * (max_len - len(v2_parts)))
# 逐位比较
for v1, v2 in zip(v1_parts, v2_parts):
if v1 > v2:
return 1
elif v1 < v2:
return -1
return 0
except Exception as e:
logger.error(f"版本比较失败: {e}")
return 0
# 使用示例
if __name__ == '__main__':
# 配置日志
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s [%(levelname)s] %(message)s'
)
# 模拟使用
class MockBackendClient:
is_connected = True
def send_message(self, msg):
print(f"发送消息: {msg}")
def on_update_available(version, url):
print(f"🔔 发现新版本: {version}, 下载地址: {url}")
client = MockBackendClient()
checker = VersionChecker(client, on_update_available)
checker.start()
# 模拟接收后端响应
time.sleep(2)
response = {
"type": "version_response",
"has_update": True,
"latest_version": "1.5.0",
"current_version": "1.4.7",
"download_url": "https://example.com/download",
"update_content": "修复了若干Bug"
}
checker.handle_version_response(response)
time.sleep(2)
checker.stop()