backend/scripts/prune_media.py

"""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()