reading: owner archive / restore / delete / merge for books (⋯ menu on cards, rows and detail pages), archived filter, tombstones so merged ids redirect and the ingest never recreates deleted works, books-quarantine.py for deleted files

scripts/books-quarantine.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Move the files of deleted books to quarantine.
+
+Deleting a book on the site removes its rows but leaves the files in place (the app sees the
+library through a read-only mount). Each delete leaves a tombstone in books.deleted_works with
+the file paths; this moves those files to $BOOKS_ROOT/quarantine/deleted/<work id>/<path> and
+stamps quarantined_at. Files that another work still references are left alone.
+
+Usage (dry run by default):
+  DB_URL=postgresql://... BOOKS_ROOT=~/books python scripts/books-quarantine.py [--apply]
+Undo: move the files back from quarantine/deleted/<id>/ into library/.
+"""
+from __future__ import annotations
+
+import argparse, os, shutil, sys
+from pathlib import Path
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--apply", action="store_true", help="actually move files (default: dry run)")
+    args = ap.parse_args()
+    db_url = os.environ.get("DB_URL") or sys.exit("DB_URL required")
+    root = Path(os.path.expanduser(os.environ.get("BOOKS_ROOT", "/srv/books")))
+    library, quarantine = root / "library", root / "quarantine" / "deleted"
+
+    import psycopg
+    with psycopg.connect(db_url) as conn, conn.cursor() as cur:
+        cur.execute("""select d.id, d.title, d.file_paths from books.deleted_works d
+                       where d.quarantined_at is null and cardinality(d.file_paths) > 0 order by d.id""")
+        rows = cur.fetchall()
+        moved = kept = missing = 0
+        for wid, title, paths in rows:
+            for rel in paths:
+                src = (library / rel).resolve()
+                if not str(src).startswith(str(library.resolve()) + os.sep):
+                    print(f"#{wid}: unsafe path, skipped: {rel}")
+                    continue
+                cur.execute("select work_id from books.files where path = %s", (rel,))
+                if cur.fetchone():
+                    kept += 1
+                    print(f"#{wid}: still used by another work, kept: {rel}")
+                    continue
+                if not src.exists():
+                    missing += 1
+                    continue
+                dest = quarantine / str(wid) / rel
+                print(f"#{wid} {title}: {rel} -> {dest.relative_to(root)}")
+                if args.apply:
+                    dest.parent.mkdir(parents=True, exist_ok=True)
+                    shutil.move(str(src), dest)
+                moved += 1
+            if args.apply:
+                cur.execute("update books.deleted_works set quarantined_at = now() where id = %s", (wid,))
+        if args.apply:
+            conn.commit()
+        print(f"{'moved' if args.apply else 'would move'} {moved}, kept {kept}, already gone {missing}")
+
+
+if __name__ == "__main__":
+    main()