Merge pull request #1735 from Flartiny/dev

feat: able to parse repo url of specific branch

astrbot/core/star/updator.py

@@ -18,7 +18,8 @@ class PluginUpdator(RepoZipUpdator):
         return self.plugin_store_path
 
     async def install(self, repo_url: str, proxy="") -> str:
-        repo_name = self.format_repo_name(repo_url)
+        _, repo_name, _ = self.parse_github_url(repo_url)
+        repo_name = self.format_name(repo_name)
         plugin_path = os.path.join(self.plugin_store_path, repo_name)
         await self.download_from_repo_url(plugin_path, repo_url, proxy)
         self.unzip_file(plugin_path + ".zip", plugin_path)
@@ -54,7 +55,7 @@ class PluginUpdator(RepoZipUpdator):
     def unzip_file(self, zip_path: str, target_dir: str):
         os.makedirs(target_dir, exist_ok=True)
         update_dir = ""
-        logger.info(f"解压文件: {zip_path}")
+        logger.info(f"正在解压压缩包: {zip_path}")
         with zipfile.ZipFile(zip_path, "r") as z:
             update_dir = z.namelist()[0]
             z.extractall(target_dir)

astrbot/core/zip_updator.py

@@ -1,5 +1,6 @@
 import aiohttp
 import os
+import re
 import zipfile
 import shutil
 
@@ -119,28 +120,60 @@ class RepoZipUpdator:
         )
 
     async def download_from_repo_url(self, target_path: str, repo_url: str, proxy=""):
-        repo_namespace = repo_url.split("/")[-2:]
-        author = repo_namespace[0]
-        repo = repo_namespace[1]
+        author, repo, branch = self.parse_github_url(repo_url)
 
         logger.info(f"正在下载更新 {repo} ...")
-        release_url = f"https://api.github.com/repos/{author}/{repo}/releases"
-        releases = await self.fetch_release_info(url=release_url)
-        if not releases:
-            # download from the default branch directly.
-            logger.info(f"正在从默认分支下载 {author}/{repo} ")
+
+        if branch:
+            logger.info(f"正在从指定分支 {branch} 下载 {author}/{repo}")
             release_url = (
-                f"https://github.com/{author}/{repo}/archive/refs/heads/master.zip"
+                f"https://github.com/{author}/{repo}/archive/refs/heads/{branch}.zip"
             )
         else:
-            release_url = releases[0]["zipball_url"]
+            try:
+                release_url = f"https://api.github.com/repos/{author}/{repo}/releases"
+                releases = await self.fetch_release_info(url=release_url)
+            except Exception as e:
+                logger.warning(
+                    f"获取 {author}/{repo} 的 GitHub Releases 失败: {e}，将尝试下载默认分支"
+                )
+                releases = []
+            if not releases:
+                # 如果没有最新版本，下载默认分支
+                logger.info(f"正在从默认分支下载 {author}/{repo}")
+                release_url = (
+                    f"https://github.com/{author}/{repo}/archive/refs/heads/master.zip"
+                )
+            else:
+                release_url = releases[0]["zipball_url"]
 
         if proxy:
             release_url = f"{proxy}/{release_url}"
-            logger.info(f"使用代理下载: {release_url}")
+            logger.info(
+                f"检查到设置了镜像站，将使用镜像站下载 {author}/{repo} 仓库源码: {release_url}"
+            )
 
         await download_file(release_url, target_path + ".zip")
 
+    def parse_github_url(self, url: str):
+        """使用正则表达式解析 GitHub 仓库 URL，支持 `.git` 后缀和 `tree/branch` 结构
+        Returns:
+            tuple[str, str, str]: 返回作者名、仓库名和分支名
+        Raises:
+            ValueError: 如果 URL 格式不正确
+        """
+        cleaned_url = url.rstrip("/")
+        pattern = r"^https://github\.com/([a-zA-Z0-9_-]+)/([a-zA-Z0-9_-]+)(\.git)?(?:/tree/([a-zA-Z0-9_-]+))?$"
+        match = re.match(pattern, cleaned_url)
+
+        if match:
+            author = match.group(1)
+            repo = match.group(2)
+            branch = match.group(4)
+            return author, repo, branch
+        else:
+            raise ValueError("无效的 GitHub URL")
+
     def unzip_file(self, zip_path: str, target_dir: str):
         """
         解压缩文件, 并将压缩包内**第一个**文件夹内的文件移动到 target_dir
@@ -174,16 +207,5 @@ class RepoZipUpdator:
                 f"删除更新文件失败，可以手动删除 {zip_path} 和 {os.path.join(target_dir, update_dir)}"
             )
 
-    def format_repo_name(self, repo_url: str) -> str:
-        if repo_url.endswith("/"):
-            repo_url = repo_url[:-1]
-
-        repo_namespace = repo_url.split("/")[-2:]
-        repo = repo_namespace[1]
-
-        repo = self.format_name(repo)
-
-        return repo
-
     def format_name(self, name: str) -> str:
         return name.replace("-", "_").lower()