Modules/File_System.py

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=(
            "<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"]