| from __future__ import annotations |
|
|
| import json |
| import os |
| import shutil |
| from typing import Annotated, Optional |
|
|
| import gradio as gr |
|
|
| from app import _log_call_end, _log_call_start, _truncate_for_log |
| from ._docstrings import autodoc |
| from ._core import ( |
| filesystem_sandbox, |
| ROOT_DIR, |
| ALLOW_ABS, |
| safe_open, |
| _fmt_size, |
| ) |
|
|
|
|
| TOOL_SUMMARY = ( |
| "Browse, search, and manage files within a safe root. " |
| "Actions: list, read, write, append, mkdir, move, copy, delete, info, search, help. " |
| "Fill other fields as needed. " |
| "Use paths like `/` or `/notes/todo.txt` because all paths are relative to the root (`/`). " |
| "Use 'help' to see action-specific required fields and examples." |
| ) |
|
|
| HELP_TEXT = ( |
| "File System — actions and usage\n\n" |
| "Root: paths resolve under Nymbo-Tools/Filesystem by default (or NYMBO_TOOLS_ROOT if set). " |
| "Start paths with '/' to refer to the tool root (e.g., /notes). " |
| "Absolute paths are disabled unless UNSAFE_ALLOW_ABS_PATHS=1.\n\n" |
| "Actions and fields:\n" |
| "- list: path='/' (default), recursive=false, show_hidden=false, max_entries=20\n" |
| "- read: path (e.g., /notes/todo.txt), offset=0, max_chars=4000 (shows next_cursor when truncated)\n" |
| "- write: path, content (UTF-8), create_dirs=true\n" |
| "- append: path, content (UTF-8), create_dirs=true\n" |
| "- mkdir: path (directory), exist_ok=true\n" |
| "- move: path (src), dest_path (dst), overwrite=false\n" |
| "- copy: path (src), dest_path (dst), overwrite=false\n" |
| "- delete: path, recursive=true (required for directories)\n" |
| "- info: path\n" |
| "- search: path (dir or file), content=query text, recursive=false, show_hidden=false, max_entries=20, case_sensitive=false, offset=0\n" |
| "- help: show this guide\n\n" |
| "Errors are returned as JSON with fields: {status:'error', code, message, path?, hint?, data?}.\n\n" |
| "Examples:\n" |
| "- list current: action=list, path='/'\n" |
| "- make folder: action=mkdir, path='/notes'\n" |
| "- write file: action=write, path='/notes/todo.txt', content='hello'\n" |
| "- read file: action=read, path='/notes/todo.txt', max_chars=200\n" |
| "- move file: action=move, path='/notes/todo.txt', dest_path='/notes/todo-old.txt', overwrite=true\n" |
| "- delete dir: action=delete, path='/notes', recursive=true\n" |
| "- search text: action=search, path='/notes', content='TODO', recursive=true, max_entries=50\n" |
| "- page search results: action=search, content='TODO', offset=10\n" |
| "- case-sensitive search: action=search, content='TODO', case_sensitive=true\n" |
| ) |
|
|
|
|
| |
| _sandbox = filesystem_sandbox |
|
|
| def _resolve_path(path: str) -> tuple[str, str]: |
| """Resolve path using the filesystem sandbox.""" |
| return _sandbox.resolve_path(path) |
|
|
| def _display_path(abs_path: str) -> str: |
| """Display path using the filesystem sandbox.""" |
| return _sandbox.display_path(abs_path) |
|
|
| def _err(code: str, message: str, *, path: Optional[str] = None, hint: Optional[str] = None, data: Optional[dict] = None) -> str: |
| """Return a structured error JSON string.""" |
| return _sandbox.err(code, message, path=path, hint=hint, data=data) |
|
|
| def _safe_err(exc: Exception | str) -> str: |
| """Return an error string with root replaced.""" |
| return _sandbox.safe_err(exc) |
|
|
|
|
| |
| |
| |
|
|
| def _ensure_parent(abs_path: str, create_dirs: bool) -> None: |
| parent = os.path.dirname(abs_path) |
| if parent and not os.path.exists(parent): |
| if create_dirs: |
| os.makedirs(parent, exist_ok=True) |
| else: |
| raise FileNotFoundError(f"Parent directory does not exist: {_display_path(parent)}") |
|
|
|
|
| def _write_file(abs_path: str, content: str, *, append: bool, create_dirs: bool) -> str: |
| try: |
| _ensure_parent(abs_path, create_dirs) |
| mode = 'a' if append else 'w' |
| with open(abs_path, mode, encoding='utf-8') as f: |
| f.write(content or "") |
| return f"{'Appended to' if append else 'Wrote'} file: {_display_path(abs_path)} (chars={len(content or '')})" |
| except Exception as exc: |
| return _err("write_failed", "Failed to write file.", path=_display_path(abs_path), data={"error": _safe_err(exc)}) |
|
|
|
|
| def _mkdir(abs_path: str, exist_ok: bool) -> str: |
| try: |
| os.makedirs(abs_path, exist_ok=exist_ok) |
| return f"Created directory: {_display_path(abs_path)}" |
| except Exception as exc: |
| return _err("mkdir_failed", "Failed to create directory.", path=_display_path(abs_path), data={"error": _safe_err(exc)}) |
|
|
|
|
| def _move_copy(action: str, src: str, dst: str, *, overwrite: bool) -> str: |
| try: |
| if not os.path.exists(src): |
| return _err("source_not_found", f"Source not found: {_display_path(src)}", path=_display_path(src)) |
| if os.path.isdir(dst): |
| dst_path = os.path.join(dst, os.path.basename(src)) |
| else: |
| dst_path = dst |
| if os.path.exists(dst_path): |
| if overwrite: |
| if os.path.isdir(dst_path): |
| shutil.rmtree(dst_path) |
| else: |
| os.remove(dst_path) |
| else: |
| return _err( |
| "destination_exists", |
| f"Destination already exists: {_display_path(dst_path)}", |
| path=_display_path(dst_path), |
| hint="Set overwrite=True to replace the destination." |
| ) |
| if action == 'move': |
| shutil.move(src, dst_path) |
| else: |
| if os.path.isdir(src): |
| shutil.copytree(src, dst_path) |
| else: |
| shutil.copy2(src, dst_path) |
| return f"{action.capitalize()}d: {_display_path(src)} -> {_display_path(dst_path)}" |
| except Exception as exc: |
| return _err(f"{action}_failed", f"Failed to {action}.", path=_display_path(src), data={"error": _safe_err(exc), "destination": _display_path(dst)}) |
|
|
|
|
| def _delete(abs_path: str, *, recursive: bool) -> str: |
| try: |
| if not os.path.exists(abs_path): |
| return _err("path_not_found", f"Path not found: {_display_path(abs_path)}", path=_display_path(abs_path)) |
| if os.path.isdir(abs_path): |
| if not recursive: |
| return _err("requires_recursive", "Refusing to delete a directory without recursive=True", path=_display_path(abs_path), hint="Pass recursive=True to delete a directory.") |
| shutil.rmtree(abs_path) |
| else: |
| os.remove(abs_path) |
| return f"Deleted: {_display_path(abs_path)}" |
| except Exception as exc: |
| return _err("delete_failed", "Failed to delete path.", path=_display_path(abs_path), data={"error": _safe_err(exc)}) |
|
|
| |
| |
| |
|
|
| def _list_dir(abs_path: str, *, show_hidden: bool, recursive: bool, max_entries: int) -> str: |
| """List directory contents as a visual tree.""" |
| return _sandbox.list_dir(abs_path, show_hidden=show_hidden, recursive=recursive, max_entries=max_entries) |
|
|
|
|
| def _search_text( |
| abs_path: str, |
| query: str, |
| *, |
| recursive: bool, |
| show_hidden: bool, |
| max_results: int, |
| case_sensitive: bool, |
| start_index: int, |
| ) -> str: |
| """Search for text within files.""" |
| return _sandbox.search_text( |
| abs_path, |
| query, |
| recursive=recursive, |
| show_hidden=show_hidden, |
| max_results=max_results, |
| case_sensitive=case_sensitive, |
| start_index=start_index, |
| ) |
|
|
|
|
| def _read_file(abs_path: str, *, offset: int, max_chars: int) -> str: |
| """Read file contents with optional offset and character limit.""" |
| return _sandbox.read_file(abs_path, offset=offset, max_chars=max_chars) |
|
|
|
|
| def _info(abs_path: str) -> str: |
| """Get file/directory metadata as JSON.""" |
| return _sandbox.info(abs_path) |
|
|
|
|
|
|
| @autodoc(summary=TOOL_SUMMARY) |
| def File_System( |
| action: Annotated[str, "Operation to perform: 'list', 'read', 'write', 'append', 'mkdir', 'move', 'copy', 'delete', 'info', 'search'."], |
| path: Annotated[str, "Target path, relative to root unless UNSAFE_ALLOW_ABS_PATHS=1."] = "/", |
| content: Annotated[Optional[str], "Content for write/append actions or search query (UTF-8)."] = None, |
| dest_path: Annotated[Optional[str], "Destination for move/copy (relative to root unless unsafe absolute allowed)."] = None, |
| recursive: Annotated[bool, "For list/search (recurse into subfolders) and delete (required for directories)."] = False, |
| show_hidden: Annotated[bool, "Include hidden files (dotfiles) for list/search."] = False, |
| max_entries: Annotated[int, "Max entries to list or matches to return (for list/search)."] = 20, |
| offset: Annotated[int, "Start offset for reading files (for read)."] = 0, |
| max_chars: Annotated[int, "Max characters to return when reading (0 = full file)."] = 4000, |
| create_dirs: Annotated[bool, "Create parent directories for write/append if missing."] = True, |
| overwrite: Annotated[bool, "Allow overwrite for move/copy destinations."] = False, |
| case_sensitive: Annotated[bool, "Match case when searching text."] = False, |
| ) -> str: |
| _log_call_start( |
| "File_System", |
| action=action, |
| path=path, |
| dest_path=dest_path, |
| recursive=recursive, |
| show_hidden=show_hidden, |
| max_entries=max_entries, |
| offset=offset, |
| max_chars=max_chars, |
| create_dirs=create_dirs, |
| overwrite=overwrite, |
| case_sensitive=case_sensitive, |
| ) |
| action = (action or "").strip().lower() |
| if action not in {"list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "search", "help"}: |
| result = _err( |
| "invalid_action", |
| "Invalid action.", |
| hint="Choose from: list, read, write, append, mkdir, move, copy, delete, info, search, help." |
| ) |
| _log_call_end("File_System", _truncate_for_log(result)) |
| return result |
|
|
| abs_path, err = _resolve_path(path) |
| if err: |
| _log_call_end("File_System", _truncate_for_log(err)) |
| return err |
|
|
| try: |
| if action == "help": |
| result = HELP_TEXT |
| elif action == "list": |
| if not os.path.exists(abs_path): |
| result = _err("path_not_found", f"Path not found: {_display_path(abs_path)}", path=_display_path(abs_path)) |
| else: |
| result = _list_dir(abs_path, show_hidden=show_hidden, recursive=recursive, max_entries=max_entries) |
| elif action == "read": |
| result = _read_file(abs_path, offset=offset, max_chars=max_chars) |
| elif action in {"write", "append"}: |
| |
| if _display_path(abs_path) == "/" or os.path.isdir(abs_path): |
| result = _err( |
| "invalid_write_path", |
| "Invalid path for write/append.", |
| path=_display_path(abs_path), |
| hint="Provide a file path under / (e.g., /notes/todo.txt)." |
| ) |
| else: |
| result = _write_file(abs_path, content or "", append=(action == "append"), create_dirs=create_dirs) |
| elif action == "mkdir": |
| result = _mkdir(abs_path, exist_ok=True) |
| elif action in {"move", "copy"}: |
| if not dest_path: |
| result = _err("missing_dest_path", "dest_path is required for move/copy (ignored for other actions).") |
| else: |
| abs_dst, err2 = _resolve_path(dest_path) |
| if err2: |
| result = err2 |
| else: |
| result = _move_copy(action, abs_path, abs_dst, overwrite=overwrite) |
| elif action == "delete": |
| result = _delete(abs_path, recursive=recursive) |
| elif action == "search": |
| query_text = content or "" |
| if query_text.strip() == "": |
| result = _err( |
| "missing_search_query", |
| "Search query is required for the search action.", |
| hint="Provide text in the Content field to search for.", |
| ) |
| else: |
| result = _search_text( |
| abs_path, |
| query_text, |
| recursive=recursive, |
| show_hidden=show_hidden, |
| max_results=max_entries, |
| case_sensitive=case_sensitive, |
| start_index=offset, |
| ) |
| else: |
| result = _info(abs_path) |
| except Exception as exc: |
| result = _err("exception", "Unhandled error during operation.", data={"error": _safe_err(exc)}) |
|
|
| _log_call_end("File_System", _truncate_for_log(result)) |
| return result |
|
|
|
|
| def build_interface() -> gr.Interface: |
| return gr.Interface( |
| fn=File_System, |
| inputs=[ |
| gr.Radio( |
| label="Action", |
| choices=["list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "search", "help"], |
| value="help", |
| info="Operation to perform", |
| ), |
| gr.Textbox(label="Path", placeholder="/ or /src/file.txt", max_lines=1, value="/", info="Target path (relative to root)"), |
| gr.Textbox(label="Content", lines=6, placeholder="Text to write or search for...", info="Content for write/append actions or search query"), |
| gr.Textbox(label="Destination", max_lines=1, info="Destination path (Move/Copy only)"), |
| gr.Checkbox(label="Recursive", value=False, info="Recurse into subfolders (List/Delete/Search)"), |
| gr.Checkbox(label="Show hidden", value=False, info="Include hidden files (List/Search)"), |
| gr.Slider(minimum=10, maximum=5000, step=10, value=20, label="Max entries / matches", info="Max entries to list or matches to return (List/Search)"), |
| gr.Slider(minimum=0, maximum=1_000_000, step=100, value=0, label="Offset", info="Start offset (Read/Search)"), |
| gr.Slider(minimum=0, maximum=100_000, step=500, value=4000, label="Max chars", info="Max characters to return (Read, 0=all)"), |
| gr.Checkbox(label="Create parent dirs", value=True, info="Create parent directories if missing (Write)"), |
| gr.Checkbox(label="Overwrite destination", value=False, info="Allow overwrite (Move/Copy)"), |
| gr.Checkbox(label="Case sensitive search", value=False, info="Match case (Search)"), |
| ], |
| outputs=gr.Textbox(label="Result", lines=20), |
| title="File System", |
| description=( |
| "<div id=\"fs-desc\" style=\"text-align:center; overflow:hidden;\">Browse, search, and interact with a filesystem. " |
| "Choose an action and fill optional fields as needed." |
| "</div>" |
| ), |
| api_description=TOOL_SUMMARY, |
| flagging_mode="never", |
| submit_btn="Run", |
| ) |
|
|
|
|
| __all__ = ["File_System", "build_interface"] |
|
|
