refactor(zhenxun): 优化 sparse_checkout_clone 函数的实现

- 将 git 操作移至临时目录中执行,避免影响目标目录中的现有内容
- 简化了稀疏检出的配置和执行过程
- 改进了错误处理和回退逻辑
- 优化了文件移动和目录清理的操作
This commit is contained in:
molanp 2025-09-11 11:18:09 +08:00
parent 81e3fb68b0
commit 3df40ccb12

View File

@ -7,6 +7,7 @@ import base64
from pathlib import Path from pathlib import Path
import re import re
import shutil import shutil
import tempfile
from zhenxun.services.log import logger from zhenxun.services.log import logger
@ -145,80 +146,85 @@ async def sparse_checkout_clone(
target_dir: Path, target_dir: Path,
) -> None: ) -> None:
""" """
使用 git 稀疏检出克隆指定路径到目标目录完全独立于主项目 git 使用 git 稀疏检出克隆指定路径到目标目录在临时目录中操作
关键保障: 关键保障:
- target_dir 下检测/初始化 .git所有 git 操作均以 cwd=target_dir 执行 - 在临时目录中执行所有 git 操作避免影响 target_dir 中的现有内容
- 强制拉取与工作区覆盖: fetch --forcecheckout -Breset --hardclean -xdf - 只操作 target_dir/sparse_path 路径不影响 target_dir 其他内容
- 反复设置 sparse-checkout 路径确保路径更新生效
""" """
target_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True)
if not await check_git(): if not await check_git():
raise GitUnavailableError() raise GitUnavailableError()
git_dir = target_dir / ".git" # 在临时目录中进行 git 操作
if not git_dir.exists(): with tempfile.TemporaryDirectory() as temp_dir:
success, out, err = await run_git_command("init", target_dir) temp_path = Path(temp_dir)
# 初始化临时目录为 git 仓库
success, out, err = await run_git_command("init", temp_path)
if not success: if not success:
raise RuntimeError(f"git init 失败: {err or out}") raise RuntimeError(f"git init 失败: {err or out}")
success, out, err = await run_git_command( success, out, err = await run_git_command(
f"remote add origin {repo_url}", target_dir f"remote add origin {repo_url}", temp_path
) )
if not success: if not success:
raise RuntimeError(f"添加远程失败: {err or out}") raise RuntimeError(f"添加远程失败: {err or out}")
else:
# 启用稀疏检出(使用 --no-cone 模式以获得更精确的控制)
await run_git_command("config core.sparseCheckout true", temp_path)
await run_git_command("sparse-checkout init --no-cone", temp_path)
# 设置需要检出的路径(每次都覆盖配置)
if not sparse_path:
raise RuntimeError("sparse-checkout 路径不能为空")
# 使用 --no-cone 模式,直接指定要检出的具体路径
success, out, err = await run_git_command( success, out, err = await run_git_command(
f"remote set-url origin {repo_url}", target_dir f"sparse-checkout set {sparse_path}/", temp_path
) )
if not success: if not success:
# 兜底尝试添加 raise RuntimeError(f"配置稀疏路径失败: {err or out}")
await run_git_command(f"remote add origin {repo_url}", target_dir)
# 启用稀疏检出(使用 --no-cone 模式以获得更精确的控制) # 强制拉取并同步到远端
await run_git_command("config core.sparseCheckout true", target_dir) success, out, err = await run_git_command(
await run_git_command("sparse-checkout init --no-cone", target_dir) f"fetch --force --depth 1 origin {branch}", temp_path
)
if not success:
raise RuntimeError(f"fetch 失败: {err or out}")
# 设置需要检出的路径(每次都覆盖配置) # 使用远端强制更新本地分支并覆盖工作区
if not sparse_path: success, out, err = await run_git_command(
raise RuntimeError("sparse-checkout 路径不能为空") f"checkout -B {branch} origin/{branch}", temp_path
)
if not success:
# 回退方案
success2, out2, err2 = await run_git_command(
f"checkout {branch}", temp_path
)
if not success2:
raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
# 使用 --no-cone 模式,直接指定要检出的具体路径 # 强制对齐工作区
# 例如sparse_path="plugins/mahiro" -> 只检出 plugins/mahiro/ 下的内容 await run_git_command(f"reset --hard origin/{branch}", temp_path)
success, out, err = await run_git_command( await run_git_command("clean -xdf", temp_path)
f"sparse-checkout set {sparse_path}/", target_dir
)
if not success:
raise RuntimeError(f"配置稀疏路径失败: {err or out}")
# 强制拉取并同步到远端 # 将检出的文件移动到目标位置
success, out, err = await run_git_command( source_path = temp_path / sparse_path
f"fetch --force --depth 1 origin {branch}", target_dir if source_path.exists():
) # 确保目标路径存在
if not success: target_path = target_dir / sparse_path
raise RuntimeError(f"fetch 失败: {err or out}") target_path.parent.mkdir(parents=True, exist_ok=True)
# 使用远端强制更新本地分支并覆盖工作区 # 如果目标路径已存在,先清理
success, out, err = await run_git_command( if target_path.exists():
f"checkout -B {branch} origin/{branch}", target_dir if target_path.is_dir():
) shutil.rmtree(target_path)
if not success: else:
# 回退方案 target_path.unlink()
success2, out2, err2 = await run_git_command(f"checkout {branch}", target_dir)
if not success2:
raise RuntimeError(f"checkout 失败: {(err or out) or (err2 or out2)}")
# 强制对齐工作区 # 移动整个目录结构到目标位置
await run_git_command(f"reset --hard origin/{branch}", target_dir) shutil.move(str(source_path), str(target_path))
await run_git_command("clean -xdf", target_dir)
dir_path = target_dir / Path(sparse_path)
for f in dir_path.iterdir():
shutil.move(f, target_dir / f.name)
dir_name = sparse_path.split("/")[0]
rm_path = target_dir / dir_name
if rm_path.exists():
shutil.rmtree(rm_path)
def prepare_aliyun_url(repo_url: str) -> str: def prepare_aliyun_url(repo_url: str) -> str: