Update Modules/File_System.py

Modules/File_System.py

@@ -2,11 +2,11 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import shutil
 import stat
 from datetime import datetime
 from typing import Annotated, Optional
-import re
 
 import gradio as gr
 
@@ -15,8 +15,8 @@ from ._docstrings import autodoc
 
 
 TOOL_SUMMARY = (
-    "Browse and manage files within a safe root. "
-    "Actions: list, read, write, append, mkdir, move, copy, delete, info, help. "
+    "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 `.` because all paths are relative to the root (`/`). "
     "Use 'help' to see action-specific required fields and examples."
@@ -27,7 +27,7 @@ HELP_TEXT = (
     "Root: paths resolve under Nymbo-Tools/Filesystem by default (or NYMBO_TOOLS_ROOT if set). "
     "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=200\n"
+    "- list: path='.' (default), recursive=false, show_hidden=false, max_entries=20\n"
     "- read: path, 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"
@@ -36,6 +36,7 @@ HELP_TEXT = (
     "- 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"
@@ -45,6 +46,9 @@ HELP_TEXT = (
     "- 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"
 )
 
 
@@ -240,6 +244,129 @@ def _list_dir(abs_path: str, *, show_hidden: bool, recursive: bool, max_entries:
     return (header + "\n" + "\n".join(lines)).strip()
 
 
+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 _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))
@@ -371,17 +498,18 @@ def _info(abs_path: str) -> str:
 
 @autodoc(summary=TOOL_SUMMARY)
 def File_System(
-    action: Annotated[str, "Operation to perform: 'list', 'read', 'write', 'append', 'mkdir', 'move', 'copy', 'delete', 'info'."],
+    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 (UTF-8)."] = None,
+    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 (recurse into subfolders) and delete (required for directories)."] = False,
-    show_hidden: Annotated[bool, "Include hidden files (dotfiles)."] = False,
-    max_entries: Annotated[int, "Max entries to list (for list)."] = 200,
+    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",
@@ -395,13 +523,14 @@ def File_System(
         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", "help"}:
+    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, help."
+            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
@@ -445,6 +574,24 @@ def File_System(
                     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:
@@ -460,25 +607,26 @@ def build_interface() -> gr.Interface:
         inputs=[
             gr.Radio(
                 label="Action",
-                choices=["list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "help"],
+                choices=["list", "read", "write", "append", "mkdir", "move", "copy", "delete", "info", "search", "help"],
                 value="help",
             ),
             gr.Textbox(label="Path", placeholder=". or src/file.txt", max_lines=1, value="."),
-            gr.Textbox(label="Content (for write/append)", lines=6, placeholder="Text to write..."),
+            gr.Textbox(label="Content (write/append/search)", lines=6, placeholder="Text to write or search for..."),
             gr.Textbox(label="Destination (for move/copy)", max_lines=1),
-            gr.Checkbox(label="Recursive (list/delete)", value=False),
-            gr.Checkbox(label="Show hidden (list)", value=False),
-            gr.Slider(minimum=10, maximum=5000, step=10, value=200, label="Max entries (list)"),
-            gr.Slider(minimum=0, maximum=1_000_000, step=100, value=0, label="Offset (read)"),
+            gr.Checkbox(label="Recursive (list/delete/search)", value=False),
+            gr.Checkbox(label="Show hidden (list/search)", value=False),
+            gr.Slider(minimum=10, maximum=5000, step=10, value=20, label="Max entries / matches (list/search)"),
+            gr.Slider(minimum=0, maximum=1_000_000, step=100, value=0, label="Offset (read/search start)"),
             gr.Slider(minimum=0, maximum=100_000, step=500, value=4000, label="Max chars (read, 0=all)"),
             gr.Checkbox(label="Create parent dirs (write)", value=True),
             gr.Checkbox(label="Overwrite destination (move/copy)", value=False),
+            gr.Checkbox(label="Case sensitive search", value=False),
         ],
         outputs=gr.Textbox(label="Result", lines=20),
         title="File System",
         description=(
-            "<div id=\"fs-desc\" style=\"text-align:center; overflow:hidden;\">Browse and interact with a filesystem. "
-            "Actions are required, fill other fields as needed." 
+            "<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,