fix: prefer bundled dashboard over stale data dist (#8172)

* fix: prefer bundled dashboard over stale dist

* fix: harden dashboard dist version checks

Author: he-yufeng

astrbot/core/utils/io.py

@@ -2,6 +2,7 @@
 import inspect
 import logging
 import os
+import re
 import shutil
 import socket
 import ssl
@@ -16,6 +17,7 @@
 from PIL import Image
 
 from .astrbot_path import get_astrbot_data_path, get_astrbot_path, get_astrbot_temp_path
+from .version_comparator import VersionComparator
 
 logger = logging.getLogger("astrbot")
 
@@ -325,20 +327,67 @@ def get_local_ip_addresses():
     return network_ips
 
 
+def _read_dashboard_dist_version(dist_dir: str | Path) -> str | None:
+    version_file = Path(dist_dir) / "assets" / "version"
+    if version_file.exists():
+        return version_file.read_text(encoding="utf-8").strip()
+    return None
+
+
+def get_bundled_dashboard_dist_path() -> Path:
+    return Path(get_astrbot_path()) / "astrbot" / "dashboard" / "dist"
+
+
+def _normalize_dashboard_version(version: str) -> str:
+    version = version.strip()
+    if version[:1].lower() == "v":
+        version = version[1:]
+    if not re.match(
+        r"^[0-9]+(?:\.[0-9]+)*"
+        r"(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?"
+        r"(?:\+.+)?$",
+        version,
+    ):
+        raise ValueError(f"invalid dashboard version: {version!r}")
+    return version
+
+
+def should_use_bundled_dashboard_dist(
+    user_dist: str | Path, current_version: str
+) -> bool:
+    user_version = _read_dashboard_dist_version(user_dist)
+    bundled_dist = get_bundled_dashboard_dist_path()
+    if user_version is None or not bundled_dist.exists():
+        return False
+    try:
+        return (
+            VersionComparator.compare_version(
+                _normalize_dashboard_version(current_version),
+                _normalize_dashboard_version(user_version),
+            )
+            > 0
+        )
+    except (TypeError, ValueError):
+        return False
+
+
 async def get_dashboard_version():
     # First check user data directory (manually updated / downloaded dashboard).
     dist_dir = os.path.join(get_astrbot_data_path(), "dist")
-    if not os.path.exists(dist_dir):
-        # Fall back to the dist bundled inside the installed wheel.
-        _bundled = Path(get_astrbot_path()) / "astrbot" / "dashboard" / "dist"
-        if _bundled.exists():
-            dist_dir = str(_bundled)
     if os.path.exists(dist_dir):
-        version_file = os.path.join(dist_dir, "assets", "version")
-        if os.path.exists(version_file):
-            with open(version_file, encoding="utf-8") as f:
-                v = f.read().strip()
-                return v
+        from astrbot.core.config.default import VERSION
+
+        if should_use_bundled_dashboard_dist(dist_dir, VERSION):
+            bundled_version = _read_dashboard_dist_version(
+                get_bundled_dashboard_dist_path()
+            )
+            if bundled_version is not None:
+                return bundled_version
+        return _read_dashboard_dist_version(dist_dir)
+
+    bundled = get_bundled_dashboard_dist_path()
+    if bundled.exists():
+        return _read_dashboard_dist_version(bundled)
     return None
 
 

astrbot/dashboard/server.py

@@ -23,7 +23,11 @@
 from astrbot.core.db import BaseDatabase
 from astrbot.core.utils.astrbot_path import get_astrbot_data_path
 from astrbot.core.utils.datetime_utils import to_utc_isoformat
-from astrbot.core.utils.io import get_local_ip_addresses
+from astrbot.core.utils.io import (
+    get_bundled_dashboard_dist_path,
+    get_local_ip_addresses,
+    should_use_bundled_dashboard_dist,
+)
 
 from .plugin_page_auth import PluginPageAuth
 from .routes import *
@@ -37,9 +41,6 @@
 from .routes.subagent import SubAgentRoute
 from .routes.t2i import T2iRoute
 
-# Static assets shipped inside the wheel (built during `hatch build`).
-_BUNDLED_DIST = Path(__file__).parent / "dist"
-
 
 class _AddrWithPort(Protocol):
     port: int
@@ -118,10 +119,14 @@ def __init__(
             self.data_path = os.path.abspath(webui_dir)
         else:
             user_dist = os.path.join(get_astrbot_data_path(), "dist")
-            if os.path.exists(user_dist):
+            bundled_dist = get_bundled_dashboard_dist_path()
+            if os.path.exists(user_dist) and not should_use_bundled_dashboard_dist(
+                user_dist,
+                VERSION,
+            ):
                 self.data_path = os.path.abspath(user_dist)
-            elif _BUNDLED_DIST.exists():
-                self.data_path = str(_BUNDLED_DIST)
+            elif bundled_dist.exists():
+                self.data_path = str(bundled_dist)
                 logger.info("Using bundled dashboard dist: %s", self.data_path)
             else:
                 # Fall back to expected user path (will fail gracefully later)

main.py

@@ -23,7 +23,9 @@
 )
 from astrbot.core.utils.io import (  # noqa: E402
     download_dashboard,
+    get_bundled_dashboard_dist_path,
     get_dashboard_version,
+    should_use_bundled_dashboard_dist,
 )
 
 # 将父目录添加到 sys.path
@@ -77,6 +79,13 @@ async def check_dashboard_files(webui_dir: str | None = None):
     data_dist_path = os.path.join(get_astrbot_data_path(), "dist")
     if os.path.exists(data_dist_path):
         v = await get_dashboard_version()
+        if should_use_bundled_dashboard_dist(data_dist_path, VERSION):
+            bundled_dist = get_bundled_dashboard_dist_path()
+            logger.info(
+                "Using bundled WebUI because data/dist is older than core version v%s.",
+                VERSION,
+            )
+            return str(bundled_dist)
         if v is not None:
             # 存在文件
             if v == f"v{VERSION}":

tests/test_dashboard.py

@@ -216,6 +216,36 @@ def _resolve_dashboard_password(core_lifecycle_td: AstrBotCoreLifecycle) -> str:
     return password
 
 
+def test_dashboard_uses_bundled_dist_when_data_dist_is_stale(
+    core_lifecycle_td: AstrBotCoreLifecycle,
+    monkeypatch,
+    tmp_path,
+):
+    data_dir = tmp_path / "data"
+    user_dist = data_dir / "dist"
+    bundled_dist = tmp_path / "bundled-dist"
+    user_dist.mkdir(parents=True)
+    bundled_dist.mkdir()
+
+    monkeypatch.setattr(
+        "astrbot.dashboard.server.get_astrbot_data_path",
+        lambda: str(data_dir),
+    )
+    monkeypatch.setattr(
+        "astrbot.dashboard.server.get_bundled_dashboard_dist_path",
+        lambda: bundled_dist,
+    )
+    monkeypatch.setattr(
+        "astrbot.dashboard.server.should_use_bundled_dashboard_dist",
+        lambda *_args, **_kwargs: True,
+    )
+
+    shutdown_event = asyncio.Event()
+    server = AstrBotDashboard(core_lifecycle_td, core_lifecycle_td.db, shutdown_event)
+
+    assert server.data_path == str(bundled_dist)
+
+
 async def _set_dashboard_password_change_required(
     core_lifecycle_td: AstrBotCoreLifecycle,
     required: bool,

tests/test_main.py

@@ -1,5 +1,6 @@
 import os
 import sys
+from pathlib import Path
 
 # 将项目根目录添加到 sys.path
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
@@ -8,6 +9,7 @@
 
 import pytest
 
+from astrbot.core.utils.io import should_use_bundled_dashboard_dist
 from main import check_dashboard_files, check_env
 
 
@@ -175,8 +177,9 @@ async def test_check_dashboard_files_exists_but_version_mismatch(monkeypatch):
     """Tests that a warning is logged when dashboard version mismatches."""
     monkeypatch.setattr(os.path, "exists", lambda x: True)
 
-    with mock.patch("main.get_dashboard_version") as mock_get_version:
-        mock_get_version.return_value = "v0.0.1"  # A different version
+    with mock.patch(
+        "main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
+    ):
 
         with mock.patch("main.logger.warning") as mock_logger_warning:
             await check_dashboard_files()
@@ -185,6 +188,64 @@ async def test_check_dashboard_files_exists_but_version_mismatch(monkeypatch):
             assert "WebUI version mismatch" in call_args[0]
 
 
+def test_should_use_bundled_dashboard_dist_when_data_dist_is_stale(tmp_path):
+    user_dist = tmp_path / "user-dist"
+    bundled_dist = tmp_path / "bundled-dist"
+    (user_dist / "assets").mkdir(parents=True)
+    (bundled_dist / "assets").mkdir(parents=True)
+    (user_dist / "assets" / "version").write_text("v4.24.2", encoding="utf-8")
+    (bundled_dist / "assets" / "version").write_text("v4.24.4", encoding="utf-8")
+
+    with mock.patch(
+        "astrbot.core.utils.io.get_bundled_dashboard_dist_path",
+        return_value=bundled_dist,
+    ):
+        assert should_use_bundled_dashboard_dist(user_dist, "v4.24.4") is True
+
+
+def test_should_keep_data_dist_when_version_file_is_malformed(tmp_path):
+    user_dist = tmp_path / "user-dist"
+    bundled_dist = tmp_path / "bundled-dist"
+    (user_dist / "assets").mkdir(parents=True)
+    (bundled_dist / "assets").mkdir(parents=True)
+    (user_dist / "assets" / "version").write_text("not-a-version", encoding="utf-8")
+
+    with mock.patch(
+        "astrbot.core.utils.io.get_bundled_dashboard_dist_path",
+        return_value=bundled_dist,
+    ):
+        assert should_use_bundled_dashboard_dist(user_dist, "4.24.4") is False
+
+
+@pytest.mark.asyncio
+async def test_check_dashboard_files_uses_bundled_dist_when_data_dist_is_stale(
+    tmp_path,
+):
+    """Tests that a stale data/dist does not override bundled dashboard assets."""
+    data_dir = tmp_path / "data"
+    data_dist = data_dir / "dist"
+    bundled_dist = tmp_path / "bundled-dist"
+    data_dist.mkdir(parents=True)
+    bundled_dist.mkdir()
+
+    with mock.patch("main.get_astrbot_data_path", return_value=str(data_dir)):
+        with mock.patch(
+            "main.get_dashboard_version", mock.AsyncMock(return_value="v0.0.1")
+        ):
+            with mock.patch(
+                "main.should_use_bundled_dashboard_dist", return_value=True
+            ):
+                with mock.patch(
+                    "main.get_bundled_dashboard_dist_path",
+                    return_value=Path(bundled_dist),
+                ):
+                    with mock.patch("main.download_dashboard") as mock_download:
+                        result = await check_dashboard_files()
+
+    assert result == str(bundled_dist)
+    mock_download.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_check_dashboard_files_with_webui_dir_arg(monkeypatch):
     """Tests that providing a valid webui_dir skips all checks."""