fix: Improve directory deletion on Windows

- Implement custom rmtree function to handle Windows deletion errors
- Replace shutil.rmtree calls with new rmtree utility function
- Move __onerror handler to utils.py for centralized error handling

Author: bobo yang

devchat/_cli/run.py

@@ -1,6 +1,7 @@
 from typing import List, Optional, Tuple
 
 import click
+from devchat.utils import rmtree
 
 
 @click.command(
@@ -115,28 +116,6 @@ def run(
             return
 
 
-def __onerror(func, path, _1):
-    """
-    Error handler for shutil.rmtree.
-
-    If the error is due to an access error (read only file)
-    it attempts to add write permission and then retries.
-
-    If the error is for another reason it re-raises the error.
-
-    Usage : shutil.rmtree(path, onerror=onerror)
-    """
-    import os
-    import stat
-
-    # Check if file access issue
-    if not os.access(path, os.W_OK):
-        # Try to change the file to be writable (remove read-only flag)
-        os.chmod(path, stat.S_IWUSR)
-        # Retry the function that failed
-        func(path)
-
-
 def __make_files_writable(directory):
     """
     Recursively make all files in the directory writable.
@@ -180,9 +159,9 @@ def _clone_or_pull_git_repo(target_dir: str, repo_urls: List[Tuple[str, str]], z
         bak_dir = target_dir + "_bak"
         new_dir = target_dir + "_old"
         if os.path.exists(new_dir):
-            shutil.rmtree(new_dir, onerror=__onerror)
+            rmtree(new_dir)
         if os.path.exists(bak_dir):
-            shutil.rmtree(bak_dir, onerror=__onerror)
+            rmtree(bak_dir)
         print(f"{target_dir} is already exists. Moved to {new_dir}")
         clone_git_repo(bak_dir, repo_urls)
         try:

devchat/_cli/utils.py

@@ -6,7 +6,7 @@
 from typing import Any, List, Optional, Tuple
 
 from devchat._cli.errors import MissContentInPromptException
-from devchat.utils import add_gitignore, find_root_dir, get_logger, setup_logger
+from devchat.utils import add_gitignore, find_root_dir, get_logger, rmtree, setup_logger
 
 logger = get_logger(__name__)
 
@@ -31,7 +31,7 @@ def download_and_extract_workflow(workflow_url, target_dir):
 
     # Delete target directory if exists
     if os.path.exists(target_dir):
-        shutil.rmtree(target_dir)
+        rmtree(target_dir)
 
     # Rename extracted directory to target directory
     extracted_dir = os.path.join(parent_dir, "workflows-main")

devchat/_service/route/workflows.py

@@ -7,7 +7,7 @@
 from fastapi.responses import JSONResponse
 
 from devchat._service.schema import response
-from devchat.utils import get_logger
+from devchat.utils import get_logger, rmtree
 from devchat.workflow.namespace import (
     WorkflowMeta,
     get_prioritized_namespace_path,
@@ -100,7 +100,7 @@ def update_custom_workflows():
 
                     if repo_path.exists():
                         logger.info(f"Repo path not empty {repo_path}, removing it.")
-                        shutil.rmtree(repo_path)
+                        rmtree(repo_path)
 
                     logger.info(
                         f"Starting update for {repo_name} at {repo_path} "

devchat/utils.py

@@ -243,3 +243,29 @@ def get_encoding(name: str):
 def openai_response_tokens(message: dict, model: str) -> int:
     """Returns the number of tokens used by a response."""
     return openai_message_tokens(message, model)
+
+def rmtree(path: str) -> None:
+    import shutil
+
+    def __onerror(func, path, _1):
+        """
+        Error handler for shutil.rmtree.
+
+        If the error is due to an access error (read only file)
+        it attempts to add write permission and then retries.
+
+        If the error is for another reason it re-raises the error.
+
+        Usage : shutil.rmtree(path, onerror=onerror)
+        """
+        import os
+        import stat
+
+        # Check if file access issue
+        if not os.access(path, os.W_OK):
+            # Try to change the file to be writable (remove read-only flag)
+            os.chmod(path, stat.S_IWUSR)
+            # Retry the function that failed
+            func(path)
+
+    shutil.rmtree(path, onerror=__onerror)

devchat/workflow/update_util.py

@@ -8,7 +8,7 @@
 
 import requests
 
-from devchat.utils import get_logger
+from devchat.utils import get_logger, rmtree
 from devchat.workflow.path import (
     CHAT_DIR,
     CUSTOM_BASE,
@@ -167,7 +167,7 @@ def update_by_zip(workflow_base: Path) -> Tuple[bool, str]:
         # Has previous workflows, download as tmp_new
         tmp_new = parent / f"{WORKFLOWS_BASE_NAME}_new"
         if tmp_new.exists():
-            shutil.rmtree(tmp_new)
+            rmtree(tmp_new)
             # TODO: handle error?
             # shutil.rmtree(tmp_new, onerror=__onerror)
 
@@ -182,7 +182,7 @@ def update_by_zip(workflow_base: Path) -> Tuple[bool, str]:
         backup_zip = _backup(workflow_base)
 
         # rename the new dir to the workflow_base
-        shutil.rmtree(workflow_base)
+        rmtree(workflow_base)
         shutil.move(tmp_new, workflow_base)
 
         msg = f"Updated {workflow_base} by zip. (backup: {backup_zip})"
@@ -223,7 +223,7 @@ def update_by_git(workflow_base: Path) -> Tuple[bool, str]:
             # try to clone the new repo to tmp_new
             tmp_new = parent / f"{WORKFLOWS_BASE_NAME}_new"
             if tmp_new.exists():
-                shutil.rmtree(tmp_new)
+                rmtree(tmp_new)
 
             clone_ok = _clone_repo_to_dir(REPO_URLS, tmp_new)
             if not clone_ok:
@@ -235,7 +235,7 @@ def update_by_git(workflow_base: Path) -> Tuple[bool, str]:
             backup_zip = _backup(workflow_base)
 
             # rename the new dir to the workflow_base
-            shutil.rmtree(workflow_base)
+            rmtree(workflow_base)
             shutil.move(tmp_new, workflow_base)
 
             msg = f"Updated {workflow_base} by git. (backup: {backup_zip})"