refactor: simplify update target resolution flow

Author: zouyonghe

astrbot/core/updator.py

@@ -1,5 +1,4 @@
 import os
-import re
 import sys
 import time
 from json import JSONDecodeError
@@ -34,6 +33,7 @@ def __init__(self, repo_mirror: str = "") -> None:
         self.GITHUB_RELEASE_API = (
             "https://api.github.com/repos/AstrBotDevs/AstrBot/releases"
         )
+        self.GITHUB_ARCHIVE_BASE = "https://github.com/AstrBotDevs/AstrBot/archive"
         self.NIGHTLY_TAG = NIGHTLY_TAG
 
     def terminate_child_processes(self) -> None:
@@ -201,51 +201,14 @@ async def get_releases_with_nightly(self) -> list:
             releases.insert(0, nightly_release)
         return releases
 
-    async def _resolve_latest_target(self) -> tuple[str, str]:
-        releases = await self.get_releases()
-        latest_release = next(
-            (
-                item
-                for item in releases
-                if (tag := item.get("tag_name", ""))
-                and tag.lower() != self.NIGHTLY_TAG
-                and not PRERELEASE_TAG_REGEX.search(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"]
-
-    def _resolve_github_archive_base(self) -> str:
-        match = re.search(
-            r"/repos/([^/]+)/([^/]+)/releases/?$",
-            self.GITHUB_RELEASE_API,
-        )
-        if match is None:
-            raise Exception("GITHUB_RELEASE_API 格式不正确，无法解析仓库信息。")
-        owner, repo = match.groups()
-        return f"https://github.com/{owner}/{repo}/archive"
-
     def _resolve_nightly_target(self) -> tuple[str, str]:
-        archive_base = self._resolve_github_archive_base()
+        archive_base = self.GITHUB_ARCHIVE_BASE
         return self.NIGHTLY_TAG, (f"{archive_base}/refs/tags/{self.NIGHTLY_TAG}.zip")
 
-    async def _resolve_tag_target(self, version_str: str) -> tuple[str, str]:
-        releases = await self.get_releases()
-        for data in releases:
-            if data["tag_name"] == version_str:
-                return version_str, data["zipball_url"]
-        raise Exception(f"未找到版本号为 {version_str} 的更新文件。")
-
     def _resolve_commit_target(self, version_str: str) -> tuple[str, str]:
         if len(version_str) != 40:
             raise Exception("commit hash 长度不正确，应为 40")
-        archive_base = self._resolve_github_archive_base()
+        archive_base = self.GITHUB_ARCHIVE_BASE
         return (
             version_str,
             f"{archive_base}/{version_str}.zip",
@@ -256,16 +219,37 @@ async def _resolve_update_target(
         latest: bool,
         version: str | None,
     ) -> tuple[str, str]:
-        version_str = str(version) if version is not None else ""
+        version_str = str(version).strip() if version is not None else ""
 
         if latest:
-            return await self._resolve_latest_target()
+            releases = await self.get_releases()
+            latest_release = next(
+                (
+                    item
+                    for item in releases
+                    if (tag := item.get("tag_name", ""))
+                    and tag.lower() != self.NIGHTLY_TAG
+                    and not PRERELEASE_TAG_REGEX.search(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"]
 
         if version_str.lower() == self.NIGHTLY_TAG:
             return self._resolve_nightly_target()
 
         if version_str.startswith("v"):
-            return await self._resolve_tag_target(version_str)
+            releases = await self.get_releases()
+            for data in releases:
+                if data.get("tag_name") == version_str:
+                    return version_str, data["zipball_url"]
+            raise Exception(f"未找到版本号为 {version_str} 的更新文件。")
 
         return self._resolve_commit_target(version_str)
 

tests/unit/test_updator.py

@@ -125,11 +125,9 @@ def mock_unzip_file(zip_path: str, target_dir: str):
     assert captured["target_dir"] == str(tmp_path)
 
 
-def test_resolve_nightly_target_uses_repo_from_release_api():
+def test_resolve_nightly_target_uses_archive_base():
     updator = AstrBotUpdator()
-    updator.GITHUB_RELEASE_API = (
-        "https://api.github.com/repos/example-org/example-repo/releases"
-    )
+    updator.GITHUB_ARCHIVE_BASE = "https://github.com/example-org/example-repo/archive"
 
     target_version, file_url = updator._resolve_nightly_target()
     assert target_version == "nightly"
@@ -139,11 +137,9 @@ def test_resolve_nightly_target_uses_repo_from_release_api():
     )
 
 
-def test_resolve_commit_target_uses_repo_from_release_api():
+def test_resolve_commit_target_uses_archive_base():
     updator = AstrBotUpdator()
-    updator.GITHUB_RELEASE_API = (
-        "https://api.github.com/repos/example-org/example-repo/releases"
-    )
+    updator.GITHUB_ARCHIVE_BASE = "https://github.com/example-org/example-repo/archive"
     version_str = "1234567890123456789012345678901234567890"
 
     target_version, file_url = updator._resolve_commit_target(version_str)
@@ -247,7 +243,7 @@ async def mock_fetch_release_info(url: str):
 
 
 @pytest.mark.asyncio
-async def test_resolve_latest_target_skips_prerelease_tags(monkeypatch):
+async def test_resolve_update_target_skips_prerelease_tags_for_latest(monkeypatch):
     updator = AstrBotUpdator()
     releases = [
         {
@@ -272,7 +268,10 @@ async def mock_get_releases():
     monkeypatch.setattr(updator, "get_releases", mock_get_releases)
     monkeypatch.setattr(updator, "compare_version", lambda _current, _target: -1)
 
-    target_version, file_url = await updator._resolve_latest_target()
+    target_version, file_url = await updator._resolve_update_target(
+        latest=True,
+        version=None,
+    )
     assert target_version == "v9.9.8"
     assert file_url == "https://example.com/stable.zip"