zhenxun/builtin_plugins/plugin_store/data_source.py

@@ -209,7 +209,9 @@ class StoreManager:
         if is_remove:
             if plugin_info.module not in modules:
                 raise PluginStoreException(f"插件 {plugin_info.name} 未安装，无法移除")
-            if plugin_obj := await PluginInfo.get_plugin(module=plugin_info.module):
+            if plugin_obj := await PluginInfo.get_or_none(
+                name=plugin_info.name, module=plugin_info.module
+            ):
                 plugin_info.module_path = plugin_obj.module_path
             return plugin_info, is_external
 

zhenxun/utils/repo_utils/utils.py

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