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" ) # Convenience aliases using the filesystem sandbox _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) # --------------------------------------------------------------------------- # File_System-specific operations (write, mkdir, move, copy, delete) # --------------------------------------------------------------------------- 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)}) # --------------------------------------------------------------------------- # Read-only operations delegated to sandbox # --------------------------------------------------------------------------- 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"}: # Prevent attempts to write to root or any directory 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: # info 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=( "
Browse, search, and interact with a filesystem. " "Choose an action and fill optional fields as needed." "
" ), api_description=TOOL_SUMMARY, flagging_mode="never", submit_btn="Run", ) __all__ = ["File_System", "build_interface"]