Files
photoanalyzer/photo_pipeline/path_policy.py

79 lines
2.8 KiB
Python

"""Library boundary and exclusion policy — the single source used everywhere.
One function decides what is inside the library, what is a supported photo, and
what is excluded (`_IGNORE/` and generated thumbnail/cache directories). Symlinks
that resolve outside the library root are rejected rather than silently followed,
so discovery can never escape the configured boundary.
Extracted from photo_analyzer.discover_photos / SUPPORTED_EXTENSIONS
(see donor_ledger.yaml: pa-discovery).
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Iterable, Iterator
SUPPORTED_EXTENSIONS = {
".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif", ".tiff", ".tif"
}
EXCLUDED_DIR_NAMES = {"_IGNORE", ".@__thumb"}
class PathPolicyError(ValueError):
"""A path violates the library boundary (e.g. a symlink escaping the root)."""
def is_supported(path: os.PathLike | str) -> bool:
return Path(path).suffix.lower() in SUPPORTED_EXTENSIONS
def is_excluded(path: os.PathLike | str) -> bool:
return any(part in EXCLUDED_DIR_NAMES for part in Path(path).parts)
def normalize_root(root: os.PathLike | str) -> Path:
return Path(root).expanduser().resolve()
def resolve_within(root: Path, path: os.PathLike | str) -> Path:
"""Resolve ``path`` (following symlinks) and confirm it stays within ``root``.
Raises ``PathPolicyError`` if the resolved path escapes the root.
"""
root = Path(root).resolve()
resolved = Path(path).resolve()
if resolved != root and root not in resolved.parents:
raise PathPolicyError(f"path escapes library root {root}: {path} -> {resolved}")
return resolved
def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
"""Yield supported, non-excluded files under ``root`` in deterministic order.
Excluded directories are never descended into. A symlinked file whose target
escapes the root raises ``PathPolicyError`` (unsafe paths fail rather than
leaking outside the library).
"""
root = normalize_root(root)
found: list[Path] = []
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
dirnames[:] = sorted(d for d in dirnames if d not in EXCLUDED_DIR_NAMES)
for name in filenames:
candidate = Path(dirpath) / name
if not is_supported(candidate) or is_excluded(candidate):
continue
if candidate.is_symlink():
resolve_within(root, candidate) # raises on escape
found.append(candidate)
return iter(sorted(found))
def discover(roots: Iterable[os.PathLike | str]) -> list[Path]:
"""Deterministic, de-duplicated discovery across one or more roots."""
seen: set[Path] = set()
for root in roots:
seen.update(iter_supported_files(root))
return sorted(seen)