feat：启动时检查webui版本并尝试自动更新

src/config/config.py

@@ -56,7 +56,7 @@ MODEL_CONFIG_PATH: Path = (CONFIG_DIR / "model_config.toml").resolve().absolute(
 LEGACY_ENV_PATH: Path = (PROJECT_ROOT / ".env").resolve().absolute()
 A_MEMORIX_LEGACY_CONFIG_PATH: Path = (CONFIG_DIR / "a_memorix.toml").resolve().absolute()
 MMC_VERSION: str = "1.0.0-pre.13"
-CONFIG_VERSION: str = "8.10.9"
+CONFIG_VERSION: str = "8.10.10"
 MODEL_CONFIG_VERSION: str = "1.15.3"
 
 logger = get_logger("config")

src/config/official_configs.py

@@ -3150,6 +3150,15 @@ class WebUIConfig(ConfigBase):
     )
     """是否启用WebUI"""
 
+    auto_update_dashboard: bool = Field(
+        default=True,
+        json_schema_extra={
+            "x-widget": "switch",
+            "x-icon": "refresh-cw",
+        },
+    )
+    """启动时是否自动检查并更新 WebUI dashboard"""
+
     host: str = Field(
         default="127.0.0.1",
         json_schema_extra={

src/main.py

@@ -1,24 +1,26 @@
-from rich.traceback import install
 from typing import TYPE_CHECKING
 
+from rich.traceback import install
+
 import asyncio
 import time
 
 from src.A_memorix.host_service import a_memorix_host_service
-from src.learners.expression_auto_check_task import ExpressionAutoCheckTask
-from src.emoji_system.emoji_manager import emoji_manager
-from src.chat.message_receive.bot import chat_bot
 from src.chat.message_receive.chat_manager import chat_manager
+from src.chat.message_receive.bot import chat_bot
 from src.chat.utils.statistic import OnlineTimeRecordTask, StatisticOutputTask
 from src.common.i18n import t
 from src.common.logger import get_logger
 from src.common.message_server.server import Server, get_global_server
 from src.config.config import config_manager, global_config
+from src.emoji_system.emoji_manager import emoji_manager
+from src.learners.expression_auto_check_task import ExpressionAutoCheckTask
 from src.manager.async_task_manager import async_task_manager
 from src.maisaka.display.stage_status_board import disable_stage_status_board, enable_stage_status_board
 from src.plugin_runtime.integration import get_plugin_runtime_manager
 from src.prompt.prompt_manager import prompt_manager
 from src.services.memory_flow_service import memory_automation_service
+from src.webui.dashboard_update import auto_update_dashboard_if_needed
 
 # from src.api.main import start_api_server
 
@@ -45,9 +47,6 @@ class MainSystem:
         self.server: Server = get_global_server()
         self.webui_server: WebUIServer | None = None  # 独立的 WebUI 服务器
 
-        # 设置独立的 WebUI 服务器
-        self._setup_webui_server()
-
     def _setup_webui_server(self) -> None:
         """设置独立的 WebUI 服务器"""
         from src.config.config import global_config
@@ -70,11 +69,30 @@ class MainSystem:
             enable_stage_status_board()
         logger.info(t("startup.waking_up", nickname=global_config.bot.nickname))
 
+        await self._auto_update_webui_dashboard()
+
+        # 设置独立的 WebUI 服务器
+        self._setup_webui_server()
+
         # 其他初始化任务
         await asyncio.gather(self._init_components())
 
         logger.info(t("startup.initialization_completed_banner", nickname=global_config.bot.nickname))
 
+    async def _auto_update_webui_dashboard(self) -> None:
+        """启动时自动检查并更新 WebUI dashboard。"""
+        if not global_config.webui.enabled:
+            return
+        if not global_config.webui.auto_update_dashboard:
+            logger.info("WebUI dashboard 自动更新已关闭")
+            return
+
+        result = await auto_update_dashboard_if_needed()
+        if result.updated:
+            logger.info(result.message)
+        elif result.checked:
+            logger.info(result.message)
+
     async def _init_components(self) -> None:
         """初始化其他组件"""
         init_start_time = time.time()

src/webui/dashboard_update.py

@@ -0,0 +1,283 @@
+"""WebUI dashboard 版本检查与自动更新。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from importlib.metadata import PackageNotFoundError, version as get_package_version
+from pathlib import Path
+from typing import Any, Dict, Literal, Optional
+
+import asyncio
+import os
+import shutil
+import subprocess
+import sys
+import time
+
+import httpx
+
+from src.common.logger import get_logger
+
+logger = get_logger("webui_dashboard_update")
+
+DASHBOARD_PACKAGE_NAME = "maibot-dashboard"
+PYPI_JSON_URL = f"https://pypi.org/pypi/{DASHBOARD_PACKAGE_NAME}/json"
+PYPI_PROJECT_URL = f"https://pypi.org/project/{DASHBOARD_PACKAGE_NAME}/"
+PYPI_CACHE_TTL_SECONDS = 60 * 60 * 6
+
+PackageRunner = Literal["uv", "pip", "unknown"]
+_pypi_version_cache: Dict[str, Any] = {"checked_at": 0.0, "latest_version": None}
+
+
+@dataclass(frozen=True)
+class DashboardVersionInfo:
+    """WebUI dashboard 版本检查结果。"""
+
+    current_version: str
+    latest_version: Optional[str]
+    has_update: bool
+    package_name: str = DASHBOARD_PACKAGE_NAME
+    pypi_url: str = PYPI_PROJECT_URL
+
+
+@dataclass(frozen=True)
+class DashboardUpdateResult:
+    """WebUI dashboard 自动更新结果。"""
+
+    checked: bool
+    updated: bool
+    current_version: str
+    latest_version: Optional[str]
+    runner: PackageRunner
+    message: str
+
+
+def get_installed_dashboard_version() -> str:
+    try:
+        return get_package_version(DASHBOARD_PACKAGE_NAME)
+    except PackageNotFoundError:
+        return "unknown"
+
+
+def normalize_version(version: str) -> tuple[int, ...]:
+    clean_version = version.strip().lower().removeprefix("v")
+    numeric_part = clean_version.split("-", 1)[0].split("+", 1)[0]
+    parts = []
+    for item in numeric_part.split("."):
+        number = ""
+        for char in item:
+            if not char.isdigit():
+                break
+            number += char
+        parts.append(int(number) if number else 0)
+    return tuple(parts)
+
+
+def is_newer_version(latest_version: Optional[str], current_version: str) -> bool:
+    if not latest_version or not current_version or current_version == "unknown":
+        return False
+
+    latest_parts = normalize_version(latest_version)
+    current_parts = normalize_version(current_version)
+    width = max(len(latest_parts), len(current_parts))
+    return latest_parts + (0,) * (width - len(latest_parts)) > current_parts + (0,) * (width - len(current_parts))
+
+
+async def get_latest_dashboard_version_from_pypi() -> Optional[str]:
+    now = time.time()
+    cached_version = _pypi_version_cache.get("latest_version")
+    checked_at = float(_pypi_version_cache.get("checked_at", 0.0))
+    if cached_version and now - checked_at < PYPI_CACHE_TTL_SECONDS:
+        return str(cached_version)
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            response = await client.get(PYPI_JSON_URL)
+            response.raise_for_status()
+            payload = response.json()
+    except Exception as e:
+        logger.debug(f"检查 WebUI PyPI 版本失败: {e}")
+        return str(cached_version) if cached_version else None
+
+    latest_version = payload.get("info", {}).get("version")
+    if isinstance(latest_version, str) and latest_version.strip():
+        _pypi_version_cache["checked_at"] = now
+        _pypi_version_cache["latest_version"] = latest_version.strip()
+        return latest_version.strip()
+
+    return str(cached_version) if cached_version else None
+
+
+async def get_dashboard_version_info(current_version: Optional[str] = None) -> DashboardVersionInfo:
+    resolved_current_version = current_version or get_installed_dashboard_version()
+    latest_version = await get_latest_dashboard_version_from_pypi()
+
+    return DashboardVersionInfo(
+        current_version=resolved_current_version,
+        latest_version=latest_version,
+        has_update=is_newer_version(latest_version, resolved_current_version),
+    )
+
+
+def _get_parent_command_line() -> str:
+    parent_pid = _get_parent_pid(os.getpid())
+    if parent_pid is None:
+        return ""
+    return _get_process_command_line(parent_pid)
+
+
+def _get_parent_pid(pid: int) -> Optional[int]:
+    if os.name == "nt":
+        return _get_windows_parent_pid(pid)
+    stat_path = Path(f"/proc/{pid}/stat")
+    try:
+        stat_text = stat_path.read_text(encoding="utf-8")
+    except OSError:
+        return None
+    parts = stat_text.split()
+    if len(parts) >= 4 and parts[3].isdigit():
+        return int(parts[3])
+    return None
+
+
+def _get_process_command_line(pid: int) -> str:
+    if os.name == "nt":
+        return _get_windows_process_command_line(pid)
+    cmdline_path = Path(f"/proc/{pid}/cmdline")
+    try:
+        return cmdline_path.read_text(encoding="utf-8").replace("\x00", " ").strip()
+    except OSError:
+        return ""
+
+
+def _get_windows_parent_pid(pid: int) -> Optional[int]:
+    output = _run_wmic_query(pid, "ParentProcessId")
+    parent_pid = output.get("ParentProcessId")
+    if parent_pid and parent_pid.isdigit():
+        return int(parent_pid)
+    return None
+
+
+def _get_windows_process_command_line(pid: int) -> str:
+    output = _run_wmic_query(pid, "CommandLine")
+    return output.get("CommandLine", "")
+
+
+def _run_wmic_query(pid: int, field: str) -> Dict[str, str]:
+    wmic_path = shutil.which("wmic")
+    if not wmic_path:
+        return {}
+
+    try:
+        result = subprocess.run(
+            [wmic_path, "process", "where", f"ProcessId={pid}", "get", field, "/format:list"],
+            check=False,
+            capture_output=True,
+            encoding="utf-8",
+            errors="ignore",
+            timeout=3,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return {}
+
+    values = {}
+    for line in result.stdout.splitlines():
+        if "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        values[key.strip()] = value.strip()
+    return values
+
+
+def detect_package_runner() -> PackageRunner:
+    """检测当前进程更像是由 uv 还是 pip/普通 python 启动。"""
+
+    uv_markers = ["UV", "UV_PROJECT_ENVIRONMENT", "UV_RUN_RECURSION_DEPTH"]
+    if any(os.getenv(marker) for marker in uv_markers):
+        return "uv"
+
+    parent_command = _get_parent_command_line().lower()
+    if parent_command:
+        executable = parent_command.split(maxsplit=1)[0]
+        if executable.endswith("uv.exe") or executable.endswith("/uv") or executable == "uv":
+            return "uv"
+        if " pip " in f" {parent_command} " or executable.endswith("pip.exe") or executable.endswith("/pip"):
+            return "pip"
+
+    return "unknown"
+
+
+def _build_update_command(runner: PackageRunner) -> list[str]:
+    if runner == "uv" and shutil.which("uv"):
+        return ["uv", "pip", "install", "--python", sys.executable, "--upgrade", DASHBOARD_PACKAGE_NAME]
+    return [sys.executable, "-m", "pip", "install", "--upgrade", DASHBOARD_PACKAGE_NAME]
+
+
+async def auto_update_dashboard_if_needed() -> DashboardUpdateResult:
+    version_info = await get_dashboard_version_info()
+    runner = detect_package_runner()
+    if not version_info.latest_version:
+        return DashboardUpdateResult(
+            checked=True,
+            updated=False,
+            current_version=version_info.current_version,
+            latest_version=None,
+            runner=runner,
+            message="无法获取 WebUI 最新版本，跳过自动更新",
+        )
+    if not version_info.has_update:
+        return DashboardUpdateResult(
+            checked=True,
+            updated=False,
+            current_version=version_info.current_version,
+            latest_version=version_info.latest_version,
+            runner=runner,
+            message="WebUI 已是最新版本",
+        )
+
+    update_runner = runner if runner != "unknown" else "pip"
+    command = _build_update_command(update_runner)
+    logger.info(
+        f"检测到 WebUI 新版本: {version_info.current_version} -> {version_info.latest_version}，"
+        f"使用 {update_runner} 自动更新"
+    )
+
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+    except OSError as e:
+        logger.warning(f"WebUI 自动更新启动失败: {e}")
+        return DashboardUpdateResult(
+            checked=True,
+            updated=False,
+            current_version=version_info.current_version,
+            latest_version=version_info.latest_version,
+            runner=update_runner,
+            message=f"自动更新启动失败: {e}",
+        )
+
+    if process.returncode != 0:
+        error_text = stderr.decode(errors="ignore").strip() or stdout.decode(errors="ignore").strip()
+        logger.warning(f"WebUI 自动更新失败: {error_text}")
+        return DashboardUpdateResult(
+            checked=True,
+            updated=False,
+            current_version=version_info.current_version,
+            latest_version=version_info.latest_version,
+            runner=update_runner,
+            message=f"自动更新失败: {error_text}",
+        )
+
+    logger.info(f"WebUI 自动更新完成: {version_info.current_version} -> {version_info.latest_version}")
+    return DashboardUpdateResult(
+        checked=True,
+        updated=True,
+        current_version=version_info.current_version,
+        latest_version=version_info.latest_version,
+        runner=update_runner,
+        message="WebUI 自动更新完成，重启后生效",
+    )

src/webui/routers/system.py

@@ -5,29 +5,28 @@
 """
 
 from datetime import datetime
-from importlib.metadata import PackageNotFoundError, version as get_package_version
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
-import httpx
 import os
 import time
 
 from src.common.logger import get_logger
 from src.config.config import MMC_VERSION
+from src.webui.dashboard_update import (
+    DASHBOARD_PACKAGE_NAME,
+    PYPI_PROJECT_URL,
+    detect_package_runner,
+    get_dashboard_version_info,
+)
 from src.webui.dependencies import require_auth
 
 router = APIRouter(prefix="/system", tags=["system"], dependencies=[Depends(require_auth)])
 logger = get_logger("webui_system")
 
-# 记录启动时间
 _start_time = time.time()
-_DASHBOARD_PACKAGE_NAME = "maibot-dashboard"
-_PYPI_JSON_URL = f"https://pypi.org/pypi/{_DASHBOARD_PACKAGE_NAME}/json"
-_PYPI_CACHE_TTL_SECONDS = 60 * 60 * 6
-_pypi_version_cache: Dict[str, Any] = {"checked_at": 0.0, "latest_version": None}
 
 
 class RestartResponse(BaseModel):
@@ -52,64 +51,9 @@ class DashboardVersionResponse(BaseModel):
     current_version: str
     latest_version: Optional[str] = None
     has_update: bool = False
-    package_name: str = _DASHBOARD_PACKAGE_NAME
-    pypi_url: str = f"https://pypi.org/project/{_DASHBOARD_PACKAGE_NAME}/"
-
-
-def _get_installed_dashboard_version() -> str:
-    try:
-        return get_package_version(_DASHBOARD_PACKAGE_NAME)
-    except PackageNotFoundError:
-        return "unknown"
-
-
-def _normalize_version(version: str) -> tuple[int, ...]:
-    clean_version = version.strip().lower().removeprefix("v")
-    numeric_part = clean_version.split("-", 1)[0].split("+", 1)[0]
-    parts = []
-    for item in numeric_part.split("."):
-        number = ""
-        for char in item:
-            if not char.isdigit():
-                break
-            number += char
-        parts.append(int(number) if number else 0)
-    return tuple(parts)
-
-
-def _is_newer_version(latest_version: Optional[str], current_version: str) -> bool:
-    if not latest_version or not current_version or current_version == "unknown":
-        return False
-
-    latest_parts = _normalize_version(latest_version)
-    current_parts = _normalize_version(current_version)
-    width = max(len(latest_parts), len(current_parts))
-    return latest_parts + (0,) * (width - len(latest_parts)) > current_parts + (0,) * (width - len(current_parts))
-
-
-async def _get_latest_dashboard_version_from_pypi() -> Optional[str]:
-    now = time.time()
-    cached_version = _pypi_version_cache.get("latest_version")
-    checked_at = float(_pypi_version_cache.get("checked_at", 0.0))
-    if cached_version and now - checked_at < _PYPI_CACHE_TTL_SECONDS:
-        return str(cached_version)
-
-    try:
-        async with httpx.AsyncClient(timeout=5.0) as client:
-            response = await client.get(_PYPI_JSON_URL)
-            response.raise_for_status()
-            payload = response.json()
-    except Exception as e:
-        logger.debug(f"检查 WebUI PyPI 版本失败: {e}")
-        return str(cached_version) if cached_version else None
-
-    latest_version = payload.get("info", {}).get("version")
-    if isinstance(latest_version, str) and latest_version.strip():
-        _pypi_version_cache["checked_at"] = now
-        _pypi_version_cache["latest_version"] = latest_version.strip()
-        return latest_version.strip()
-
-    return str(cached_version) if cached_version else None
+    runner: str = "unknown"
+    package_name: str = DASHBOARD_PACKAGE_NAME
+    pypi_url: str = PYPI_PROJECT_URL
 
 
 @router.post("/restart", response_model=RestartResponse)
@@ -166,13 +110,13 @@ async def get_maibot_status():
 @router.get("/dashboard-version", response_model=DashboardVersionResponse)
 async def get_dashboard_version(current_version: Optional[str] = None):
     """获取 WebUI 当前版本和 PyPI 最新版本。"""
-    resolved_current_version = current_version or _get_installed_dashboard_version()
-    latest_version = await _get_latest_dashboard_version_from_pypi()
+    version_info = await get_dashboard_version_info(current_version)
 
     return DashboardVersionResponse(
-        current_version=resolved_current_version,
-        latest_version=latest_version,
-        has_update=_is_newer_version(latest_version, resolved_current_version),
+        current_version=version_info.current_version,
+        latest_version=version_info.latest_version,
+        has_update=version_info.has_update,
+        runner=detect_package_runner(),
     )