adds ability to pattern match and list recursive (#116)

Author: shhlife

griptape_nodes_library/files/list_files.py

@@ -1,4 +1,6 @@
 import logging
+from fnmatch import fnmatchcase
+from pathlib import Path
 from typing import Any
 
 from griptape_nodes.exe_types.core_types import Parameter
@@ -46,6 +48,23 @@ def __init__(
             )
         )
 
+        self.match_pattern = ParameterString(
+            name="match_pattern",
+            default_value="",
+            tooltip="Optional shell-style glob for entry names (e.g. *.jpg, report_*.txt). Empty includes all. Applies to both files and folders when listed.",
+        )
+        self.match_pattern_case_sensitive = ParameterBool(
+            name="match_pattern_case_sensitive",
+            default_value=True,
+            tooltip="When match_pattern is set: if True, letter case must match; if False, matching is case-insensitive (e.g. *.jpg matches .JPG). Ignored when match_pattern is empty.",
+        )
+        self.show_hidden = ParameterBool(
+            name="show_hidden",
+            allow_output=False,
+            default_value=False,
+            tooltip="Whether to show hidden files/folders.",
+        )
+
         self.list_options = ParameterString(
             name="list_options",
             allow_output=False,
@@ -54,11 +73,11 @@ def __init__(
             traits={Options(choices=LIST_OPTIONS)},
         )
 
-        self.show_hidden = ParameterBool(
-            name="show_hidden",
+        self.recursive = ParameterBool(
+            name="recursive",
             allow_output=False,
             default_value=False,
-            tooltip="Whether to show hidden files/folders.",
+            tooltip="If True, walk subdirectories and include matches at every depth. If False, only the directory_path level is listed.",
         )
 
         self.use_absolute_paths = ParameterBool(
@@ -69,8 +88,11 @@ def __init__(
         )
 
         self.add_parameter(self.directory_path)
-        self.add_parameter(self.show_hidden)
+        self.add_parameter(self.match_pattern)
+        self.add_parameter(self.match_pattern_case_sensitive)
         self.add_parameter(self.list_options)
+        self.add_parameter(self.recursive)
+        self.add_parameter(self.show_hidden)
         self.add_parameter(self.use_absolute_paths)
 
         # Add output parameters
@@ -81,7 +103,7 @@ def __init__(
                 allow_property=False,
                 output_type="list",
                 default_value=[],
-                tooltip="List of full file paths found in the directory.",
+                tooltip="Paths for matched entries. When recursive is True, includes nested files/folders under directory_path.",
             )
         )
 
@@ -92,7 +114,7 @@ def __init__(
                 allow_property=False,
                 output_type="list",
                 default_value=[],
-                tooltip="List of file names (without path) found in the directory.",
+                tooltip="Base names for matched entries only (no parent path). Same for nested matches when recursive is True.",
             )
         )
 
@@ -110,14 +132,45 @@ def __init__(
             result_details_placeholder="Details on the list file attempt will be presented here.",
         )
 
-    def _filter_entries(self, entries: list, *, include_files: bool, include_folders: bool) -> list:
-        """Filter entries based on include_files and include_folders."""
+    @staticmethod
+    def _directory_visit_key(directory_path: str | None) -> str:
+        """Stable key for visited dirs (avoids repeated work on symlink cycles)."""
+        if not directory_path:
+            return "\x00<list_directory_root>\x00"
+        try:
+            return str(Path(directory_path).resolve())
+        except (OSError, RuntimeError):
+            return str(Path(directory_path).absolute())
+
+    def _name_matches_pattern(self, name: str, pattern: str, *, case_sensitive: bool) -> bool:
+        if case_sensitive:
+            return fnmatchcase(name, pattern)
+        return fnmatchcase(name.casefold(), pattern.casefold())
+
+    def _filter_entries(
+        self,
+        entries: list,
+        *,
+        include_files: bool,
+        include_folders: bool,
+        match_pattern: str,
+        match_pattern_case_sensitive: bool,
+    ) -> list:
+        """Filter entries based on include_files, include_folders, and optional fnmatch pattern on entry.name."""
+        pattern = match_pattern.strip()
+        use_pattern = bool(pattern)
         filtered_entries = []
         for entry in entries:
             if entry.is_dir:
-                if include_folders:
+                if include_folders and (
+                    not use_pattern
+                    or self._name_matches_pattern(entry.name, pattern, case_sensitive=match_pattern_case_sensitive)
+                ):
                     filtered_entries.append(entry)
-            elif include_files:
+            elif include_files and (
+                not use_pattern
+                or self._name_matches_pattern(entry.name, pattern, case_sensitive=match_pattern_case_sensitive)
+            ):
                 filtered_entries.append(entry)
         return filtered_entries
 
@@ -127,14 +180,71 @@ def _convert_paths(self, entries: list, *, use_absolute_paths: bool) -> list[str
             return [entry.absolute_path for entry in entries]
         return [entry.path for entry in entries]
 
+    def _collect_entries_recursive(
+        self,
+        root_directory_path: str | None,
+        *,
+        show_hidden: bool,
+        include_files: bool,
+        include_folders: bool,
+        match_pattern: str,
+        match_pattern_case_sensitive: bool,
+    ) -> tuple[list, str | None]:
+        """Walk subdirectories depth-first. Returns (filtered entries, error_message)."""
+        collected: list = []
+        stack: list[str | None] = [root_directory_path]
+        visited: set[str] = set()
+
+        while stack:
+            current = stack.pop()
+            visit_key = self._directory_visit_key(current)
+            if visit_key in visited:
+                continue
+            visited.add(visit_key)
+
+            request = ListDirectoryRequest(
+                directory_path=current,
+                show_hidden=show_hidden,
+                workspace_only=False,
+            )
+            result = GriptapeNodes.handle_request(request)
+
+            if isinstance(result, ListDirectoryResultFailure):
+                error_msg = getattr(result, "error_message", "Unknown error occurred")
+                return [], str(error_msg)
+            if not isinstance(result, ListDirectoryResultSuccess):
+                return [], "unexpected result type from directory listing"
+
+            # Descend into subdirs in a stable order (reverse so first listed dir is processed next on pop).
+            subdirs: list[str] = []
+            for entry in result.entries:
+                collected.extend(
+                    self._filter_entries(
+                        [entry],
+                        include_files=include_files,
+                        include_folders=include_folders,
+                        match_pattern=match_pattern,
+                        match_pattern_case_sensitive=match_pattern_case_sensitive,
+                    )
+                )
+                if entry.is_dir:
+                    subdirs.append(entry.path)
+            for d in reversed(subdirs):
+                stack.append(d)
+
+        return collected, None
+
     def process(self) -> None:
         self._clear_execution_status()
         directory_path = self.get_parameter_value("directory_path")
         # Clean directory path to remove newlines/carriage returns that cause Windows errors
         if directory_path:
             directory_path = GriptapeNodes.OSManager().sanitize_path_string(directory_path)
         show_hidden = self.get_parameter_value("show_hidden")
+        match_pattern = self.get_parameter_value("match_pattern") or ""
+        match_pattern_case_sensitive = self.get_parameter_value("match_pattern_case_sensitive")
         list_options = self.get_parameter_value("list_options")
+        recursive = self.get_parameter_value("recursive")
         use_absolute_paths = self.get_parameter_value("use_absolute_paths")
 
         # Determine include_files and include_folders based on list_options
@@ -154,30 +264,47 @@ def process(self) -> None:
             include_files = True
             include_folders = True
 
-        # Create the os_events request
-        request = ListDirectoryRequest(
-            directory_path=directory_path if directory_path else None,
-            show_hidden=show_hidden,
-            workspace_only=False,  # Allow system-wide browsing
-        )
+        root = directory_path if directory_path else None
 
-        # Send request through GriptapeNodes.handle_request
-        result = GriptapeNodes.handle_request(request)
+        if recursive:
+            filtered_entries, list_error = self._collect_entries_recursive(
+                root,
+                show_hidden=show_hidden,
+                include_files=include_files,
+                include_folders=include_folders,
+                match_pattern=match_pattern,
+                match_pattern_case_sensitive=match_pattern_case_sensitive,
+            )
+            if list_error:
+                msg = f"{self.name} failed to list directory: {list_error}"
+                self._set_status_results(was_successful=False, result_details=f"Failure: {msg}")
+                return
+        else:
+            request = ListDirectoryRequest(
+                directory_path=root,
+                show_hidden=show_hidden,
+                workspace_only=False,  # Allow system-wide browsing
+            )
+            result = GriptapeNodes.handle_request(request)
 
-        if isinstance(result, ListDirectoryResultFailure):
-            error_msg = getattr(result, "error_message", "Unknown error occurred")
-            msg = f"{self.name} failed to list directory: {error_msg}"
-            self._set_status_results(was_successful=False, result_details=f"Failure: {msg}")
-            return
+            if isinstance(result, ListDirectoryResultFailure):
+                error_msg = getattr(result, "error_message", "Unknown error occurred")
+                msg = f"{self.name} failed to list directory: {error_msg}"
+                self._set_status_results(was_successful=False, result_details=f"Failure: {msg}")
+                return
 
-        if not isinstance(result, ListDirectoryResultSuccess):
-            msg = f"{self.name} received unexpected result type from directory listing"
-            self._set_status_results(was_successful=False, result_details=f"Failure: {msg}")
-            return
+            if not isinstance(result, ListDirectoryResultSuccess):
+                msg = f"{self.name} received unexpected result type from directory listing"
+                self._set_status_results(was_successful=False, result_details=f"Failure: {msg}")
+                return
 
-        filtered_entries = self._filter_entries(
-            result.entries, include_files=include_files, include_folders=include_folders
-        )
+            filtered_entries = self._filter_entries(
+                result.entries,
+                include_files=include_files,
+                include_folders=include_folders,
+                match_pattern=match_pattern,
+                match_pattern_case_sensitive=match_pattern_case_sensitive,
+            )
         file_paths = self._convert_paths(filtered_entries, use_absolute_paths=use_absolute_paths)
         file_names = [entry.name for entry in filtered_entries]