Modules/File_System.py

from __future__ import annotations

import difflib
import json
import os
import re
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, edit, 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"
    "- edit: path, content (SEARCH/REPLACE blocks, see format below)\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"
    "Edit format (SEARCH/REPLACE blocks):\n"
    "<<<<<<< SEARCH\n"
    "[exact content to find]\n"
    "=======\n"
    "[new content to replace with]\n"
    ">>>>>>> REPLACE\n\n"
    "Multiple blocks can be included; each is applied in order. "
    "Search text must match exactly (whitespace, indentation). "
    "Only the first occurrence of each search text is replaced.\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"
    "- edit file: action=edit, path='/notes/todo.txt', content='<<<<<<< SEARCH\\nhello\\n=======\\nhi\\n>>>>>>> REPLACE'\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)})


# ---------------------------------------------------------------------------
# Edit (search/replace) operation
# ---------------------------------------------------------------------------

# Regex to parse SEARCH/REPLACE blocks (5+ markers required)
SEARCH_REPLACE_BLOCK_RE = re.compile(
    r"<{5,} SEARCH\r?\n(.*?)\r?\n?={5,}\r?\n(.*?)\r?\n?>{5,} REPLACE", flags=re.DOTALL
)


def _parse_search_replace_blocks(content: str) -> list[tuple[str, str]]:
    """Parse SEARCH/REPLACE blocks from content. Returns list of (search, replace) tuples."""
    matches = SEARCH_REPLACE_BLOCK_RE.findall(content)
    return [
        (search.rstrip("\r\n"), replace.rstrip("\r\n"))
        for search, replace in matches
    ]


def _find_search_context(content: str, search_text: str, max_context: int = 5) -> str:
    """Find where parts of the search text appear in the content for debugging."""
    lines = content.split("\n")
    search_lines = search_text.split("\n")

    if not search_lines:
        return "Search text is empty"

    first_search_line = search_lines[0].strip()
    if not first_search_line:
        return "First line of search text is empty or whitespace only"

    matches = []
    for i, line in enumerate(lines):
        if first_search_line in line:
            matches.append(i)

    if not matches:
        return f"First search line '{first_search_line[:60]}...' not found anywhere in file"

    context_lines = []
    for match_idx in matches[:3]:
        start = max(0, match_idx - max_context)
        end = min(len(lines), match_idx + max_context + 1)

        context_lines.append(f"\nPotential match area around line {match_idx + 1}:")
        for i in range(start, end):
            marker = ">>>" if i == match_idx else "   "
            context_lines.append(f"{marker} {i + 1:3d}: {lines[i]}")

    return "\n".join(context_lines)


def _find_best_fuzzy_match(
    content: str, search_text: str, threshold: float = 0.9
) -> tuple[float, int, int, str] | None:
    """
    Find the best fuzzy match for search_text within content.
    Returns (similarity, start_line, end_line, matched_text) or None if no match above threshold.
    """
    content_lines = content.split("\n")
    search_lines = search_text.split("\n")
    window_size = len(search_lines)

    if window_size == 0:
        return None

    non_empty_search = [line for line in search_lines if line.strip()]
    if not non_empty_search:
        return None

    first_anchor = non_empty_search[0]
    last_anchor = non_empty_search[-1] if len(non_empty_search) > 1 else first_anchor

    # Find candidate starting positions near anchor lines
    candidate_starts = set()
    spread = 5

    for i, line in enumerate(content_lines):
        if first_anchor in line or last_anchor in line:
            start_min = max(0, i - spread)
            start_max = min(len(content_lines) - window_size + 1, i + spread + 1)
            for s in range(start_min, start_max):
                candidate_starts.add(s)

    # Fall back to first N positions if no anchors found
    if not candidate_starts:
        max_positions = min(len(content_lines) - window_size + 1, 100)
        candidate_starts = set(range(0, max(0, max_positions)))

    best_match = None
    best_similarity = 0.0

    for start in candidate_starts:
        end = start + window_size
        window_text = "\n".join(content_lines[start:end])

        matcher = difflib.SequenceMatcher(None, search_text, window_text)
        similarity = matcher.ratio()

        if similarity >= threshold and similarity > best_similarity:
            best_similarity = similarity
            best_match = (similarity, start + 1, end, window_text)  # 1-based line numbers

    return best_match


def _create_unified_diff(text1: str, text2: str, label1: str = "SEARCH", label2: str = "CLOSEST MATCH") -> str:
    """Create a unified diff between two texts."""
    lines1 = text1.splitlines(keepends=True)
    lines2 = text2.splitlines(keepends=True)

    lines1 = [line if line.endswith("\n") else line + "\n" for line in lines1]
    lines2 = [line if line.endswith("\n") else line + "\n" for line in lines2]

    diff = difflib.unified_diff(lines1, lines2, fromfile=label1, tofile=label2, lineterm="", n=3)
    diff_lines = list(diff)

    if diff_lines and not diff_lines[0].startswith("==="):
        diff_lines.insert(2, "=" * 67 + "\n")

    result = "".join(diff_lines)

    # Truncate long diffs
    max_chars = 2000
    if len(result) > max_chars:
        result = result[:max_chars] + "\n...(diff truncated)"

    return result.rstrip()


def _edit_file(abs_path: str, content: str) -> str:
    """
    Apply SEARCH/REPLACE blocks to a file.
    
    Returns a success message with stats, or an error message with debugging context.
    """
    display = _display_path(abs_path)
    
    # Validate file exists
    if not os.path.exists(abs_path):
        return _err("file_not_found", f"File not found: {display}", path=display)
    if os.path.isdir(abs_path):
        return _err("is_directory", f"Path is a directory, not a file: {display}", path=display, hint="Provide a file path.")

    # Validate content has blocks
    content = (content or "").strip()
    if not content:
        return _err(
            "empty_content",
            "No content provided for edit action.",
            hint="Provide SEARCH/REPLACE blocks in the Content field."
        )

    blocks = _parse_search_replace_blocks(content)
    if not blocks:
        return _err(
            "no_blocks_found",
            "No valid SEARCH/REPLACE blocks found in content.",
            hint=(
                "Expected format:\n"
                "<<<<<<< SEARCH\n"
                "[exact content to find]\n"
                "=======\n"
                "[new content to replace with]\n"
                ">>>>>>> REPLACE"
            )
        )

    # Read file
    try:
        with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
            file_content = f.read()
    except Exception as exc:
        return _err("read_failed", "Failed to read file.", path=display, data={"error": _safe_err(exc)})

    # Apply blocks
    current_content = file_content
    applied = 0
    errors: list[str] = []
    warnings: list[str] = []

    for i, (search, replace) in enumerate(blocks, 1):
        if search not in current_content:
            # Search text not found - provide helpful debugging context
            context = _find_search_context(current_content, search)
            
            error_msg = (
                f"Block {i} failed: Search text not found in {display}\n"
                f"Search text was:\n{search!r}\n"
                f"Context analysis:\n{context}"
            )
            
            # Try fuzzy matching
            fuzzy = _find_best_fuzzy_match(current_content, search, 0.9)
            if fuzzy:
                similarity, start_line, end_line, matched_text = fuzzy
                diff = _create_unified_diff(search, matched_text)
                error_msg += (
                    f"\n\nClosest fuzzy match (similarity {similarity * 100:.1f}%) "
                    f"at lines {start_line}–{end_line}:\n```diff\n{diff}\n```"
                )
            
            error_msg += (
                "\n\nDebugging tips:\n"
                "1. Check for exact whitespace/indentation match\n"
                "2. Verify line endings match the file exactly (\\r\\n vs \\n)\n"
                "3. Ensure the search text hasn't been modified by previous blocks\n"
                "4. Check for typos or case sensitivity issues"
            )
            
            errors.append(error_msg)
            continue

        # Check for multiple occurrences
        occurrences = current_content.count(search)
        if occurrences > 1:
            warnings.append(
                f"Block {i}: Search text appears {occurrences} times. "
                f"Only the first occurrence will be replaced."
            )

        # Apply replacement (first occurrence only)
        current_content = current_content.replace(search, replace, 1)
        applied += 1

    # If all blocks failed, return error
    if errors and applied == 0:
        error_message = "All SEARCH/REPLACE blocks failed:\n\n" + "\n\n---\n\n".join(errors)
        if warnings:
            error_message += "\n\nWarnings:\n" + "\n".join(warnings)
        return _err("edit_failed", error_message)

    # If some blocks failed, still try to apply successful ones but report errors
    if errors:
        partial_msg = f"Partial edit: {applied} of {len(blocks)} blocks applied.\n\n"
        partial_msg += "Failed blocks:\n\n" + "\n\n---\n\n".join(errors)
        if warnings:
            partial_msg += "\n\nWarnings:\n" + "\n".join(warnings)
        
        # Write partial changes
        try:
            with open(abs_path, "w", encoding="utf-8") as f:
                f.write(current_content)
        except Exception as exc:
            return _err("write_failed", "Failed to write file after partial edit.", path=display, data={"error": _safe_err(exc)})
        
        return partial_msg

    # All blocks succeeded - write file
    try:
        with open(abs_path, "w", encoding="utf-8") as f:
            f.write(current_content)
    except Exception as exc:
        return _err("write_failed", "Failed to write file.", path=display, data={"error": _safe_err(exc)})

    # Build success message
    original_lines = len(file_content.splitlines())
    new_lines = len(current_content.splitlines())
    line_delta = new_lines - original_lines
    delta_str = f"+{line_delta}" if line_delta > 0 else str(line_delta)

    result = f"Edited {display}: {applied} block{'s' if applied != 1 else ''} applied"
    if line_delta != 0:
        result += f" ({delta_str} lines)"
    
    if warnings:
        result += "\n\nWarnings:\n" + "\n".join(warnings)
    
    return result

# ---------------------------------------------------------------------------
# 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', 'edit', '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", "edit", "mkdir", "move", "copy", "delete", "info", "search", "help"}:
        result = _err(
            "invalid_action",
            "Invalid action.",
            hint="Choose from: list, read, write, append, edit, 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 == "edit":
            result = _edit_file(abs_path, content or "")
        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", "edit", "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, SEARCH/REPLACE blocks for edit, or search query...", value="<<<<<<< SEARCH\n[exact text to find in the file]\n=======\n[exact text to replace it with]\n>>>>>>> REPLACE", info="Content for write/append/edit 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"]