fix(scripts): harden Phase C dir deletion before live run

Code review surfaced two issues that matter once the migration runs
against real data, both around storage-dir removal:

1. The bib branch had no foreign-content guard — Zotero's branch
   inspected d.iterdir() for non-md files before rmtree, but bib's
   branch unconditionally rmtree'd the parent dir. bib's actual
   storage layout is one-attachment-per-dir so this is theoretically
   safe, but a defensive guard (matching the Zotero pattern) costs
   nothing and prevents silent data loss if the layout ever drifts.

2. Counters incremented unconditionally after `rmtree(d, ignore_errors
   =True)`. Silent rmtree failure would still bump bib_dirs_removed /
   zot_dirs_removed, masking the failure in the post-run stats.
   Switch to bare `rmtree` + try/except OSError, and only increment
   on the success path. Failures land in two new counters
   (bib_dirs_rmtree_failed, zot_dirs_rmtree_failed) so the operator
   sees them in the final log.

Surfaced by code review on commit c9b8a51.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
kert
2026-04-29 10:38:00 -04:00
parent c9b8a51f53
commit db7ea65287

View File

@@ -129,19 +129,36 @@ def phase_c_cleanup() -> dict[str, int]:
from conf import path from conf import path
stats = {"bib_rows_deleted": 0, "bib_dirs_removed": 0, stats = {"bib_rows_deleted": 0, "bib_dirs_removed": 0,
"bib_dirs_skipped_nonempty": 0, "bib_dirs_rmtree_failed": 0,
"zot_rows_deleted": 0, "zot_dirs_removed": 0, "zot_rows_deleted": 0, "zot_dirs_removed": 0,
"zot_dirs_skipped_nonempty": 0} "zot_dirs_skipped_nonempty": 0, "zot_dirs_rmtree_failed": 0}
# ── bib ────────────────────────────────────────── # ── bib ──────────────────────────────────────────
# bib's storage layout is one attachment per dir (att_key/<filename>),
# so the parent dir should contain only the .md we're deleting. Guard
# against unexpected siblings the same way the Zotero branch does.
bib_con = sqlite3.connect(str(path("db.bib"))) bib_con = sqlite3.connect(str(path("db.bib")))
bib_rows = bib_con.execute( bib_rows = bib_con.execute(
"SELECT id, storage_path FROM attachments WHERE filename LIKE '%.md'" "SELECT id, storage_path FROM attachments WHERE filename LIKE '%.md'"
).fetchall() ).fetchall()
for _att_id, storage_path in bib_rows: for _att_id, storage_path in bib_rows:
if storage_path: if not storage_path:
continue
d = Path(storage_path).parent d = Path(storage_path).parent
if d.is_dir(): if not d.is_dir():
shutil.rmtree(d, ignore_errors=True) continue
non_md = [p for p in d.iterdir() if not p.name.endswith(".md")]
if non_md:
log.warning("skip bib storage dir %s (contains non-md: %s)",
d, [p.name for p in non_md])
stats["bib_dirs_skipped_nonempty"] += 1
continue
try:
shutil.rmtree(d)
except OSError as e:
log.warning("rmtree failed for %s: %s", d, e)
stats["bib_dirs_rmtree_failed"] += 1
continue
stats["bib_dirs_removed"] += 1 stats["bib_dirs_removed"] += 1
bib_con.execute("DELETE FROM attachments WHERE filename LIKE '%.md'") bib_con.execute("DELETE FROM attachments WHERE filename LIKE '%.md'")
stats["bib_rows_deleted"] = bib_con.total_changes stats["bib_rows_deleted"] = bib_con.total_changes
@@ -164,14 +181,18 @@ def phase_c_cleanup() -> dict[str, int]:
if d.is_dir(): if d.is_dir():
non_md = [p for p in d.iterdir() if not p.name.endswith(".md")] non_md = [p for p in d.iterdir() if not p.name.endswith(".md")]
if non_md: if non_md:
log.warning("skip storage dir %s (contains non-md: %s)", log.warning("skip zot storage dir %s (contains non-md: %s)",
d, [p.name for p in non_md]) d, [p.name for p in non_md])
stats["zot_dirs_skipped_nonempty"] += 1 stats["zot_dirs_skipped_nonempty"] += 1
# We still drop the rows — Zotero will show a missing # We still drop the rows — Zotero will show a missing
# attachment, easier to clean than a phantom row. # attachment, easier to clean than a phantom row.
else: else:
shutil.rmtree(d, ignore_errors=True) try:
shutil.rmtree(d)
stats["zot_dirs_removed"] += 1 stats["zot_dirs_removed"] += 1
except OSError as e:
log.warning("rmtree failed for %s: %s", d, e)
stats["zot_dirs_rmtree_failed"] += 1
item_ids_to_delete.append(item_id) item_ids_to_delete.append(item_id)
if item_ids_to_delete: if item_ids_to_delete: