fix: type release fetch errors and streamline updater resolution

Author: zouyonghe

astrbot/core/updator.py

@@ -1,4 +1,3 @@
-import asyncio
 import os
 import sys
 import time
@@ -12,7 +11,7 @@
 from astrbot.core.utils.astrbot_path import get_astrbot_path
 from astrbot.core.utils.io import download_file
 
-from .zip_updator import ReleaseInfo, RepoZipUpdator
+from .zip_updator import FetchReleaseError, ReleaseInfo, RepoZipUpdator
 
 
 class AstrBotUpdator(RepoZipUpdator):
@@ -154,25 +153,23 @@ async def get_releases(self, latest: bool = True) -> list:
     @staticmethod
     def _is_expected_nightly_fetch_error(exc: BaseException) -> bool:
         expected_types = (
-            asyncio.TimeoutError,
+            TimeoutError,
             aiohttp.ClientError,
             JSONDecodeError,
+            FetchReleaseError,
         )
-
-        def _matches(error: BaseException | None) -> bool:
-            if error is None:
-                return False
-            if isinstance(error, expected_types):
-                return True
-            if str(error).startswith("请求失败，状态码:"):
+        current: BaseException | None = exc
+        seen: set[int] = set()
+        while current is not None and id(current) not in seen:
+            seen.add(id(current))
+            if isinstance(current, expected_types):
                 return True
-            return False
-
-        return (
-            _matches(exc)
-            or _matches(getattr(exc, "__cause__", None))
-            or _matches(getattr(exc, "__context__", None))
-        )
+            current = getattr(current, "__cause__", None) or getattr(
+                current,
+                "__context__",
+                None,
+            )
+        return False
 
     async def get_nightly_release(self) -> dict | None:
         nightly_release_url = f"{self.GITHUB_RELEASE_API}/tags/{self.NIGHTLY_TAG}"
@@ -204,35 +201,43 @@ async def get_releases_with_nightly(self, latest: bool = True) -> list:
             releases.insert(0, nightly_release)
         return releases
 
-    def _resolve_latest(self, releases: list) -> tuple[str, str]:
-        latest_release = next(
-            (
-                item
-                for item in releases
-                if item.get("tag_name", "").lower() != self.NIGHTLY_TAG
-            ),
-            None,
-        )
-        if latest_release is None:
-            raise Exception("未找到可用的发布版本。")
+    async def _resolve_update_target(
+        self,
+        latest: bool,
+        version: str | None,
+    ) -> tuple[str, str]:
+        version_str = str(version) if version is not None else ""
+
+        if latest:
+            releases = await self.get_releases(latest=True)
+            latest_release = next(
+                (
+                    item
+                    for item in releases
+                    if item.get("tag_name", "").lower() != self.NIGHTLY_TAG
+                ),
+                None,
+            )
+            if latest_release is None:
+                raise Exception("未找到可用的发布版本。")
 
-        latest_version = latest_release["tag_name"]
-        if self.compare_version(VERSION, latest_version) >= 0:
-            raise Exception("当前已经是最新版本。")
-        return latest_version, latest_release["zipball_url"]
+            latest_version = latest_release["tag_name"]
+            if self.compare_version(VERSION, latest_version) >= 0:
+                raise Exception("当前已经是最新版本。")
+            return latest_version, latest_release["zipball_url"]
 
-    def _resolve_nightly(self) -> tuple[str, str]:
-        return self.NIGHTLY_TAG, (
-            f"https://github.com/AstrBotDevs/AstrBot/archive/refs/tags/{self.NIGHTLY_TAG}.zip"
-        )
+        if version_str.lower() == self.NIGHTLY_TAG:
+            return self.NIGHTLY_TAG, (
+                f"https://github.com/AstrBotDevs/AstrBot/archive/refs/tags/{self.NIGHTLY_TAG}.zip"
+            )
 
-    def _resolve_tag(self, releases: list, version_str: str) -> tuple[str, str]:
-        for data in releases:
-            if data["tag_name"] == version_str:
-                return version_str, data["zipball_url"]
-        raise Exception(f"未找到版本号为 {version_str} 的更新文件。")
+        if version_str.startswith("v"):
+            releases = await self.get_releases(latest=False)
+            for data in releases:
+                if data["tag_name"] == version_str:
+                    return version_str, data["zipball_url"]
+            raise Exception(f"未找到版本号为 {version_str} 的更新文件。")
 
-    def _resolve_commit(self, version_str: str) -> tuple[str, str]:
         if len(version_str) != 40:
             raise Exception("commit hash 长度不正确，应为 40")
         return (
@@ -246,17 +251,7 @@ async def update(self, reboot=False, latest=True, version=None, proxy="") -> Non
                 "Error: You are running AstrBot via CLI, please use `pip` or `uv tool upgrade` to update AstrBot."
             )  # 避免版本管理混乱
 
-        version_str = str(version) if version is not None else ""
-        if latest:
-            releases = await self.get_releases(latest=True)
-            target_version, file_url = self._resolve_latest(releases)
-        elif version_str.lower() == self.NIGHTLY_TAG:
-            target_version, file_url = self._resolve_nightly()
-        elif version_str.startswith("v"):
-            releases = await self.get_releases(latest=False)
-            target_version, file_url = self._resolve_tag(releases, version_str)
-        else:
-            target_version, file_url = self._resolve_commit(version_str)
+        target_version, file_url = await self._resolve_update_target(latest, version)
 
         logger.info(f"准备更新至 AstrBot Core: {target_version}")
 

astrbot/core/zip_updator.py

@@ -3,6 +3,7 @@
 import shutil
 import ssl
 import zipfile
+from json import JSONDecodeError
 from typing import NoReturn
 
 import aiohttp
@@ -38,6 +39,10 @@ def __str__(self) -> str:
         return f"\n{self.body}\n\n版本: {self.version} | 发布于: {self.published_at}"
 
 
+class FetchReleaseError(Exception):
+    """Expected errors while fetching release metadata from remote services."""
+
+
 class RepoZipUpdator:
     def __init__(self, repo_mirror: str = "") -> None:
         self.repo_mirror = repo_mirror
@@ -67,30 +72,33 @@ async def fetch_release_info(self, url: str, latest: bool = True) -> list:
                     logger.error(
                         f"请求 {url} 失败，状态码: {response.status}, 内容: {text}",
                     )
-                    raise Exception(f"请求失败，状态码: {response.status}")
+                    raise FetchReleaseError(f"请求失败，状态码: {response.status}")
                 result = await response.json()
-            if isinstance(result, dict):
-                result = [result]
-            if not result:
-                return []
-            # if latest:
-            #     ret = self.github_api_release_parser([result[0]])
-            # else:
-            #     ret = self.github_api_release_parser(result)
-            ret = []
-            for release in result:
-                ret.append(
-                    {
-                        "version": release["name"],
-                        "published_at": release["published_at"],
-                        "body": release["body"],
-                        "tag_name": release["tag_name"],
-                        "zipball_url": release["zipball_url"],
-                    },
-                )
-        except Exception as e:
+        except FetchReleaseError:
+            raise
+        except (TimeoutError, aiohttp.ClientError, JSONDecodeError) as e:
             logger.error(f"解析版本信息时发生异常: {e}")
-            raise Exception("解析版本信息失败") from e
+            raise FetchReleaseError("解析版本信息失败") from e
+
+        if isinstance(result, dict):
+            result = [result]
+        if not result:
+            return []
+        # if latest:
+        #     ret = self.github_api_release_parser([result[0]])
+        # else:
+        #     ret = self.github_api_release_parser(result)
+        ret = []
+        for release in result:
+            ret.append(
+                {
+                    "version": release["name"],
+                    "published_at": release["published_at"],
+                    "body": release["body"],
+                    "tag_name": release["tag_name"],
+                    "zipball_url": release["zipball_url"],
+                },
+            )
         return ret
 
     def github_api_release_parser(self, releases: list) -> list:

tests/unit/test_prerelease_rule_sync.py

@@ -9,6 +9,11 @@ def test_prerelease_rule_is_synced_with_dashboard():
     vue_path = repo_root / "dashboard/src/layouts/full/vertical-header/VerticalHeader.vue"
     content = vue_path.read_text(encoding="utf-8")
 
-    match = re.search(r"const PRE_RELEASE_TAG_REGEX = /(.+?)/i;", content)
+    match = re.search(
+        r"const\s+PRE_RELEASE_TAG_REGEX\s*=\s*/(.+?)/([a-z]*)\s*;?",
+        content,
+    )
     assert match is not None
-    assert match.group(1) == PRERELEASE_TAG_REGEX.pattern
+    vue_pattern, vue_flags = match.groups()
+    assert vue_pattern == PRERELEASE_TAG_REGEX.pattern
+    assert ("i" in vue_flags) == bool(PRERELEASE_TAG_REGEX.flags & re.IGNORECASE)

tests/unit/test_updator.py

@@ -1,7 +1,7 @@
 import pytest
 
 from astrbot.core.updator import AstrBotUpdator
-from astrbot.core.zip_updator import RepoZipUpdator
+from astrbot.core.zip_updator import FetchReleaseError, RepoZipUpdator
 
 
 @pytest.mark.asyncio
@@ -136,7 +136,7 @@ async def test_get_nightly_release_returns_none_for_expected_fetch_error(monkeyp
 
     async def mock_fetch_release_info(url: str, latest: bool = True):
         _ = url, latest
-        raise Exception("请求失败，状态码: 404")
+        raise FetchReleaseError("请求失败，状态码: 404")
 
     monkeypatch.setattr(updator, "fetch_release_info", mock_fetch_release_info)