core/tools/file: copy, move, and delete paths via shutil

Workspace-bound copy_path/move_path/delete_path for files and
directories; overwrite and recursive flags where needed.
This commit is contained in:
2026-07-14 21:47:57 +08:00
parent 25ff849efe
commit f26e13c2c1
5 changed files with 272 additions and 2 deletions
+3
View File
@@ -1,6 +1,9 @@
from .file import FILE_TOOLS as FILE_TOOLS
from .file import copy_path as copy_path
from .file import delete_path as delete_path
from .file import edit_replace as edit_replace
from .file import listdir as listdir
from .file import move_path as move_path
from .file import read_file as read_file
from .file import tree as tree
from .file import write_file as write_file
+13 -1
View File
@@ -1,7 +1,19 @@
from .edit_replace import edit_replace as edit_replace
from .fs_ops import copy_path as copy_path
from .fs_ops import delete_path as delete_path
from .fs_ops import move_path as move_path
from .listdir import listdir as listdir
from .read import read_file as read_file
from .tree import tree as tree
from .write import write_file as write_file
FILE_TOOLS = [read_file, write_file, listdir, tree, edit_replace]
FILE_TOOLS = [
read_file,
write_file,
listdir,
tree,
edit_replace,
copy_path,
move_path,
delete_path,
]
+156
View File
@@ -0,0 +1,156 @@
from __future__ import annotations
import shutil
from pathlib import Path
from plyngent.agent import tool
from plyngent.tools.workspace import WorkspaceError, get_workspace_root, resolve_path
def _kind(path: Path) -> str:
if path.is_dir():
return "directory"
if path.is_file():
return "file"
if path.is_symlink():
return "symlink"
return "path"
def _remove_existing(path: Path) -> None:
if path.is_dir() and not path.is_symlink():
shutil.rmtree(path)
elif path.exists() or path.is_symlink():
path.unlink()
def _resolve_pair(src: str, dst: str) -> tuple[Path, Path, Path] | str:
try:
source = resolve_path(src)
dest = resolve_path(dst)
root = get_workspace_root()
except WorkspaceError as exc:
return f"error: {exc}"
return source, dest, root
def _prepare_dest(source: Path, dest: Path, root: Path, *, overwrite: bool, dst_label: str) -> Path | str:
"""Resolve copy/move destination; return error string on conflict."""
if source.is_file() and dest.is_dir():
dest = resolve_path(str((dest / source.name).relative_to(root)))
if dest.exists() or dest.is_symlink():
if not overwrite:
return f"error: destination exists: {dst_label} (set overwrite=true)"
_remove_existing(dest)
dest.parent.mkdir(parents=True, exist_ok=True)
return dest
def _copy_or_move_validated( # noqa: PLR0913
source: Path,
dest: Path,
root: Path,
*,
src: str,
dst: str,
overwrite: bool,
action: str,
) -> str:
prepared = _prepare_dest(source, dest, root, overwrite=overwrite, dst_label=dst)
if isinstance(prepared, str):
return prepared
dest = prepared
try:
if action == "copy":
if source.is_dir() and not source.is_symlink():
_ = shutil.copytree(source, dest, symlinks=True)
label = "directory"
else:
_ = shutil.copy2(source, dest, follow_symlinks=False)
label = _kind(source)
return f"copied {label} {src} -> {dest.relative_to(root)}"
moved = Path(shutil.move(str(source), str(dest))).resolve()
return f"moved {_kind(moved)} {src} -> {moved.relative_to(root)}"
except WorkspaceError as exc:
return f"error: {exc}"
except OSError as exc:
return f"error: {action} failed: {exc}"
@tool
def copy_path(src: str, dst: str, *, overwrite: bool = False) -> str:
"""Copy a file or directory under the workspace (``shutil.copy2`` / ``copytree``).
``dst`` parent directories are created as needed. If ``dst`` exists, set
``overwrite=true`` to replace it.
"""
pair = _resolve_pair(src, dst)
if isinstance(pair, str):
return pair
source, dest, root = pair
if not source.exists() and not source.is_symlink():
return f"error: source does not exist: {src}"
return _copy_or_move_validated(
source, dest, root, src=src, dst=dst, overwrite=overwrite, action="copy"
)
@tool
def move_path(src: str, dst: str, *, overwrite: bool = False) -> str:
"""Move/rename a file or directory under the workspace (``shutil.move``)."""
pair = _resolve_pair(src, dst)
if isinstance(pair, str):
return pair
source, dest, root = pair
if not source.exists() and not source.is_symlink():
return f"error: source does not exist: {src}"
if source == root:
return "error: cannot move the workspace root"
return _copy_or_move_validated(
source, dest, root, src=src, dst=dst, overwrite=overwrite, action="move"
)
def _delete_directory(target: Path, path: str, *, recursive: bool) -> str:
if recursive:
shutil.rmtree(target)
return f"deleted directory {path} (recursive)"
try:
_ = next(target.iterdir())
except StopIteration:
target.rmdir()
return f"deleted empty directory {path}"
return f"error: directory not empty: {path} (set recursive=true)"
def _delete_target(target: Path, path: str, *, recursive: bool) -> str:
if target.is_symlink() or target.is_file():
kind = _kind(target)
target.unlink()
return f"deleted {kind} {path}"
if target.is_dir():
return _delete_directory(target, path, recursive=recursive)
return f"error: unsupported path type: {path}"
@tool
def delete_path(path: str, *, recursive: bool = False) -> str:
"""Delete a file or directory under the workspace.
Files and empty directories are removed always. Non-empty directories require
``recursive=true`` (uses ``shutil.rmtree``). The workspace root cannot be deleted.
"""
try:
target = resolve_path(path)
root = get_workspace_root()
except WorkspaceError as exc:
return f"error: {exc}"
if not target.exists() and not target.is_symlink():
return f"error: path does not exist: {path}"
if target == root:
return "error: cannot delete the workspace root"
try:
return _delete_target(target, path, recursive=recursive)
except OSError as exc:
return f"error: delete failed: {exc}"