File size: 2,929 Bytes
de1e3fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """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()
|