144 lines
4.5 KiB
Python
144 lines
4.5 KiB
Python
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)
|