import os import shutil import subprocess from pathlib import Path def path_matches_dir_names(path: str, dir_names: tuple) -> bool: if not dir_names: return False wanted = {name.lower() for name in dir_names} return any(part.lower() in wanted for part in Path(path).parts) def should_exclude_path(path: str, exclude_path_fragments) -> bool: return any(fragment in path for fragment in exclude_path_fragments) def _run_command_capture_stdout(args: list) -> str: try: completed = subprocess.run( args, capture_output=True, text=True, check=False, ) except Exception: return "" if completed.returncode not in (0, 1): return "" return completed.stdout def _discover_files_via_os_walk(scan_dirs: list, allowed_exts: tuple, exclude_path_fragments, skip_dir_names: tuple = (), basename_prefixes: tuple = ()) -> list: files = [] allowed_exts = set(allowed_exts) for scan_dir in scan_dirs: for root, dirs, filenames in os.walk(scan_dir): dirs[:] = [ directory for directory in dirs if not any(fragment.strip("/") in directory for fragment in exclude_path_fragments) ] if path_matches_dir_names(root, skip_dir_names): continue for filename in filenames: ext = os.path.splitext(filename)[1] if ext not in allowed_exts: continue name_no_ext = os.path.splitext(filename)[0] if basename_prefixes and not any(name_no_ext.startswith(prefix) for prefix in basename_prefixes): continue full_path = os.path.normpath(os.path.join(root, filename)) if should_exclude_path(full_path, exclude_path_fragments): continue if path_matches_dir_names(full_path, skip_dir_names): continue files.append(full_path) return sorted(set(files)) def _discover_files_via_rg(scan_dirs: list, allowed_exts: tuple, exclude_path_fragments, skip_dir_names: tuple = (), basename_prefixes: tuple = ()) -> list: rg = shutil.which("rg") if not rg: return [] cmd = [rg, "--files", "--no-ignore", "--hidden"] for ext in allowed_exts: cmd.extend(["-g", f"*{ext}"]) cmd.extend(scan_dirs) output = _run_command_capture_stdout(cmd) if not output: return [] files = [] for line in output.splitlines(): path = os.path.normpath(line.strip()) if not path or should_exclude_path(path, exclude_path_fragments): continue if path_matches_dir_names(path, skip_dir_names): continue if basename_prefixes: name_no_ext = os.path.splitext(os.path.basename(path))[0] if not any(name_no_ext.startswith(prefix) for prefix in basename_prefixes): continue files.append(path) return sorted(set(files)) def discover_files(scan_dirs: list, allowed_exts: tuple, exclude_path_fragments, skip_dir_names: tuple = (), basename_prefixes: tuple = ()) -> list: files = _discover_files_via_rg( scan_dirs, allowed_exts, exclude_path_fragments, skip_dir_names=skip_dir_names, basename_prefixes=basename_prefixes, ) if files: return files return _discover_files_via_os_walk( scan_dirs, allowed_exts, exclude_path_fragments, skip_dir_names=skip_dir_names, basename_prefixes=basename_prefixes, ) def _chunked(items: list, chunk_size: int): for index in range(0, len(items), chunk_size): yield items[index:index + chunk_size] def prefilter_files_with_rg(files: list, pattern: str, chunk_size: int = 256) -> list: files = sorted(set(files)) if not files: return [] rg = shutil.which("rg") if not rg: return files matched = set() for chunk in _chunked(files, chunk_size): try: completed = subprocess.run( [rg, "-l", "--no-messages", "-e", pattern, *chunk], capture_output=True, text=True, check=False, ) except Exception: return files if completed.returncode not in (0, 1): return files for line in completed.stdout.splitlines(): path = os.path.normpath(line.strip()) if path: matched.add(path) return sorted(matched)