Update Modules/File_System.py

Modules/File_System.py

@@ -2,16 +2,20 @@ from __future__ import annotations
 
 import json
 import os
-import re
 import shutil
-import stat
-from datetime import datetime
 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 = (
@@ -53,335 +57,29 @@ HELP_TEXT = (
 )
 
 
-def _default_root() -> str:
-    # Prefer explicit root via env var
-    root = os.getenv("NYMBO_TOOLS_ROOT")
-    if root and root.strip():
-        return os.path.abspath(os.path.expanduser(root.strip()))
-    # Default to "Nymbo-Tools/Filesystem" alongside this module package
-    try:
-        here = os.path.abspath(__file__)
-        tools_dir = os.path.dirname(os.path.dirname(here))  # .../Nymbo-Tools
-        default_root = os.path.abspath(os.path.join(tools_dir, "Filesystem"))
-        return default_root
-    except Exception:
-        # Final fallback
-        return os.path.abspath(os.getcwd())
-
-
-ROOT_DIR = _default_root()
-# Ensure the default root directory exists to make listing/writing more convenient
-try:
-    os.makedirs(ROOT_DIR, exist_ok=True)
-except Exception:
-    pass
-ALLOW_ABS = bool(int(os.getenv("UNSAFE_ALLOW_ABS_PATHS", "0")))
-
-def _safe_err(exc: Exception | str) -> str:
-    """Return an error string with any absolute root replaced by '/' and slashes normalized.
-    This handles variants like backslashes and duplicate slashes in OS messages.
-    """
-    s = str(exc)
-    # Normalize to forward slashes for comparison
-    s_norm = s.replace("\\", "/")
-    root_fwd = ROOT_DIR.replace("\\", "/")
-    # Collapse duplicate slashes in root representation
-    root_variants = {ROOT_DIR, root_fwd, re.sub(r"/+", "/", root_fwd)}
-    for variant in root_variants:
-        if variant:
-            s_norm = s_norm.replace(variant, "/")
-    # Collapse duplicate slashes in final output
-    s_norm = re.sub(r"/+", "/", s_norm)
-    return s_norm
-
-
-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.
-    Fields: status='error', code, message, path?, hint?, data?, root='/'
-    """
-    payload = {
-        "status": "error",
-        "code": code,
-        "message": message,
-        "root": "/",
-    }
-    if path is not None and path != "":
-        payload["path"] = path
-    if hint:
-        payload["hint"] = hint
-    if data:
-        payload["data"] = data
-    return json.dumps(payload, ensure_ascii=False)
-
+# Convenience aliases using the filesystem sandbox
+_sandbox = filesystem_sandbox
 
 def _resolve_path(path: str) -> tuple[str, str]:
-    """
-    Resolve a user-provided path to an absolute, normalized path constrained to ROOT_DIR
-    (unless UNSAFE_ALLOW_ABS_PATHS=1). Returns (abs_path, error_message). error_message is empty when ok.
-    """
-    try:
-        user_input = (path or "/").strip() or "/"
-        if user_input.startswith("/"):
-            # Treat leading '/' as the virtual root for the tool.
-            rel_part = user_input.lstrip("/") or "."
-            raw = os.path.expanduser(rel_part)
-            treat_as_relative = True
-        else:
-            raw = os.path.expanduser(user_input)
-            treat_as_relative = False
-
-        if not treat_as_relative and os.path.isabs(raw):
-            if not ALLOW_ABS:
-                # Absolute paths are not allowed in safe mode
-                return "", _err(
-                    "absolute_path_disabled",
-                    "Absolute paths are disabled in safe mode.",
-                    path=raw.replace("\\", "/"),
-                    hint="Use a path relative to / (e.g., /notes/todo.txt)."
-                )
-            abs_path = os.path.abspath(raw)
-        else:
-            abs_path = os.path.abspath(os.path.join(ROOT_DIR, raw))
-        # Constrain to ROOT when not unsafe mode
-        if not ALLOW_ABS:
-            try:
-                common = os.path.commonpath([os.path.normpath(ROOT_DIR), os.path.normpath(abs_path)])
-                if common != os.path.normpath(ROOT_DIR):
-                    return "", _err("path_outside_root", "Path is outside the sandbox root.", path=abs_path)
-            except Exception:
-                return "", _err("path_outside_root", "Path is outside the sandbox root.", path=abs_path)
-
-        return abs_path, ""
-    except Exception as exc:
-        return "", _err(
-            "resolve_path_failed",
-            "Failed to resolve path.",
-            path=(path or ""),
-            data={"error": _safe_err(exc)}
-        )
-
-
-def safe_open(file, *args, **kwargs):
-    """
-    A drop-in replacement for open() that enforces sandbox constraints.
-    """
-    # Handle integer file descriptors (pass through) or bytes (decode)
-    if isinstance(file, int):
-        return open(file, *args, **kwargs)
-    
-    path_str = os.fspath(file)
-    abs_path, err = _resolve_path(path_str)
-    if err:
-        # Extract message from JSON error if possible
-        try:
-            msg = json.loads(err)["message"]
-        except:
-            msg = err
-        raise PermissionError(f"Sandboxed open() failed: {msg}")
-    
-    return open(abs_path, *args, **kwargs)
-
-
-
-
-
-def _fmt_size(num_bytes: int) -> str:
-    units = ["B", "KB", "MB", "GB", "TB"]
-    size = float(num_bytes)
-    for unit in units:
-        if size < 1024.0:
-            return f"{size:.1f} {unit}"
-        size /= 1024.0
-    return f"{size:.1f} PB"
-
+    """Resolve path using the filesystem sandbox."""
+    return _sandbox.resolve_path(path)
 
 def _display_path(abs_path: str) -> str:
-    """Return a user-friendly path relative to ROOT_DIR using forward slashes.
-    Example: ROOT_DIR -> '/', a file under it -> '/sub/dir/file.txt'."""
-    try:
-        norm_root = os.path.normpath(ROOT_DIR)
-        norm_abs = os.path.normpath(abs_path)
-        common = os.path.commonpath([norm_root, norm_abs])
-        if os.path.normcase(common) == os.path.normcase(norm_root):
-            rel = os.path.relpath(norm_abs, norm_root)
-            if rel == ".":
-                return "/"
-            return "/" + rel.replace("\\", "/")
-    except Exception:
-        pass
-    # Fallback to original absolute path
-    return abs_path.replace("\\", "/")
-
-
-def _list_dir(abs_path: str, *, show_hidden: bool, recursive: bool, max_entries: int) -> str:
-    from ._tree_utils import format_dir_listing
-    
-    return format_dir_listing(
-        abs_path,
-        _display_path(abs_path),
-        show_hidden=show_hidden,
-        recursive=recursive,
-        max_entries=max_entries,
-        fmt_size_fn=_fmt_size,
-    )
-
+    """Display path using the filesystem sandbox."""
+    return _sandbox.display_path(abs_path)
 
-def _search_text(
-    abs_path: str,
-    query: str,
-    *,
-    recursive: bool,
-    show_hidden: bool,
-    max_results: int,
-    case_sensitive: bool,
-    start_index: int,
-) -> str:
-    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))
-
-    query = query or ""
-    normalized_query = query if case_sensitive else query.lower()
-    if normalized_query == "":
-        return _err(
-            "missing_search_query",
-            "Search query is required for the search action.",
-            hint="Provide text in the Content field to search for.",
-        )
-
-    max_results = max(1, int(max_results) if max_results is not None else 20)
-    start_index = max(0, int(start_index) if start_index is not None else 0)
-    matches: list[tuple[str, int, str]] = []
-    errors: list[str] = []
-    files_scanned = 0
-    truncated = False
-    total_matches = 0
-
-    def _should_skip(name: str) -> bool:
-        return not show_hidden and name.startswith('.')
-
-    def _handle_match(file_path: str, line_no: int, line_text: str) -> bool:
-        nonlocal truncated, total_matches
-        total_matches += 1
-        if total_matches <= start_index:
-            return False
-        if len(matches) < max_results:
-            snippet = line_text.strip()
-            if len(snippet) > 200:
-                snippet = snippet[:197] + "…"
-            matches.append((_display_path(file_path), line_no, snippet))
-            return False
-        truncated = True
-        return True
-
-    def _search_file(file_path: str) -> bool:
-        nonlocal files_scanned
-        files_scanned += 1
-        try:
-            with open(file_path, 'r', encoding='utf-8', errors='replace') as handle:
-                for line_no, line in enumerate(handle, start=1):
-                    haystack = line if case_sensitive else line.lower()
-                    if normalized_query in haystack:
-                        if _handle_match(file_path, line_no, line):
-                            return True
-        except Exception as exc:
-            errors.append(f"{_display_path(file_path)} ({_safe_err(exc)})")
-        return truncated
-
-    if os.path.isfile(abs_path):
-        _search_file(abs_path)
-    else:
-        for root, dirs, files in os.walk(abs_path):
-            dirs[:] = [d for d in dirs if not _should_skip(d)]
-            visible_files = [f for f in files if show_hidden or not f.startswith('.')]
-            for name in visible_files:
-                file_path = os.path.join(root, name)
-                if _search_file(file_path):
-                    break
-            if truncated:
-                break
-            if not recursive:
-                break
-
-    header_lines = [
-        f"Search results for {query!r}",
-        f"Scope: {_display_path(abs_path)}",
-        f"Recursive: {'yes' if recursive else 'no'}, Hidden: {'yes' if show_hidden else 'no'}, Case-sensitive: {'yes' if case_sensitive else 'no'}",
-        f"Start offset: {start_index}",
-        f"Matches returned: {len(matches)}" + (" (truncated)" if truncated else ""),
-        f"Files scanned: {files_scanned}",
-    ]
-
-    next_cursor = start_index + len(matches) if truncated else None
-
-    if truncated:
-        header_lines.append(f"Matches encountered before truncation: {total_matches}")
-        header_lines.append(f"Truncated: yes — re-run with offset={next_cursor} to continue.")
-        header_lines.append(f"Next cursor: {next_cursor}")
-    else:
-        header_lines.append(f"Total matches found: {total_matches}")
-        header_lines.append("Truncated: no — end of results.")
-        header_lines.append("Next cursor: None")
-
-    if not matches:
-        if total_matches > 0 and start_index >= total_matches:
-            hint_limit = max(total_matches - 1, 0)
-            body_lines = [
-                f"No matches found at or after offset {start_index}. Total matches available: {total_matches}.",
-                (f"Try a smaller offset (≤ {hint_limit})." if hint_limit >= 0 else ""),
-            ]
-            body_lines = [line for line in body_lines if line]
-        else:
-            body_lines = [
-                "No matches found.",
-                (f"Total matches encountered: {total_matches}." if total_matches else ""),
-            ]
-            body_lines = [line for line in body_lines if line]
-    else:
-        body_lines = [f"{idx}. {path}:{line_no}: {text}" for idx, (path, line_no, text) in enumerate(matches, start=1)]
-
-    if errors:
-        shown = errors[:5]
-        body_lines.extend(["", "Warnings:"])
-        body_lines.extend(shown)
-        if len(errors) > len(shown):
-            body_lines.append(f"… {len(errors) - len(shown)} additional files could not be read.")
-
-    return "\n".join(header_lines) + "\n\n" + "\n".join(body_lines)
+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 _read_file(abs_path: str, *, offset: int, max_chars: int) -> str:
-    if not os.path.exists(abs_path):
-        return _err("file_not_found", f"File not found: {_display_path(abs_path)}", path=_display_path(abs_path))
-    if os.path.isdir(abs_path):
-        return _err("is_directory", f"Path is a directory, not a file: {_display_path(abs_path)}", path=_display_path(abs_path), hint="Provide a file path.")
-    try:
-        with open(abs_path, 'r', encoding='utf-8', errors='replace') as f:
-            data = f.read()
-    except Exception as exc:
-        return _err("read_failed", "Failed to read file.", path=_display_path(abs_path), data={"error": _safe_err(exc)})
-    total = len(data)
-    start = max(0, min(offset, total))
-    if max_chars > 0:
-        end = min(total, start + max_chars)
-    else:
-        end = total
-    chunk = data[start:end]
-    next_cursor = end if end < total else None
-    meta = {
-        "offset": start,
-        "returned": len(chunk),
-        "total": total,
-        "next_cursor": next_cursor,
-        "path": _display_path(abs_path),
-    }
-    header = (
-        f"Reading {_display_path(abs_path)}\n"
-        f"Offset {start}, returned {len(chunk)} of {total}."
-        + (f"\nNext cursor: {next_cursor}" if next_cursor is not None else "")
-    )
-    sep = "\n\n---\n\n"
-    return header + sep + chunk
 
+# ---------------------------------------------------------------------------
+# 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)
@@ -416,7 +114,6 @@ def _move_copy(action: str, src: str, dst: str, *, overwrite: bool) -> str:
         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):
-            # allow moving into an existing directory
             dst_path = os.path.join(dst, os.path.basename(src))
         else:
             dst_path = dst
@@ -451,7 +148,6 @@ def _delete(abs_path: str, *, recursive: bool) -> str:
             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:
-                # Refuse to delete a dir unless recursive=True
                 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:
@@ -460,22 +156,46 @@ def _delete(abs_path: str, *, recursive: bool) -> str:
     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:
-    try:
-        st = os.stat(abs_path)
-    except Exception as exc:
-        return _err("stat_failed", "Failed to stat path.", path=_display_path(abs_path), data={"error": _safe_err(exc)})
-    info = {
-        "path": _display_path(abs_path),
-        "type": "directory" if stat.S_ISDIR(st.st_mode) else "file",
-        "size": st.st_size,
-        "modified": datetime.fromtimestamp(st.st_mtime).isoformat(sep=' ', timespec='seconds'),
-        "created": datetime.fromtimestamp(st.st_ctime).isoformat(sep=' ', timespec='seconds'),
-        "mode": oct(st.st_mode),
-        "root": "/",
-    }
-    return json.dumps(info, indent=2)
+    """Get file/directory metadata as JSON."""
+    return _sandbox.info(abs_path)
+
 
 
 @autodoc(summary=TOOL_SUMMARY)