add banbanmini backend

talkingq-url/services/firmware_scanner.py

@@ -0,0 +1,88 @@
+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()