feat(zot): fetch.attach — storage copy, attachment row, source tag and md5 provenance (refs Z1)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

src/zot/fetch/attach.py

@@ -0,0 +1,74 @@
+"""Put a delivered file into Zotero storage and register it as a child.
+
+Generalises ``dev/scripts/add_ama_coding_to_zotero.py::_attach``: the
+storage tree is owned by the zotero container's uid (100999), so the
+default copier shells out to ``sudo``; tests inject a plain copier.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from collections.abc import Callable
+from pathlib import Path
+
+from zot.db import Db, generate_key
+
+FULL_TEXT_TYPES = {"application/pdf", "application/epub+zip"}
+CONTENT_TYPES = {".pdf": "application/pdf", ".epub": "application/epub+zip"}
+ZOTERO_UID_GID = "100999:100999"
+
+Copier = Callable[[Path, Path], None]
+
+
+def has_full_text(db: Db, item_id: int) -> bool:
+    """True when *item_id* already has a PDF or EPUB child attachment."""
+    rows = db.con.execute(
+        "SELECT contentType FROM itemAttachments WHERE parentItemID = ?", (item_id,)
+    ).fetchall()
+    return any((r[0] or "") in FULL_TEXT_TYPES for r in rows)
+
+
+def sudo_copier(src: Path, dest_dir: Path) -> None:
+    """Copy *src* into *dest_dir* as the zotero container's owner."""
+    subprocess.run(["sudo", "mkdir", "-p", str(dest_dir)], check=True)
+    subprocess.run(["sudo", "cp", str(src), str(dest_dir / src.name)], check=True)
+    subprocess.run(["sudo", "chown", "-R", ZOTERO_UID_GID, str(dest_dir)], check=True)
+
+
+def append_extra(db: Db, item_id: int, line: str) -> None:
+    """Append *line* to the item's ``extra`` field unless already present."""
+    existing = db.get_field(item_id, "extra") or ""
+    if line in existing.splitlines():
+        return
+    db.set_field(item_id, "extra", f"{existing.rstrip()}\n{line}".strip())
+
+
+def attach_file(
+    db: Db,
+    parent_id: int,
+    path: Path,
+    *,
+    source: str,
+    storage_dir: Path,
+    md5: str = "",
+    copier: Copier = sudo_copier,
+) -> int:
+    """Attach *path* to *parent_id*; returns the attachment ``itemID``.
+
+    Provenance: tag ``source:<source>`` on the parent and, when an md5
+    is known, an ``md5: <md5>`` line in ``extra``. Nothing else on the
+    parent changes.
+    """
+    ctype = CONTENT_TYPES.get(path.suffix.lower())
+    if ctype is None:
+        raise ValueError(f"unsupported attachment type: {path.suffix!r} ({path.name})")
+    att_key = generate_key()
+    copier(path, Path(storage_dir) / att_key)
+    att_id = db.add_attachment(
+        parent_id, key=att_key, content_type=ctype, path=f"storage:{path.name}"
+    )
+    db.set_field(att_id, "title", path.name)
+    db.tag_item(parent_id, f"source:{source}")
+    if md5:
+        append_extra(db, parent_id, f"md5: {md5}")
+    return att_id

tests/zot/fetch/test_attach.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from tests.zot.fetch.conftest import mk_item
+from zot.db import Db
+from zot.fetch.attach import (
+    CONTENT_TYPES,
+    append_extra,
+    attach_file,
+    has_full_text,
+    sudo_copier,
+)
+
+
+def _py_copier(src: Path, dest_dir: Path) -> None:
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    (dest_dir / src.name).write_bytes(src.read_bytes())
+
+
+def _att(db: Db, parent: int, ctype: str) -> None:
+    db.add_attachment(parent, content_type=ctype, path="storage:x")
+
+
+@pytest.mark.parametrize(
+    "ctypes, expected",
+    [
+        ([], False),
+        (["text/html"], False),
+        (["application/pdf"], True),
+        (["application/epub+zip"], True),
+        (["text/html", "application/pdf"], True),
+    ],
+)
+def test_has_full_text_truth_table(zdb: Db, ctypes, expected):
+    parent = mk_item(zdb, title="P")
+    for c in ctypes:
+        _att(zdb, parent, c)
+    assert has_full_text(zdb, parent) is expected
+
+
+def test_has_full_text_ignores_other_parents(zdb: Db):
+    a = mk_item(zdb, title="A")
+    b = mk_item(zdb, title="B")
+    _att(zdb, a, "application/pdf")
+    assert has_full_text(zdb, b) is False
+
+
+def test_attach_file_writes_row_tags_and_extra(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, title="Paper", extra="PMID: 1")
+    src = tmp_path / "Paper.pdf"
+    src.write_bytes(b"%PDF-1.4 x")
+    storage = tmp_path / "storage"
+    att_id = attach_file(
+        zdb,
+        parent,
+        src,
+        source="libgen",
+        md5="a" * 32,
+        storage_dir=storage,
+        copier=_py_copier,
+    )
+    row = zdb.con.execute(
+        "SELECT parentItemID, linkMode, contentType, path FROM itemAttachments "
+        "WHERE itemID = ?",
+        (att_id,),
+    ).fetchone()
+    assert tuple(row) == (parent, 0, "application/pdf", "storage:Paper.pdf")
+    assert zdb.get_field(att_id, "title") == "Paper.pdf"
+    att_key = zdb.get_item(att_id)["key"]
+    assert (storage / att_key / "Paper.pdf").read_bytes() == b"%PDF-1.4 x"
+    assert "source:libgen" in zdb.get_tags(parent)
+    assert zdb.get_field(parent, "extra") == "PMID: 1\nmd5: " + "a" * 32
+    assert has_full_text(zdb, parent) is True
+
+
+def test_attach_file_epub_content_type(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, type_name="book", title="Book")
+    src = tmp_path / "Book.epub"
+    src.write_bytes(b"PK\x03\x04")
+    att_id = attach_file(
+        zdb, parent, src, source="annas", storage_dir=tmp_path / "s", copier=_py_copier
+    )
+    ctype = zdb.con.execute(
+        "SELECT contentType FROM itemAttachments WHERE itemID = ?", (att_id,)
+    ).fetchone()[0]
+    assert ctype == CONTENT_TYPES[".epub"]
+    assert zdb.get_field(parent, "extra") is None  # no md5 → extra untouched
+
+
+def test_attach_file_rejects_unknown_extension(zdb: Db, tmp_path: Path):
+    parent = mk_item(zdb, title="P")
+    src = tmp_path / "x.mobi"
+    src.write_bytes(b"x")
+    with pytest.raises(ValueError, match="mobi"):
+        attach_file(
+            zdb, parent, src, source="libgen", storage_dir=tmp_path, copier=_py_copier
+        )
+
+
+def test_append_extra_is_idempotent_and_keeps_lines(zdb: Db):
+    item = mk_item(zdb, title="P", extra="PMID: 1")
+    append_extra(zdb, item, "md5: abc")
+    append_extra(zdb, item, "md5: abc")
+    assert zdb.get_field(item, "extra") == "PMID: 1\nmd5: abc"
+
+
+def test_append_extra_on_empty_extra(zdb: Db):
+    item = mk_item(zdb, title="P")
+    append_extra(zdb, item, "md5: abc")
+    assert zdb.get_field(item, "extra") == "md5: abc"
+
+
+def test_sudo_copier_runs_mkdir_cp_chown(monkeypatch, tmp_path: Path):
+    calls: list[list[str]] = []
+
+    def fake_run(cmd, check):
+        calls.append(cmd)
+
+    monkeypatch.setattr("zot.fetch.attach.subprocess.run", fake_run)
+    sudo_copier(tmp_path / "a.pdf", tmp_path / "storage" / "KEY")
+    dest = tmp_path / "storage" / "KEY"
+    assert calls == [
+        ["sudo", "mkdir", "-p", str(dest)],
+        ["sudo", "cp", str(tmp_path / "a.pdf"), str(dest / "a.pdf")],
+        ["sudo", "chown", "-R", "100999:100999", str(dest)],
+    ]