From 023e865f34bf627d432be4115192d6b4c08fdc9c Mon Sep 17 00:00:00 2001 From: molanp <104612722+molanp@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:31:23 +0800 Subject: [PATCH] =?UTF-8?q?fix(plugin-store-downlaod):=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E6=8F=92=E4=BB=B6=E5=95=86=E5=BA=97=E4=B8=8B=E8=BD=BD?= =?UTF-8?q?=E4=BA=8C=E8=BF=9B=E5=88=B6=E6=96=87=E4=BB=B6=E9=94=99=E8=AF=AF?= =?UTF-8?q?=20(#2152)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(plugin-store-downlaod): 修复插件商店下载二进制文件错误 * fix(plugin_store): 修复插件商店中仓库类型判断逻辑错误 当处理外部插件时，修正了仓库类型的判断条件，确保正确的仓库类型被设置。 同时修复了目录文件列表获取时传递错误的仓库类型参数问题。 * feat(repo_utils): 更新文件管理器返回类型并改进仓库URL处理 - 将 RepoFileManager 的文件内容返回类型从 bytes 改为 str， 统一字符串类型的返回值格式 - 支持在 repo_url 中直接包含 /tree/ 路径， 自动解析分支信息并正确处理仓库地址 - 改进 GitHub 和阿里云仓库的优先级顺序， 现在优先尝试阿里云然后是 GitHub - 修复 clone_url 处理逻辑，确保正确的 .git 后缀添加 - 移除未使用的 Sequence 类型导入，统一使用 list[str] 类型注解 * refactor(plugin_store): 移除StoreManager中不必要的repo_type参数 移除了data_source.py文件中StoreManager类的list_directory_files方法调用时 多余的repo_type=RepoType.GITHUB参数 * feat(plugin_store): 优化插件依赖文件下载和安装逻辑 - 移动依赖文件下载逻辑到下载阶段，避免重复下载 - 使用随机数生成唯一的临时依赖文件名以避免冲突 - 添加ignore_error参数来处理依赖文件下载失败的情况 - 统一依赖文件路径变量命名，提高代码可读性 - 确保requirement.txt和requirements.txt都能正确安装 * Add repo_type parameter to list_directory_files call --- .../auto_update/_data_source.py | 6 +- .../plugin_store/data_source.py | 42 +-- zhenxun/utils/github_utils/models.py | 10 +- zhenxun/utils/repo_utils/file_manager.py | 297 ++++++------------ zhenxun/utils/repo_utils/utils.py | 52 +-- zhenxun/utils/utils.py | 36 --- 6 files changed, 152 insertions(+), 291 deletions(-) diff --git a/zhenxun/builtin_plugins/auto_update/_data_source.py b/zhenxun/builtin_plugins/auto_update/_data_source.py index 18f80f38..94d518f6 100644 --- a/zhenxun/builtin_plugins/auto_update/_data_source.py +++ b/zhenxun/builtin_plugins/auto_update/_data_source.py @@ -39,7 +39,7 @@ class UpdateManager: bot_cur_version = cls.__get_version() release_task = ZhenxunRepoManager.zhenxun_get_latest_releases_data() - dev_version_task = RepoFileManager.get_file_content( + dev_version_task = RepoFileManager.get_text_content( ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "__version__" ) bot_commit_date_task = cls._get_latest_commit_date( @@ -108,7 +108,7 @@ class UpdateManager: res_latest_version = "获取失败" try: - res_latest_version_text = await RepoFileManager.get_file_content( + res_latest_version_text = await RepoFileManager.get_text_content( ZhenxunRepoConfig.RESOURCE_GITHUB_URL, "__version__" ) res_latest_version = res_latest_version_text.split(":")[-1].strip() @@ -264,7 +264,7 @@ class UpdateManager: resource_warning = "" if version_type == "main": try: - spec_content = await RepoFileManager.get_file_content( + spec_content = await RepoFileManager.get_text_content( ZhenxunRepoConfig.ZHENXUN_BOT_GITHUB_URL, "resources.spec" ) required_spec_str = None diff --git a/zhenxun/builtin_plugins/plugin_store/data_source.py b/zhenxun/builtin_plugins/plugin_store/data_source.py index 7ae89805..4278fabe 100644 --- a/zhenxun/builtin_plugins/plugin_store/data_source.py +++ b/zhenxun/builtin_plugins/plugin_store/data_source.py @@ -76,10 +76,10 @@ class StoreManager: if cached_data := await _PLUGIN_STORE_DATA_CACHE.get(cache_key): return cached_data - plugins = await RepoFileManager.get_file_content( + plugins = await RepoFileManager.get_text_content( DEFAULT_GITHUB_URL, "plugins.json" ) - extra_plugins = await RepoFileManager.get_file_content( + extra_plugins = await RepoFileManager.get_text_content( EXTRA_GITHUB_URL, "plugins.json", "index" ) result = ( @@ -337,9 +337,9 @@ class StoreManager: source: 源 """ repo_type = RepoType.GITHUB if is_external else None - if ( - source != "ali" and source != "git" and plugin_info.ali_url - ) or source == "ali": + if not is_external: + repo_type = RepoType.ALIYUN + elif (source is None and plugin_info.ali_url) or source == "ali": repo_type = RepoType.ALIYUN elif source == "git": repo_type = RepoType.GITHUB @@ -370,11 +370,17 @@ class StoreManager: dst_path = target_dir / f"{plugin_module}.py" download_files.append((src_path, dst_path)) + rand = random.randint(1, 10000) + requirement_path_ = TEMP_PATH / f"plugin_store_{rand}_req.txt" + requirements_path_ = TEMP_PATH / f"plugin_store_{rand}_reqs.txt" + download_files.append(("requirement.txt", requirement_path_)) + download_files.append(("requirements.txt", requirements_path_)) result = await RepoFileManager.download_files( github_url, download_files, branch, repo_type=repo_type, + ignore_error=True, ) if not result.success: raise PluginStoreException(result.error_message) @@ -396,32 +402,18 @@ class StoreManager: await VirtualEnvPackageManager.install_requirement(requirement_file) if not is_install_req: - # 从仓库根目录查找文件 - rand = random.randint(1, 10000) - requirement_path = TEMP_PATH / f"plugin_store_{rand}_req.txt" - requirements_path = TEMP_PATH / f"plugin_store_{rand}_reqs.txt" - await RepoFileManager.download_files( - github_url, - [ - ("requirement.txt", requirement_path), - ("requirements.txt", requirements_path), - ], - branch, - repo_type=repo_type, - ignore_error=True, - ) - if requirement_path.exists(): + if requirement_path_.exists(): logger.info( - f"开始安装插件 {module_path} 依赖文件: {requirement_path}", + f"开始安装插件 {module_path} 依赖文件: {requirement_path_}", LOG_COMMAND, ) - await VirtualEnvPackageManager.install_requirement(requirement_path) - if requirements_path.exists(): + await VirtualEnvPackageManager.install_requirement(requirement_path_) + if requirements_path_.exists(): logger.info( - f"开始安装插件 {module_path} 依赖文件: {requirements_path}", + f"开始安装插件 {module_path} 依赖文件: {requirements_path_}", LOG_COMMAND, ) - await VirtualEnvPackageManager.install_requirement(requirements_path) + await VirtualEnvPackageManager.install_requirement(requirements_path_) @classmethod async def remove_plugin(cls, index_or_module: str) -> str: diff --git a/zhenxun/utils/github_utils/models.py b/zhenxun/utils/github_utils/models.py index 9c5c3fef..df31e1ab 100644 --- a/zhenxun/utils/github_utils/models.py +++ b/zhenxun/utils/github_utils/models.py @@ -536,15 +536,7 @@ class AliyunFileInfo: ).decode(), ) - runtime = util_models.RuntimeOptions() - headers = {} - - response = await client.get_file_blobs_with_options_async( - repository_id, - request, - headers, - runtime, - ) + response = await client.get_file_blobs_async(repository_id, request) if response and response.body and response.body.result: if not response.body.success: diff --git a/zhenxun/utils/repo_utils/file_manager.py b/zhenxun/utils/repo_utils/file_manager.py index 18693bb3..36087382 100644 --- a/zhenxun/utils/repo_utils/file_manager.py +++ b/zhenxun/utils/repo_utils/file_manager.py @@ -4,16 +4,16 @@ import contextlib from pathlib import Path -from typing import cast, overload +import shutil +import tempfile +from typing import overload -import aiofiles from httpx import Response from zhenxun.services.log import logger from zhenxun.utils.github_utils import GithubUtils 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 from .config import LOG_COMMAND, RepoConfig from .exceptions import ( @@ -40,12 +40,12 @@ class RepoFileManager: self.config.ensure_dirs() @overload - async def get_github_file_content( + async def get_github_text_content( self, url: str, file_path: str, branch: str = "main", ignore_error: bool = False ) -> str: ... @overload - async def get_github_file_content( + async def get_github_text_content( self, url: str, file_path: list[str], @@ -53,7 +53,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> list[tuple[str, str]]: ... - async def get_github_file_content( + async def get_github_text_content( self, url: str, file_path: str | list[str], @@ -61,7 +61,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> str | list[tuple[str, str]]: """ - 获取GitHub仓库文件内容 + 获取GitHub仓库文本文件内容 参数: url: 仓库URL @@ -71,7 +71,7 @@ class RepoFileManager: 返回: list[tuple[str, str]]: 文件路径，文件内容 """ - results = [] + results: list[tuple[str, str]] = [] is_str_input = isinstance(file_path, str) try: if is_str_input: @@ -91,20 +91,9 @@ class RepoFileManager: ) if response.status_code == 200: logger.info(f"获取github文件内容成功: {f}", LOG_COMMAND) - text_content = response.content - # 确保使用UTF-8编码解析响应内容 - if not is_binary_file(f): - try: - text_content = response.content.decode("utf-8") - except UnicodeDecodeError: - # 如果UTF-8解码失败，尝试其他编码 - text_content = response.content.decode( - "utf-8", errors="ignore" - ) - logger.warning( - f"解码文件内容时出现错误，使用忽略错误模式:{f}", - LOG_COMMAND, - ) + text_content = response.content.decode( + "utf-8", errors="ignore" + ) results.append((f, text_content)) break else: @@ -124,7 +113,7 @@ class RepoFileManager: return results[0][1] if is_str_input and results else results @overload - async def get_aliyun_file_content( + async def get_aliyun_text_content( self, repo_name: str, file_path: str, @@ -133,7 +122,7 @@ class RepoFileManager: ) -> str: ... @overload - async def get_aliyun_file_content( + async def get_aliyun_text_content( self, repo_name: str, file_path: list[str], @@ -141,7 +130,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> list[tuple[str, str]]: ... - async def get_aliyun_file_content( + async def get_aliyun_text_content( self, repo_name: str, file_path: str | list[str], @@ -149,7 +138,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> str | list[tuple[str, str]]: """ - 获取阿里云CodeUp仓库文件内容 + 获取阿里云CodeUp仓库文本文件内容 参数: repo: 仓库名称 @@ -183,7 +172,7 @@ class RepoFileManager: return results[0][1] if is_str_input and results else results @overload - async def get_file_content( + async def get_text_content( self, repo_url: str, file_path: str, @@ -193,7 +182,7 @@ class RepoFileManager: ) -> str: ... @overload - async def get_file_content( + async def get_text_content( self, repo_url: str, file_path: list[str], @@ -202,7 +191,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> list[tuple[str, str]]: ... - async def get_file_content( + async def get_text_content( self, repo_url: str, file_path: str | list[str], @@ -211,7 +200,7 @@ class RepoFileManager: ignore_error: bool = False, ) -> str | list[tuple[str, str]]: """ - 获取仓库文件内容 + 获取仓库文本文件内容 参数: repo_url: 仓库URL @@ -229,22 +218,22 @@ class RepoFileManager: ) if repo_type is None: try: - return await self.get_aliyun_file_content( + return await self.get_aliyun_text_content( repo_name, file_path, branch, ignore_error ) except Exception: - return await self.get_github_file_content( + return await self.get_github_text_content( repo_url, file_path, branch, ignore_error ) try: if repo_type == RepoType.GITHUB: - return await self.get_github_file_content( + return await self.get_github_text_content( repo_url, file_path, branch, ignore_error ) elif repo_type == RepoType.ALIYUN: - return await self.get_aliyun_file_content( + return await self.get_aliyun_text_content( repo_name, file_path, branch, ignore_error ) @@ -262,24 +251,29 @@ class RepoFileManager: recursive: bool = True, ) -> list[RepoFileInfo]: """ - 获取仓库目录下的所有文件路径 + 获取仓库目录下的所有文件路径。 参数: - repo_url: 仓库URL + repo_url: 仓库URL，可以包含 /tree/，会自动解析出分支和仓库地址。 directory_path: 目录路径，默认为仓库根目录 - branch: 分支名称 + branch: 分支名称（若 repo_url 中包含 /tree/，则以 URL 中的为准） repo_type: 仓库类型，如果为None则自动判断 recursive: 是否递归获取子目录文件 返回: list[RepoFileInfo]: 文件信息列表 """ - repo_name = ( - repo_url.split("/tree/")[0].split("/")[-1].replace(".git", "").strip() - ) + base_url = repo_url + if "/tree/" in repo_url: + base_url, tree_part = repo_url.split("/tree/", maxsplit=1) + if tree_branch := tree_part.split("/", maxsplit=1)[0].strip(): + branch = tree_branch + + repo_name = base_url.split("/")[-1].replace(".git", "").strip() + try: if repo_type is None: - # 尝试GitHub，失败则尝试阿里云 + # 尝试阿里云，失败则尝试 GitHub try: return await self._list_aliyun_directory_files( repo_name, directory_path, branch, recursive @@ -289,11 +283,11 @@ class RepoFileManager: "获取阿里云目录文件失败，尝试GitHub", LOG_COMMAND, e=e ) return await self._list_github_directory_files( - repo_url, directory_path, branch, recursive + base_url, directory_path, branch, recursive ) if repo_type == RepoType.GITHUB: return await self._list_github_directory_files( - repo_url, directory_path, branch, recursive + base_url, directory_path, branch, recursive ) elif repo_type == RepoType.ALIYUN: return await self._list_aliyun_directory_files( @@ -303,7 +297,7 @@ class RepoFileManager: logger.error(f"获取目录文件列表失败: {directory_path}", LOG_COMMAND, e=e) if isinstance(e, FileNotFoundError | NetworkError | RepoManagerError): raise - raise RepoManagerError(f"获取目录文件列表失败: {e}") + raise RepoManagerError(f"获取目录文件列表失败: {e}") from e async def _list_github_directory_files( self, @@ -479,35 +473,6 @@ class RepoFileManager: if all(f.path != file.path for f in file_list if f != file) ] - def _clean_requirements_content(self, content: str) -> str: - """ - 清理 requirements.txt 内容，移除包含非ASCII字符的注释行 - - 这是为了防止 Windows 上 pip 使用 GBK 编码读取 UTF-8 文件时出错 - - 参数: - content: requirements.txt 文件内容 - - 返回: - str: 清理后的内容 - """ - lines = content.splitlines() - cleaned_lines = [] - for line in lines: - stripped = line.strip() - # 跳过空行 - if not stripped: - continue - # 如果是注释行且包含非ASCII字符，跳过 - if stripped.startswith("#"): - try: - stripped.encode("ascii") - except UnicodeEncodeError: - # 包含非ASCII字符的注释行，跳过 - continue - cleaned_lines.append(line) - return "\n".join(cleaned_lines) + "\n" if cleaned_lines else "" - async def download_files( self, repo_url: str, @@ -515,168 +480,108 @@ class RepoFileManager: branch: str = "main", repo_type: RepoType | None = None, ignore_error: bool = False, - sparse_path: str | None = None, - target_dir: Path | None = None, ) -> FileDownloadResult: """ - 下载多个文件 + 使用 Git 稀疏检出下载仓库中的文件或目录 参数: repo_url: 仓库URL - file_path: 文件在仓库中的路径，本地存储路径 + file_path: 仓库路径与本地目标路径的映射 branch: 分支名称 - repo_type: 仓库类型，如果为None则自动判断 - ignore_error: 是否忽略错误 - sparse_path: 稀疏检出路径 - target_dir: 稀疏目标目录 + repo_type: 仓库类型，阿里云类型会自动转换 CodeUp 地址 + ignore_error: 是否忽略不存在的仓库路径 返回: FileDownloadResult: 下载结果 """ + file_paths = [file_path] if isinstance(file_path, tuple) else file_path + if not file_paths: + raise RepoManagerError("参数错误: file_path 不能为空") + if any(not sparse_path.strip() for sparse_path, _ in file_paths): + raise RepoManagerError("参数错误: 仓库路径不能为空") - # 参数一致性校验：sparse_path 与 target_dir 必须同时有值或同时为 None - if (sparse_path is None) ^ (target_dir is None): - raise RepoManagerError( - "参数错误: sparse_path 与 target_dir 必须同时提供或同时为 None" - ) - - # 确定仓库类型和所有者 repo_name = ( repo_url.split("/tree/")[0].split("/")[-1].replace(".git", "").strip() ) - - if isinstance(file_path, tuple): - file_path = [file_path] - - file_path_mapping = {f[0]: f[1] for f in file_path} - - # 创建结果对象 result = FileDownloadResult( repo_type=repo_type, repo_name=repo_name, - file_path=file_path, + file_path=file_paths, version=branch, ) - if ( - any(is_binary_file(file_name) for file_name in file_path_mapping) - and repo_type != RepoType.GITHUB - and sparse_path - and target_dir - ): - return await self._handle_binary_with_sparse_checkout( - repo_url=repo_url, - branch=branch, - sparse_path=sparse_path, - target_dir=target_dir, - result=result, - ) - else: - # 不包含二进制时 - return await self._download_and_write_files( - repo_url=repo_url, - file_paths=[f[0] for f in file_path], - file_path_mapping=file_path_mapping, - branch=branch, - repo_type=repo_type, - ignore_error=ignore_error, - result=result, - ) + return await self._handle_with_sparse_checkout( + repo_url=repo_url, + branch=branch, + file_paths=file_paths, + repo_type=repo_type, + ignore_error=ignore_error, + result=result, + ) - async def _download_and_write_files( + async def _handle_with_sparse_checkout( self, repo_url: str, - file_paths: list[str], - file_path_mapping: dict[str, Path], branch: str, + file_paths: list[tuple[str, Path]], repo_type: RepoType | None, ignore_error: bool, result: FileDownloadResult, ) -> FileDownloadResult: try: - if len(file_paths) == 1: - file_contents_result = await self.get_file_content( - repo_url, file_paths[0], branch, repo_type, ignore_error - ) - if isinstance(file_contents_result, tuple): - file_contents = [file_contents_result] - elif isinstance(file_contents_result, str): - file_contents = [(file_paths[0], file_contents_result)] - else: - file_contents = cast(list[tuple[str, str]], file_contents_result) - else: - file_contents = cast( - list[tuple[str, str]], - await self.get_file_content( - repo_url, file_paths, branch, repo_type, ignore_error - ), + clone_url = repo_url.split("/tree/", maxsplit=1)[0].rstrip("/") + if not clone_url.endswith(".git"): + clone_url += ".git" + if repo_type == RepoType.ALIYUN: + repo_name = clone_url.rsplit("/", maxsplit=1)[-1].removesuffix(".git") + group_name = await get_aliyun_group_for_repo(repo_name) + clone_url = prepare_aliyun_url(clone_url, group_name) + + file_path_mapping = dict(file_paths) + with tempfile.TemporaryDirectory( + prefix="repo_sparse_", dir=self.config.cache_dir + ) as temp_dir: + staging_dir = Path(temp_dir) + downloaded_paths = await sparse_checkout_clone( + repo_url=clone_url, + branch=branch, + sparse_path=list(file_path_mapping), + target_dir=staging_dir, ) + missing_paths = set(file_path_mapping) - set(downloaded_paths) + if missing_paths and not ignore_error: + missing = ", ".join(sorted(missing_paths)) + raise RuntimeError(f"稀疏检出路径不存在: {missing}") - for repo_file_path, content in file_contents: - local_path = file_path_mapping[repo_file_path] - local_path.parent.mkdir(parents=True, exist_ok=True) - if isinstance(content, str): - # 对 requirements 文件特殊处理：移除包含非ASCII字符的注释行 - # 防止 Windows GBK 编码问题 - if repo_file_path.endswith(("requirements.txt", "requirement.txt")): - content = self._clean_requirements_content(content) - content_bytes = content.encode("utf-8") - else: - content_bytes = content - logger.debug(f"写入文件: {local_path}") - async with aiofiles.open(local_path, "wb") as f: - await f.write(content_bytes) - result.success = True - result.file_size = sum( - len(content.encode("utf-8") if isinstance(content, str) else content) - for _, content in file_contents - ) - logger.info(f"下载文件成功: {[f[0] for f in file_contents]}") - return result - except Exception as e: - logger.error(f"下载文件失败: {e}") - result.success = False - result.error_message = str(e) - return result + for sparse_path in downloaded_paths: + source_path = staging_dir / sparse_path + target_path = file_path_mapping[sparse_path] + target_path.parent.mkdir(parents=True, exist_ok=True) + if target_path.exists(): + if target_path.is_dir(): + shutil.rmtree(target_path) + else: + target_path.unlink() + shutil.move(str(source_path), str(target_path)) - async def _handle_binary_with_sparse_checkout( - self, - repo_url: str, - branch: str, - sparse_path: str, - target_dir: Path, - result: FileDownloadResult, - ) -> FileDownloadResult: - try: - # 获取仓库所属的分组名（外部插件仓库可能在不同分组下） - repo_name = ( - repo_url.split("/tree/")[0].split("/")[-1].replace(".git", "").strip() - ) - group_name = await get_aliyun_group_for_repo(repo_name) - - aliyun_repo_url = prepare_aliyun_url(repo_url, group_name) - - await sparse_checkout_clone( - repo_url=aliyun_repo_url, - branch=branch, - sparse_path=sparse_path, - target_dir=target_dir, - ) total_size = 0 - if target_dir.exists(): - for f in target_dir.rglob("*"): - if f.is_file(): - with contextlib.suppress(Exception): - total_size += f.stat().st_size + for sparse_path in downloaded_paths: + downloaded_path = file_path_mapping[sparse_path] + if downloaded_path.is_file(): + total_size += downloaded_path.stat().st_size + elif downloaded_path.is_dir(): + for file in downloaded_path.rglob("*"): + if file.is_file(): + with contextlib.suppress(Exception): + total_size += file.stat().st_size result.success = True result.file_size = total_size - logger.info(f"sparse-checkout 克隆成功: {target_dir}:{aliyun_repo_url}") + logger.info(f"sparse-checkout 下载成功: {downloaded_paths}") return result except GitUnavailableError as e: logger.error(f"Git不可用: {e}") result.success = False result.error_message = ( - "当前插件包含二进制文件，因ali限制需要使用git，" - "当前Git不可用，请尝试添加参数 -s git 或 安装 git" + "下载仓库文件需要使用 Git，当前 Git 不可用，请安装 Git 后重试" ) return result except Exception as e: diff --git a/zhenxun/utils/repo_utils/utils.py b/zhenxun/utils/repo_utils/utils.py index 5461d366..4205d6d5 100644 --- a/zhenxun/utils/repo_utils/utils.py +++ b/zhenxun/utils/repo_utils/utils.py @@ -48,20 +48,20 @@ async def clean_git(cwd: Path): async def run_git_command( - command: str, cwd: Path | None = None + command: str | list[str], cwd: Path | None = None ) -> tuple[bool, str, str]: """ 运行git命令，实时输出 stderr 进度信息（如 git clone --progress）。 参数: - command: 命令 + command: 命令字符串或参数列表 cwd: 工作目录 返回: tuple[bool, str, str]: (是否成功, 标准输出, 标准错误) """ try: - args = command.split() + args = command.split() if isinstance(command, str) else list(command) process = await asyncio.create_subprocess_exec( "git", *args, @@ -190,17 +190,21 @@ def filter_files( async def sparse_checkout_clone( repo_url: str, branch: str, - sparse_path: str, + sparse_path: str | list[str], target_dir: Path, -) -> None: +) -> list[str]: """ 使用 git 稀疏检出克隆指定路径到目标目录（在临时目录中操作）。 关键保障: - 在临时目录中执行所有 git 操作，避免影响 target_dir 中的现有内容 - 只操作 target_dir/sparse_path 路径，不影响 target_dir 其他内容 + + 返回: + list[str]: 成功检出的路径 """ target_dir.mkdir(parents=True, exist_ok=True) + sparse_paths = [sparse_path] if isinstance(sparse_path, str) else list(sparse_path) if not await check_git(): raise GitUnavailableError() @@ -214,7 +218,7 @@ async def sparse_checkout_clone( if not success: raise RuntimeError(f"git init 失败: {err or out}") success, out, err = await run_git_command( - f"remote add origin {repo_url}", temp_path + ["remote", "add", "origin", repo_url], temp_path ) if not success: raise RuntimeError(f"添加远程失败: {err or out}") @@ -224,55 +228,59 @@ async def sparse_checkout_clone( await run_git_command("sparse-checkout init --no-cone", temp_path) # 设置需要检出的路径（每次都覆盖配置） - if not sparse_path: + if not sparse_paths or any(not path for path in sparse_paths): raise RuntimeError("sparse-checkout 路径不能为空") # 使用 --no-cone 模式，直接指定要检出的具体路径 success, out, err = await run_git_command( - f"sparse-checkout set {sparse_path}/", temp_path + ["sparse-checkout", "set", "--", *sparse_paths], temp_path ) if not success: raise RuntimeError(f"配置稀疏路径失败: {err or out}") # 强制拉取并同步到远端 success, out, err = await run_git_command( - f"fetch --force --depth 1 origin {branch}", temp_path + ["fetch", "--force", "--depth", "1", "origin", branch], temp_path ) if not success: raise RuntimeError(f"fetch 失败: {err or out}") # 使用远端强制更新本地分支并覆盖工作区 success, out, err = await run_git_command( - f"checkout -B {branch} origin/{branch}", temp_path + ["checkout", "-B", branch, f"origin/{branch}"], temp_path ) if not success: # 回退方案 success2, out2, err2 = await run_git_command( - f"checkout {branch}", temp_path + ["checkout", branch], temp_path ) if not success2: raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}") # 强制对齐工作区 - await run_git_command(f"reset --hard origin/{branch}", temp_path) + await run_git_command(["reset", "--hard", f"origin/{branch}"], temp_path) await run_git_command("clean -xdf", temp_path) - # 将检出的文件移动到目标位置 - source_path = temp_path / sparse_path - if source_path.exists(): - # 确保目标路径存在 - target_path = target_dir / sparse_path - target_path.parent.mkdir(parents=True, exist_ok=True) + # 将成功检出的文件复制到 staging，缺失路径由调用方决定是否忽略 + downloaded_paths = [] + for path in sparse_paths: + source_path = temp_path / path + if not source_path.exists(): + continue - # 如果目标路径已存在，先清理 + target_path = target_dir / path + target_path.parent.mkdir(parents=True, exist_ok=True) if target_path.exists(): if target_path.is_dir(): shutil.rmtree(target_path) else: target_path.unlink() - - # 移动整个目录结构到目标位置 - shutil.move(str(source_path), str(target_path)) + if source_path.is_dir(): + shutil.copytree(source_path, target_path) + else: + shutil.copy2(source_path, target_path) + downloaded_paths.append(path) + return downloaded_paths def prepare_aliyun_url(repo_url: str, group_name: str | None = None) -> str: diff --git a/zhenxun/utils/utils.py b/zhenxun/utils/utils.py index df045972..5750748d 100644 --- a/zhenxun/utils/utils.py +++ b/zhenxun/utils/utils.py @@ -67,42 +67,6 @@ class ResourceDirManager: cls.__tree_append(path, deep) -def is_binary_file(file_path: str) -> bool: - """判断是否为二进制文件 - - 参数: - file_path: 文件路径 - - 返回: - bool: 是否为二进制文件 - """ - # fmt: off - # 精简但包含图片和字体的二进制文件扩展名集合 - BINARY_EXTENSIONS = frozenset({ - # 图片文件 - "jpg", "jpeg", "png", "gif", "bmp", "ico", "webp", "tiff", "tif", "svg", - # 字体文件 - "ttf", "otf", "woff", "woff2", "eot", - # 压缩文件 - "zip", "rar", "7z", "tar", "gz", "bz2", "xz", - # 可执行文件和库 - "exe", "dll", "so", "dylib", - # 文档文件 - "pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx", - # 多媒体文件 - "mp3", "mp4", "avi", "mov", "wmv", "flv", - # 其他常见二进制文件 - "bin", "dat", "db", "class", "pyc" - }) - - # 使用os.path.splitext高效提取扩展名 - _, ext = os.path.splitext(file_path) - # 去除点号并转换为小写 - ext_clean = ext.lstrip(".").lower() - - return ext_clean in BINARY_EXTENSIONS - - def cn2py(word: str) -> str: """将字符串转化为拼音