89 lines
4.1 KiB
Python
89 lines
4.1 KiB
Python
import os
|
|
import re
|
|
from typing import List, Tuple, Optional
|
|
from urllib.parse import urljoin
|
|
from config import settings
|
|
from utils.logger import session_logger
|
|
from services.system_config_manager import system_config_manager
|
|
|
|
class FirmwareScanner:
|
|
def __init__(self):
|
|
self.firmware_dir = os.path.join(settings.assets_dir, "firmware")
|
|
self.firmware_file_pattern = re.compile(r'^(\d+\.\d+\.\d+)\.bin$')
|
|
|
|
async def scan_firmware_files(self) -> None:
|
|
"""扫描固件目录并更新系统配置数据库"""
|
|
try:
|
|
if not os.path.exists(self.firmware_dir):
|
|
os.makedirs(self.firmware_dir, exist_ok=True)
|
|
session_logger.info("system", "firmware_scanner", f"创建固件目录: {self.firmware_dir}")
|
|
return
|
|
|
|
firmware_files = self._get_firmware_files()
|
|
if not firmware_files:
|
|
session_logger.info("system", "firmware_scanner", "未找到固件文件")
|
|
return
|
|
|
|
latest_version, latest_file = self._find_latest_version(firmware_files)
|
|
if not latest_version:
|
|
session_logger.info("system", "firmware_scanner", "无法确定最新版本")
|
|
return
|
|
|
|
relative_path = os.path.join("assets", "firmware", latest_file)
|
|
base_url = f"http://{settings.server_host}:{settings.server_port}/"
|
|
firmware_url = urljoin(base_url, relative_path)
|
|
|
|
current_version_config = await system_config_manager.get_config('latest_firmware_version')
|
|
current_url_config = await system_config_manager.get_config('update_firmware_url')
|
|
|
|
current_version = current_version_config.config_value if current_version_config else ""
|
|
current_url = current_url_config.config_value if current_url_config else ""
|
|
|
|
if current_version != latest_version:
|
|
await system_config_manager.update_config('latest_firmware_version', latest_version)
|
|
session_logger.info("system", "firmware_scanner", f"更新了最新固件版本: {latest_version}")
|
|
|
|
if current_url != firmware_url:
|
|
await system_config_manager.update_config('update_firmware_url', firmware_url)
|
|
session_logger.info("system", "firmware_scanner", f"更新了固件URL: {firmware_url}")
|
|
|
|
session_logger.info("system", "firmware_scanner", f"固件扫描完成，最新版本: {latest_version}")
|
|
|
|
except Exception as e:
|
|
session_logger.error("system", "firmware_scanner", f"扫描固件文件出错: {str(e)}")
|
|
|
|
def _get_firmware_files(self) -> List[str]:
|
|
"""获取固件目录中的所有固件文件"""
|
|
try:
|
|
all_files = os.listdir(self.firmware_dir)
|
|
firmware_files = [f for f in all_files if self.firmware_file_pattern.match(f)]
|
|
return firmware_files
|
|
except Exception as e:
|
|
session_logger.error("system", "firmware_scanner", f"获取固件文件列表出错: {str(e)}")
|
|
return []
|
|
|
|
def _find_latest_version(self, firmware_files: List[str]) -> Tuple[Optional[str], Optional[str]]:
|
|
"""查找最新版本的固件"""
|
|
try:
|
|
version_files = []
|
|
for file in firmware_files:
|
|
match = self.firmware_file_pattern.match(file)
|
|
if match:
|
|
version = match.group(1)
|
|
version_files.append((version, file))
|
|
|
|
if not version_files:
|
|
return None, None
|
|
|
|
latest = sorted(version_files, key=lambda x: self._version_to_tuple(x[0]), reverse=True)[0]
|
|
return latest
|
|
except Exception as e:
|
|
session_logger.error("system", "firmware_scanner", f"查找最新版本出错: {str(e)}")
|
|
return None, None
|
|
|
|
def _version_to_tuple(self, version: str) -> Tuple[int, ...]:
|
|
"""将版本号转换为数字元组，用于比较"""
|
|
return tuple(map(int, version.split('.')))
|
|
|
|
firmware_scanner = FirmwareScanner()
|