| """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) |
|
|
| |
| 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) |
|
|
| |
| if raw.is_absolute() and raw.is_file(): |
| return raw.resolve() |
|
|
| |
| win = PureWindowsPath(stored) |
| rel = _after_uploads(win.parts) |
| if rel is not None: |
| candidate = (root / rel).resolve() |
| if candidate.is_file(): |
| return candidate |
|
|
| |
| 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) |
|
|
