stack/dev/scripts/dedupe_attachments.py

#!/usr/bin/env python3
"""One-time cleanup of duplicate bib attachments (refs #615).

``Store.attach_file`` used to mint a new row + storage copy on every
call, so re-farms produced thousands of ``(item_id, filename)``
duplicates (33,379 groups / 66,917 of 78,017 rows on 2026-09-08). This
keeps the oldest row of each group, deletes the others' rows and their
storage copies, and prints a report. Dry-run by default.

Zotero is not touched: ``bib.sync`` already dedupes child attachments by
filename, so a duplicate that was synced once is a single Zotero child
attachment and stays valid.

Usage::

    uv run python dev/scripts/dedupe_attachments.py            # report only
    uv run python dev/scripts/dedupe_attachments.py --apply    # delete
"""

from __future__ import annotations

import argparse
import os
import sqlite3
import sys
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class Group:
    item_id: int
    filename: str
    keep: str
    remove: list[str] = field(default_factory=list)
    remove_paths: list[str] = field(default_factory=list)
    conflict: bool = False  # sizes differ → do not touch


@dataclass
class Report:
    groups: int = 0
    rows_removed: int = 0
    files_removed: int = 0
    bytes_freed: int = 0
    skipped_conflicts: int = 0
    removed_keys: list[str] = field(default_factory=list)


def plan(con: sqlite3.Connection) -> list[Group]:
    con.row_factory = sqlite3.Row
    rows = con.execute(
        """
        SELECT a.item_id, a.filename, a.key, a.storage_path, a.rowid AS rid
        FROM attachments a
        WHERE (a.item_id, a.filename) IN (
            SELECT item_id, filename FROM attachments
            GROUP BY item_id, filename HAVING count(*) > 1
        )
        ORDER BY a.item_id, a.filename, a.rowid
        """
    ).fetchall()
    groups: dict[tuple[int, str], Group] = {}
    for r in rows:
        gkey = (r["item_id"], r["filename"])
        g = groups.get(gkey)
        if g is None:
            groups[gkey] = Group(
                item_id=r["item_id"], filename=r["filename"], keep=r["key"]
            )
            continue
        g.remove.append(r["key"])
        g.remove_paths.append(r["storage_path"])
    # conflict check: every copy must have the same size as the kept one
    for g in groups.values():
        keep_path = con.execute(
            "SELECT storage_path FROM attachments WHERE key = ?", (g.keep,)
        ).fetchone()["storage_path"]
        keep_size = _size(keep_path)
        for p in g.remove_paths:
            if _size(p) not in (keep_size, -1):
                g.conflict = True
                break
    return list(groups.values())


def _size(path: str) -> int:
    try:
        return os.stat(path).st_size
    except OSError:
        return -1


def apply(con: sqlite3.Connection, groups: list[Group]) -> Report:
    rep = Report(groups=len(groups))
    con.execute("BEGIN")
    try:
        for g in groups:
            if g.conflict:
                rep.skipped_conflicts += 1
                continue
            for key, path in zip(g.remove, g.remove_paths):
                still_referenced = con.execute(
                    "SELECT count(*) FROM attachments WHERE storage_path = ? AND key <> ?",
                    (path, key),
                ).fetchone()[0]
                con.execute("DELETE FROM attachments WHERE key = ?", (key,))
                rep.rows_removed += 1
                rep.removed_keys.append(key)
                if not still_referenced:
                    p = Path(path)
                    if p.is_file():
                        rep.bytes_freed += p.stat().st_size
                        p.unlink()
                        rep.files_removed += 1
                        try:
                            p.parent.rmdir()  # the per-key dir, if now empty
                        except OSError:
                            pass
        con.execute("COMMIT")
    except Exception:
        con.execute("ROLLBACK")
        raise
    return rep


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    ap.add_argument(
        "--db", default="", help="bib.sqlite path (default: stack.toml db.bib)"
    )
    ap.add_argument(
        "--apply", action="store_true", help="delete duplicates (default: report only)"
    )
    args = ap.parse_args(argv)
    if args.db:
        db = args.db
    else:
        sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
        from conf import path

        db = str(path("db.bib"))
    con = sqlite3.connect(db, isolation_level=None)
    groups = plan(con)
    conflicts = sum(1 for g in groups if g.conflict)
    rows = sum(len(g.remove) for g in groups)
    print(
        f"{db}: {len(groups)} duplicate groups, {rows} rows to remove, {conflicts} conflicts (size mismatch, skipped)"
    )
    if not args.apply:
        print("dry run — pass --apply to delete")
        return 0
    rep = apply(con, groups)
    print(
        f"removed rows={rep.rows_removed} files={rep.files_removed} "
        f"freed={rep.bytes_freed / 1e6:.1f} MB skipped_conflicts={rep.skipped_conflicts}"
    )
    print("reopen the store once (any `stack bib` command) to create the unique index")
    return 0


if __name__ == "__main__":
    sys.exit(main())