🎈 perf(github_utils): 支持github url下载遍历 (#1632)

* 🎈 perf(github_utils): 支持github url下载遍历

* 🐞 fix(http_utils): 修复一些下载问题

* 🦄 refactor(http_utils): 部分重构

* chore(version): Update version to v0.2.2-e6f17c4

---------

Co-authored-by: AkashiCoin <AkashiCoin@users.noreply.github.com>

zhenxun/utils/github_utils/__init__.py

@@ -1,23 +1,27 @@
+from collections.abc import Generator
+
 from .consts import GITHUB_REPO_URL_PATTERN
-from .func import get_fastest_raw_format, get_fastest_archive_format
+from .func import get_fastest_raw_formats, get_fastest_archive_formats
 from .models import RepoAPI, RepoInfo, GitHubStrategy, JsdelivrStrategy
 
 __all__ = [
-    "parse_github_url",
-    "get_fastest_raw_format",
-    "get_fastest_archive_format",
-    "api_strategy",
+    "get_fastest_raw_formats",
+    "get_fastest_archive_formats",
+    "GithubUtils",
 ]
 
 
-def parse_github_url(github_url: str) -> "RepoInfo":
-    if matched := GITHUB_REPO_URL_PATTERN.match(github_url):
-        return RepoInfo(**{k: v for k, v in matched.groupdict().items() if v})
-    raise ValueError("github地址格式错误")
+class GithubUtils:
+    # 使用
+    jsdelivr_api = RepoAPI(JsdelivrStrategy())  # type: ignore
+    github_api = RepoAPI(GitHubStrategy())  # type: ignore
 
+    @classmethod
+    def iter_api_strategies(cls) -> Generator[RepoAPI]:
+        yield from [cls.github_api, cls.jsdelivr_api]
 
-# 使用
-jsdelivr_api = RepoAPI(JsdelivrStrategy())  # type: ignore
-github_api = RepoAPI(GitHubStrategy())  # type: ignore
-
-api_strategy = [github_api, jsdelivr_api]
+    @classmethod
+    def parse_github_url(cls, github_url: str) -> "RepoInfo":
+        if matched := GITHUB_REPO_URL_PATTERN.match(github_url):
+            return RepoInfo(**{k: v for k, v in matched.groupdict().items() if v})
+        raise ValueError("github地址格式错误")

zhenxun/utils/github_utils/func.py

@@ -9,15 +9,15 @@ from .consts import (
 )
 
 
-async def __get_fastest_format(formats: dict[str, str]) -> str:
+async def __get_fastest_formats(formats: dict[str, str]) -> list[str]:
     sorted_urls = await AsyncHttpx.get_fastest_mirror(list(formats.keys()))
     if not sorted_urls:
         raise Exception("无法获取任意GitHub资源加速地址，请检查网络")
-    return formats[sorted_urls[0]]
+    return [formats[url] for url in sorted_urls]
 
 
 @cached()
-async def get_fastest_raw_format() -> str:
+async def get_fastest_raw_formats() -> list[str]:
     """获取最快的raw下载地址格式"""
     formats: dict[str, str] = {
         "https://raw.githubusercontent.com/": RAW_CONTENT_FORMAT,
@@ -26,11 +26,11 @@ async def get_fastest_raw_format() -> str:
         "https://gh-proxy.com/": f"https://gh-proxy.com/{RAW_CONTENT_FORMAT}",
         "https://cdn.jsdelivr.net/": "https://cdn.jsdelivr.net/gh/{owner}/{repo}@{branch}/{path}",
     }
-    return await __get_fastest_format(formats)
+    return await __get_fastest_formats(formats)
 
 
 @cached()
-async def get_fastest_archive_format() -> str:
+async def get_fastest_archive_formats() -> list[str]:
     """获取最快的归档下载地址格式"""
     formats: dict[str, str] = {
         "https://github.com/": ARCHIVE_URL_FORMAT,
@@ -38,11 +38,11 @@ async def get_fastest_archive_format() -> str:
         "https://mirror.ghproxy.com/": f"https://mirror.ghproxy.com/{ARCHIVE_URL_FORMAT}",
         "https://gh-proxy.com/": f"https://gh-proxy.com/{ARCHIVE_URL_FORMAT}",
     }
-    return await __get_fastest_format(formats)
+    return await __get_fastest_formats(formats)
 
 
 @cached()
-async def get_fastest_release_format() -> str:
+async def get_fastest_release_formats() -> list[str]:
     """获取最快的发行版资源下载地址格式"""
     formats: dict[str, str] = {
         "https://objects.githubusercontent.com/": RELEASE_ASSETS_FORMAT,
@@ -50,14 +50,14 @@ async def get_fastest_release_format() -> str:
         "https://mirror.ghproxy.com/": f"https://mirror.ghproxy.com/{RELEASE_ASSETS_FORMAT}",
         "https://gh-proxy.com/": f"https://gh-proxy.com/{RELEASE_ASSETS_FORMAT}",
     }
-    return await __get_fastest_format(formats)
+    return await __get_fastest_formats(formats)
 
 
 @cached()
-async def get_fastest_release_source_format() -> str:
+async def get_fastest_release_source_formats() -> list[str]:
     """获取最快的发行版源码下载地址格式"""
     formats: dict[str, str] = {
         "https://codeload.github.com/": RELEASE_SOURCE_FORMAT,
         "https://p.102333.xyz/": f"https://p.102333.xyz/{RELEASE_SOURCE_FORMAT}",
     }
-    return await __get_fastest_format(formats)
+    return await __get_fastest_formats(formats)

zhenxun/utils/github_utils/models.py

@@ -7,9 +7,9 @@ from pydantic import BaseModel
 from ..http_utils import AsyncHttpx
 from .consts import CACHED_API_TTL, GIT_API_TREES_FORMAT, JSD_PACKAGE_API_FORMAT
 from .func import (
-    get_fastest_raw_format,
-    get_fastest_archive_format,
-    get_fastest_release_source_format,
+    get_fastest_raw_formats,
+    get_fastest_archive_formats,
+    get_fastest_release_source_formats,
 )
 
 
@@ -20,21 +20,41 @@ class RepoInfo(BaseModel):
     repo: str
     branch: str = "main"
 
-    async def get_raw_download_url(self, path: str):
-        url_format = await get_fastest_raw_format()
-        return url_format.format(**self.dict(), path=path)
+    async def get_raw_download_url(self, path: str) -> str:
+        return (await self.get_raw_download_urls(path))[0]
 
-    async def get_archive_download_url(self):
-        url_format = await get_fastest_archive_format()
-        return url_format.format(**self.dict())
+    async def get_archive_download_url(self) -> str:
+        return (await self.get_archive_download_urls())[0]
 
-    async def get_release_source_download_url_tgz(self, version: str):
-        url_format = await get_fastest_release_source_format()
-        return url_format.format(**self.dict(), version=version, compress="tar.gz")
+    async def get_release_source_download_url_tgz(self, version: str) -> str:
+        return (await self.get_release_source_download_urls_tgz(version))[0]
 
-    async def get_release_source_download_url_zip(self, version: str):
-        url_format = await get_fastest_release_source_format()
-        return url_format.format(**self.dict(), version=version, compress="zip")
+    async def get_release_source_download_url_zip(self, version: str) -> str:
+        return (await self.get_release_source_download_urls_zip(version))[0]
+
+    async def get_raw_download_urls(self, path: str) -> list[str]:
+        url_formats = await get_fastest_raw_formats()
+        return [
+            url_format.format(**self.dict(), path=path) for url_format in url_formats
+        ]
+
+    async def get_archive_download_urls(self) -> list[str]:
+        url_formats = await get_fastest_archive_formats()
+        return [url_format.format(**self.dict()) for url_format in url_formats]
+
+    async def get_release_source_download_urls_tgz(self, version: str) -> list[str]:
+        url_formats = await get_fastest_release_source_formats()
+        return [
+            url_format.format(**self.dict(), version=version, compress="tar.gz")
+            for url_format in url_formats
+        ]
+
+    async def get_release_source_download_urls_zip(self, version: str) -> list[str]:
+        url_formats = await get_fastest_release_source_formats()
+        return [
+            url_format.format(**self.dict(), version=version, compress="zip")
+            for url_format in url_formats
+        ]
 
 
 class APIStrategy(Protocol):