"""Resolve stored upload paths (Docker-safe).""" from __future__ import annotations import re from pathlib import Path, PureWindowsPath from app.services.ingest import ensure_upload_dir def to_stored_path(path: Path) -> str: """Store path relative to upload root (portable).""" root = ensure_upload_dir().resolve() resolved = path.resolve() try: return resolved.relative_to(root).as_posix() except ValueError: return resolved.name def _after_uploads(parts: tuple[str, ...] | list[str]) -> Path | None: """Return path segments after an 'uploads' directory.""" lowered = [p.lower() for p in parts] if "uploads" not in lowered: return None idx = lowered.index("uploads") rest = parts[idx + 1 :] if not rest: return None return Path(*rest) def resolve_stored_path(stored: str) -> Path: """ Resolve a DB file_path to an on-disk Path. Supports relative (preferred) and legacy absolute paths from Windows host or Linux container. """ root = ensure_upload_dir().resolve() raw = Path(stored) # Preferred: relative to upload root if not raw.is_absolute() and not re.match(r"^[A-Za-z]:[\\/]", stored): candidate = (root / raw).resolve() if candidate.is_file(): return candidate raise FileNotFoundError(stored) # Legacy absolute that still exists (same machine) if raw.is_absolute() and raw.is_file(): return raw.resolve() # Windows absolute path string while running in Linux/Docker win = PureWindowsPath(stored) rel = _after_uploads(win.parts) if rel is not None: candidate = (root / rel).resolve() if candidate.is_file(): return candidate # POSIX absolute with /.../uploads/... rel = _after_uploads(raw.parts) if rel is not None: candidate = (root / rel).resolve() if candidate.is_file(): return candidate by_name = root / Path(stored.replace("\\", "/")).name if by_name.is_file(): return by_name.resolve() raise FileNotFoundError(stored)