chore(scripts): migration Phase C — destructive .md attachment cleanup

Gated by --cleanup flag and y/N confirmation. Walks bib.attachments
and Zotero itemAttachments + items, removes storage dirs that
contain only the .md file (non-empty dirs with foreign content are
logged and the dir is skipped, but the SQL rows still drop).

Safety pre-flight refuses to run unless src/bib/sync.py contains
_sync_notes — sanity check that re-syncing won't recreate the
attachments we're deleting.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scripts/migrate_comment_md_to_notes.py

@@ -21,6 +21,8 @@ from __future__ import annotations
 
 import argparse
 import logging
+import shutil
+import sqlite3
 import sys
 from pathlib import Path
 
@@ -101,6 +103,94 @@ def phase_a_backfill(root: Path) -> dict[str, int]:
     return stats
 
 
+def _confirm(msg: str) -> bool:
+    """Prompt y/N. Defaults to N."""
+    reply = input(f"{msg} [y/N] ").strip().lower()
+    return reply == "y"
+
+
+def _phase_c_preview() -> dict[str, int]:
+    from conf import path
+    bib_db = sqlite3.connect(str(path("db.bib")))
+    n_bib = bib_db.execute(
+        "SELECT COUNT(*) FROM attachments WHERE filename LIKE '%.md'"
+    ).fetchone()[0]
+    bib_db.close()
+    zot_db = sqlite3.connect(str(path("db.zotero")))
+    n_zot = zot_db.execute(
+        "SELECT COUNT(*) FROM itemAttachments WHERE path LIKE 'storage:%.md'"
+    ).fetchone()[0]
+    zot_db.close()
+    return {"bib_md_attachments": n_bib, "zotero_md_attachments": n_zot}
+
+
+def phase_c_cleanup() -> dict[str, int]:
+    """Delete every .md attachment row + storage dir on bib and Zotero."""
+    from conf import path
+
+    stats = {"bib_rows_deleted": 0, "bib_dirs_removed": 0,
+             "zot_rows_deleted": 0, "zot_dirs_removed": 0,
+             "zot_dirs_skipped_nonempty": 0}
+
+    # ── bib ──────────────────────────────────────────
+    bib_con = sqlite3.connect(str(path("db.bib")))
+    bib_rows = bib_con.execute(
+        "SELECT id, storage_path FROM attachments WHERE filename LIKE '%.md'"
+    ).fetchall()
+    for _att_id, storage_path in bib_rows:
+        if storage_path:
+            d = Path(storage_path).parent
+            if d.is_dir():
+                shutil.rmtree(d, ignore_errors=True)
+                stats["bib_dirs_removed"] += 1
+    bib_con.execute("DELETE FROM attachments WHERE filename LIKE '%.md'")
+    stats["bib_rows_deleted"] = bib_con.total_changes
+    bib_con.commit()
+    bib_con.close()
+
+    # ── Zotero ───────────────────────────────────────
+    zot_db_path = str(path("db.zotero"))
+    zot_storage = Path(zot_db_path).parent / "storage"
+    zot_con = sqlite3.connect(zot_db_path)
+    zot_rows = zot_con.execute(
+        "SELECT i.itemID, i.key FROM items i "
+        "JOIN itemAttachments ia ON ia.itemID = i.itemID "
+        "WHERE ia.path LIKE 'storage:%.md'"
+    ).fetchall()
+
+    item_ids_to_delete: list[int] = []
+    for item_id, item_key in zot_rows:
+        d = zot_storage / item_key
+        if d.is_dir():
+            non_md = [p for p in d.iterdir() if not p.name.endswith(".md")]
+            if non_md:
+                log.warning("skip storage dir %s (contains non-md: %s)",
+                            d, [p.name for p in non_md])
+                stats["zot_dirs_skipped_nonempty"] += 1
+                # We still drop the rows — Zotero will show a missing
+                # attachment, easier to clean than a phantom row.
+            else:
+                shutil.rmtree(d, ignore_errors=True)
+                stats["zot_dirs_removed"] += 1
+        item_ids_to_delete.append(item_id)
+
+    if item_ids_to_delete:
+        placeholders = ",".join("?" * len(item_ids_to_delete))
+        zot_con.execute(
+            f"DELETE FROM itemAttachments WHERE itemID IN ({placeholders})",
+            item_ids_to_delete,
+        )
+        zot_con.execute(
+            f"DELETE FROM items WHERE itemID IN ({placeholders})",
+            item_ids_to_delete,
+        )
+        stats["zot_rows_deleted"] = len(item_ids_to_delete)
+    zot_con.commit()
+    zot_con.close()
+
+    return stats
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -130,8 +220,14 @@ def main() -> None:
         )
         return
 
-    # Phase C lands in Task 6.
+    log.info("Phase C: destructive cleanup of .md file attachments")
-    log.warning("--cleanup not yet implemented; landing in Task 6")
+    counts = _phase_c_preview()
+    log.info("Will delete: %s", counts)
+    if not _confirm("Proceed with destructive cleanup?"):
+        log.info("Aborted.")
+        return
+    stats_c = phase_c_cleanup()
+    log.info("Phase C complete: %s", stats_c)
 
 
 if __name__ == "__main__":