Add source cleanup helpers

scene_search/maintenance.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+
+from .database import connect, init_db
+
+
+@dataclass(frozen=True)
+class RemovedMedia:
+    media_id: str
+    title: str
+    source_kind: str
+
+
+def list_media_by_source(db_path: Path, source_kind: str) -> list[RemovedMedia]:
+    init_db(db_path)
+    with connect(db_path) as con:
+        rows = con.execute(
+            """
+            SELECT id, title, source_kind
+            FROM media
+            WHERE source_kind = ?
+            ORDER BY title, id
+            """,
+            (source_kind,),
+        ).fetchall()
+
+    return [
+        RemovedMedia(
+            media_id=str(row["id"]),
+            title=str(row["title"]),
+            source_kind=str(row["source_kind"]),
+        )
+        for row in rows
+    ]
+
+
+def remove_media_by_source(
+    db_path: Path,
+    thumbs_dir: Path,
+    source_kind: str,
+) -> list[RemovedMedia]:
+    matches = list_media_by_source(db_path, source_kind)
+    if not matches:
+        return []
+
+    ids = [item.media_id for item in matches]
+    placeholders = ",".join("?" for _ in ids)
+
+    with connect(db_path) as con:
+        con.execute(
+            f"DELETE FROM media WHERE id IN ({placeholders})",
+            ids,
+        )
+
+    for item in matches:
+        shutil.rmtree(thumbs_dir / item.media_id, ignore_errors=True)
+
+    return matches