"""Delete media files not referenced by any `pictures` row (orphans left by purges / re-seeds). The DB is the source of truth: a file is an orphan if no Picture row points at it via `file_path` or `thumb_path`. Dry-run by default — pass --delete to actually remove. Usage: python -m scripts.prune_media # report only (safe) python -m scripts.prune_media --delete # remove orphan files (+ now-empty dirs) """ from __future__ import annotations import argparse import os from pathlib import Path from sqlalchemy import select from app.config import settings from app.db import SessionLocal from app.models import Picture def run(delete: bool) -> None: media = settings.media_path if not media.exists(): print(f"Media dir does not exist: {media}") return db = SessionLocal() referenced: set[Path] = set() for file_path, thumb_path in db.execute(select(Picture.file_path, Picture.thumb_path)).all(): if file_path: referenced.add((media / file_path).resolve()) if thumb_path: referenced.add((media / thumb_path).resolve()) db.close() on_disk = 0 orphans: list[Path] = [] orphan_bytes = 0 for root, _dirs, files in os.walk(media): for f in files: p = (Path(root) / f).resolve() on_disk += 1 if p not in referenced: orphans.append(p) try: orphan_bytes += p.stat().st_size except OSError: pass print(f"media dir: {media}") print(f"files on disk: {on_disk}") print(f"referenced by DB: {len(referenced)}") print(f"orphan files: {len(orphans)} ({orphan_bytes / 1e6:.1f} MB)") if not delete: print("\nDRY RUN — nothing deleted. Re-run with --delete to remove them.") for p in orphans[:8]: print(f" would delete: {p}") if len(orphans) > 8: print(f" ... and {len(orphans) - 8} more") return removed = 0 for p in orphans: try: p.unlink() removed += 1 except OSError as exc: print(f" ! could not delete {p}: {exc}") # Clean up now-empty subdirectories (deepest first). for root, _dirs, _files in os.walk(media, topdown=False): rp = Path(root) if rp == media: continue try: if not os.listdir(rp): rp.rmdir() except OSError: pass print(f"\nDeleted {removed} orphan file(s); {len(referenced)} live file(s) kept.") def main() -> None: parser = argparse.ArgumentParser(description="Prune media files not referenced by the DB.") parser.add_argument("--delete", action="store_true", help="Actually delete (default: dry run)") args = parser.parse_args() run(args.delete) if __name__ == "__main__": main()