♻️ refactor(repo): 统一仓库最新提交获取接口并迁移文件提交日期逻辑

zhenxun/builtin_plugins/auto_update/_data_source.py

@@ -6,7 +6,6 @@ from packaging.specifiers import SpecifierSet
 from packaging.version import InvalidVersion, Version
 
 from zhenxun.services.log import logger
-from zhenxun.utils.http_utils import AsyncHttpx
 from zhenxun.utils.manager.virtual_env_package_manager import VirtualEnvPackageManager
 from zhenxun.utils.manager.zhenxun_repo_manager import (
     ZhenxunRepoConfig,
@@ -19,19 +18,6 @@ LOG_COMMAND = "AutoUpdate"
 
 
 class UpdateManager:
-    @staticmethod
-    async def _get_latest_commit_date(owner: str, repo: str, path: str) -> str:
-        """获取文件最新 commit 日期"""
-        api_url = f"https://api.github.com/repos/{owner}/{repo}/commits"
-        params = {"path": path, "page": 1, "per_page": 1}
-        try:
-            data = await AsyncHttpx.get_json(api_url, params=params)
-            if data and isinstance(data, list) and data[0]:
-                date_str = data[0]["commit"]["committer"]["date"]
-                return date_str.split("T")[0]
-        except Exception as e:
-            logger.warning(f"获取 {owner}/{repo}/{path} 的 commit 日期失败", e=e)
-        return "获取失败"
 
     @classmethod
     async def check_version(cls) -> str:
@@ -42,11 +28,11 @@ class UpdateManager:
         dev_version_task = RepoFileManager.get_file_content(
             ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
         )
-        bot_commit_date_task = cls._get_latest_commit_date(
-            "HibiKier", "zhenxun_bot", "__version__"
+        bot_commit_date_task = RepoFileManager.get_file_last_commit_date(
+            ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__"
         )
-        res_commit_date_task = cls._get_latest_commit_date(
-            "zhenxun-org", "zhenxun-bot-resources", "__version__"
+        res_commit_date_task = RepoFileManager.get_file_last_commit_date(
+            ZhenxunRepoConfig.RESOURCE_GITHUB_URL, "__version__"
         )
 
         (

zhenxun/utils/github_utils/const.py

@@ -40,6 +40,9 @@ RELEASE_SOURCE_FORMAT = (
 GIT_API_COMMIT_FORMAT = "https://api.github.com/repos/{owner}/{repo}/commits/{branch}"
 """git api commit地址格式"""
 
+GIT_API_COMMIT_LIST_FORMAT = "https://api.github.com/repos/{owner}/{repo}/commits"
+"""git api 列出commits的地址格式"""
+
 GIT_API_PROXY_COMMIT_FORMAT = (
     "https://git-api.zhenxun.org/repos/{owner}/{repo}/commits/{branch}"
 )

zhenxun/utils/repo_utils/aliyun_manager.py

@@ -348,6 +348,11 @@ class AliyunCodeupManager(BaseRepoManager):
         if not self.config.aliyun_codeup.organization_id:
             raise AuthenticationError("阿里云CodeUp")
 
+    async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
+        """获取阿里云CodeUp仓库指定分支的最新提交哈希值。"""
+        repo_name = repo_url.split("/tree/")[0].split("/")[-1].replace(".git", "")
+        return await self._get_newest_commit(repo_name, branch)
+
     async def _get_newest_commit(self, repo_name: str, branch: str) -> str:
         """
         获取仓库最新提交ID

zhenxun/utils/repo_utils/base_manager.py

@@ -117,6 +117,20 @@ class BaseRepoManager(ABC):
         """
         pass
 
+    @abstractmethod
+    async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
+        """
+        获取仓库指定分支的最新提交哈希值。
+
+        参数:
+            repo_url: 仓库URL或名称。
+            branch: 分支名称。
+
+        返回:
+            str: 最新的提交哈希值。
+        """
+        pass
+
     async def save_file_content(self, content: bytes, local_path: Path) -> int:
         """
         保存文件内容

zhenxun/utils/repo_utils/file_manager.py

@@ -11,6 +11,7 @@ from httpx import Response
 
 from zhenxun.services.log import logger
 from zhenxun.utils.github_utils import GithubUtils
+from zhenxun.utils.github_utils.const import GIT_API_COMMIT_LIST_FORMAT
 from zhenxun.utils.github_utils.models import AliyunTreeType, GitHubStrategy, TreeType
 from zhenxun.utils.http_utils import AsyncHttpx
 from zhenxun.utils.utils import is_binary_file
@@ -633,3 +634,33 @@ class RepoFileManager:
             result.success = False
             result.error_message = str(e)
             return result
+    async def get_file_last_commit_date(
+        self, repo_url: str, file_path: str
+    ) -> str | None:
+        """
+        获取 GitHub 仓库中指定文件的最新提交日期。
+
+        参数:
+            repo_url: 仓库的URL。
+            file_path: 文件在仓库中的路径。
+
+        返回:
+            str | None: "YYYY-MM-DD" 格式的日期字符串，如果失败则返回 None。
+        """
+        try:
+            repo_info = GithubUtils.parse_github_url(repo_url)
+            api_url = GIT_API_COMMIT_LIST_FORMAT.format(
+                owner=repo_info.owner, repo=repo_info.repo
+            )
+            params = {"sha": repo_info.branch, "path": file_path,
+                      "page": 1, "per_page": 1}
+
+            data = await AsyncHttpx.get_json(api_url, params=params)
+            if data and isinstance(data, list) and data[0]:
+                date_str = data[0]["commit"]["committer"]["date"]
+                return date_str.split("T")[0]
+        except Exception as e:
+            logger.warning(
+                f"获取 {repo_url} 中 {file_path} 的 commit 日期失败", LOG_COMMAND, e=e
+            )
+        return None

zhenxun/utils/repo_utils/github_manager.py

@@ -320,6 +320,12 @@ class GithubManager(BaseRepoManager):
             logger.error("获取提交信息失败", LOG_COMMAND, e=e)
             return None
 
+    async def get_latest_commit(self, repo_url: str, branch: str = "main") -> str:
+        """获取GitHub仓库指定分支的最新提交哈希值。"""
+        repo_info = GithubUtils.parse_github_url(repo_url)
+        repo_name = repo_info.repo.replace(".git", "")
+        return await self._get_newest_commit(repo_info.owner, repo_name, branch)
+
     async def _get_newest_commit(self, owner: str, repo: str, branch: str) -> str:
         """
         获取仓库最新提交ID