46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
"""Recursive file discovery with extension filtering.
|
|
|
|
Discovers files for ingestion by walking directories and filtering
|
|
to supported document types.
|
|
"""
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SUPPORTED_EXTENSIONS = {".pdf", ".jpg", ".jpeg", ".png", ".md", ".txt", ".docx"}
|
|
|
|
|
|
def discover_files(path: Path) -> list[Path]:
|
|
"""Discover supported files from a path (file or directory).
|
|
|
|
Args:
|
|
path: A single file path or a directory to scan recursively.
|
|
|
|
Returns:
|
|
Sorted list of Path objects with supported extensions.
|
|
|
|
Raises:
|
|
FileNotFoundError: If the path does not exist.
|
|
"""
|
|
path = Path(path)
|
|
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Path does not exist: {path}")
|
|
|
|
if path.is_file():
|
|
if path.suffix.lower() in SUPPORTED_EXTENSIONS:
|
|
return [path]
|
|
logger.warning("Unsupported file type: %s", path)
|
|
return []
|
|
|
|
# Recursive directory traversal
|
|
files: list[Path] = []
|
|
for item in sorted(path.rglob("*")):
|
|
if item.is_file() and item.suffix.lower() in SUPPORTED_EXTENSIONS:
|
|
files.append(item)
|
|
|
|
logger.info("Discovered %d supported files in %s", len(files), path)
|
|
return files
|