diff --git a/astrbot/core/star/updator.py b/astrbot/core/star/updator.py index 45f8b8a23..14cb5331a 100644 --- a/astrbot/core/star/updator.py +++ b/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) diff --git a/astrbot/core/zip_updator.py b/astrbot/core/zip_updator.py index 137c7444a..2d2b7b834 100644 --- a/astrbot/core/zip_updator.py +++ b/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()