From 3615f6e781a6412e6c0ec61cc4ffd01616954403 Mon Sep 17 00:00:00 2001 From: kert Date: Wed, 23 Sep 2026 20:58:52 -0400 Subject: [PATCH] =?UTF-8?q?docs(plan):=20Z1=20of=20the=20Zotero=20full-tex?= =?UTF-8?q?t=20connector=20=E2=80=94=20stack=20zot=20fetch=20(spec=202026-?= =?UTF-8?q?09-09)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- .../plans/2026-09-22-zotero-fetch-z1.md | 2578 +++++++++++++++++ 1 file changed, 2578 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md diff --git a/docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md b/docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md new file mode 100644 index 0000000..bb18dac --- /dev/null +++ b/docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md @@ -0,0 +1,2578 @@ +# Zotero full-text connector — Z1 (`stack zot fetch`) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** `stack zot fetch` resolves Zotero items that lack a PDF/EPUB to downloadable candidates (Unpaywall, libgen JSON index), delivers a file (libgen `ads→get`, then Anna's fast download by md5), attaches it to the item, and writes a ledger line for every attempt so misses are not retried forever. + +**Architecture:** A new package `src/zot/fetch/` with one module per stage — `select` (which items), `resolve` (identifiers → `Candidate` list), `deliver` (candidate → file on disk, quota-aware), `attach` (file → Zotero storage + attachment row + provenance), `ledger` (JSONL of attempts), `run` (the per-item loop) — and a `fetch` verb in `src/cli/zot.py` under the existing `--hold` guard. Discovery and delivery go through the gluetun HTTP proxy; Unpaywall goes direct. All HTTP is `httpx` and every test uses `httpx.MockTransport`; no network in tests. + +**Tech Stack:** Python 3.13, httpx ≥ 0.28 (already a dependency), typer, sqlite3 via `zot.db.Db`, pytest (CI gate: `--cov-fail-under=99`), ruff (E, F, I; line length 88). + +**Spec:** `docs/superpowers/specs/2026-09-09-zotero-annas-connector-design.md` (Z1 slice). Approved for build by the operator 2026-09-22. + +## Global Constraints + +- Coverage gate is 99% over `src/` (`.gitea/workflows/ci.yml:62`). Every branch you write needs a test, including error branches. +- Ruff: `select = ["E", "F", "I"]`, line length 88. Run `uv run ruff check src tests && uv run ruff format src tests` before every commit. +- libgen and Anna's traffic **must** go through `VPN_HTTP_PROXY` (`http://127.0.0.1:19035`). A missing proxy is an error naming the compose step, never a fallback to direct. Unpaywall may go direct. +- libgen answers a bare nginx page unless the request carries a browser `User-Agent` (verified 2026-09-22). The proxied client always sends `BROWSER_UA`. +- Pacing: libgen queries 3 s apart, Unpaywall 1 req/s. Pacing must be injectable (a `sleep` callable) so tests never sleep. +- Anna's daily budget guard: `ANNAS_DAILY_RESERVE` (default 5) kept back. Budget is unknown until the first Anna's response (the API only reports `account_fast_download_info` on a successful download call). +- Ledger path `data/zotero/fetch_ledger.jsonl` (gitignored by `data/*`). Terminal misses (`no_candidate`, `all_failed`) retry only after 30 days or with `--retry`. +- Never edit a parent item beyond tags (`source:unpaywall|libgen|annas`) and an appended `md5: ` line in `extra`. +- Always excluded from selection: items that already have a PDF/EPUB child, items in `deletedItems`, and the `document` item type. +- Zotero storage files are owned by uid/gid `100999:100999`; the copy uses `sudo mkdir/cp/chown` exactly as `dev/scripts/add_ama_coding_to_zotero.py:_attach` does. The copier is injectable so tests copy with plain Python. +- `stack` does not load `.env`. Live runs are `set -a; . ./.env; set +a; uv run stack zot fetch …` from `/home/kert/stack` (never from a worktree — `conf.path` resolves `data/zotero` relative to the checkout). +- Commit messages: end with `Co-Authored-By: Claude Fable 5.1 ` per the session's attribution rule. Before every commit run `git status --short` and stage only your files (other sessions share this checkout). + +## Review Focus + +Inputs the spec implies but no acceptance criterion exercises. Each has a test pinned to the owning task. + +1. **libgen `ads.php` returns HTTP 200 with an error div and no `get.php` link** (seen live: "Could not connect to the database 3306 … max_user_connections"). Expected: treated as a throttle — one retry after 3 s, then the next candidate. → Task 5 `test_libgen_ads_error_div_retries_once_then_gives_up`. +2. **A DOI stored as a URL** (`https://doi.org/10.1056/NEJMoa1000678`) or with surrounding whitespace. Expected: normalised to the bare DOI before any lookup. → Task 3 `test_doi_normalised_from_url_form`. +3. **An ISBN field holding several ISBNs with hyphens** (`978-1-64016-228-0 1640162283`). Expected: the first 13-digit ISBN wins, digits only. → Task 3 `test_isbn_prefers_13_digit_and_strips_hyphens`. +4. **The libgen file lookup (`object=f`) fails or omits an entry** so a candidate has no extension. Expected: papers default to `pdf` (the scimag corpus), books with unknown extension are dropped, run continues. → Task 4 `test_libgen_file_lookup_missing_defaults_paper_to_pdf_drops_book`. +5. **Anna's reports fewer downloads left than the reserve mid-run.** Expected: Anna's is skipped for the rest of the run, libgen-only delivery continues, and items whose only remaining route was Anna's are ledgered `quota`, not `all_failed`. → Task 5 `test_budget_reserve_blocks_annas` and Task 6 `test_run_ledgers_quota_when_only_annas_was_left`. + +--- + +## File map + +| Path | Responsibility | +|---|---| +| `src/zot/fetch/__init__.py` | re-exports (`Candidate`, `ItemRef`, `Ledger`, `Budget`, `run`) | +| `src/zot/fetch/ledger.py` | `Ledger`: JSONL attempts, 30-day retry rule, run summary | +| `src/zot/fetch/attach.py` | `has_full_text`, `attach_file`, `sudo_copier`, content types | +| `src/zot/fetch/select.py` | `ItemRef`, `select_items`, DOI/ISBN/year normalisation | +| `src/zot/fetch/resolve.py` | `Candidate`, `Pacer`, `BROWSER_UA`, Unpaywall + libgen JSON + title search | +| `src/zot/fetch/deliver.py` | `Budget`, `AnnasAuthError`, `fetch` (libgen ads→get, Anna's fast download), file validation | +| `src/zot/fetch/run.py` | `make_clients`, `run` (per-item loop, dry-run, stats) | +| `src/cli/zot.py` | `@app.command("fetch")` | +| `tests/zot/fetch/conftest.py` | scratch Zotero DB fixture with `deletedItems`, `mk_item` helper | +| `tests/zot/fetch/fixtures/*.json|*.html` | pinned API shapes captured through the proxy 2026-09-22 | +| `tests/zot/fetch/test_{ledger,attach,select,resolve,deliver,run}.py` | unit tests | +| `tests/cli/test_zot_fetch.py` | CLI test with `CliRunner`, `--no-hold`, `--dry-run` | +| `docs/docs/cli/zot-fetch.md`, `docs/docs/cli/zot.md` | generated by `docs/scripts/extract_cli.py` | + +--- + +### Task 1: Test scaffold + ledger + +**Files:** +- Create: `src/zot/fetch/__init__.py`, `src/zot/fetch/ledger.py` +- Create: `tests/zot/fetch/__init__.py`, `tests/zot/fetch/conftest.py`, `tests/zot/fetch/test_ledger.py` + +**Interfaces:** +- Produces: `Ledger(path: Path)`, `Ledger.record(item_key, *, outcome, source="", md5="", ids=None, note="", now=None) -> Attempt`, `Ledger.should_try(item_key, *, retry=False, now=None) -> bool`, `Ledger.summary() -> dict[str, int]`, `Attempt` dataclass, `TERMINAL = {"no_candidate", "all_failed"}`, `RETRY_AFTER = timedelta(days=30)`. +- Produces (tests): `zdb` fixture (a `zot.db.Db` on a scratch SQLite built from `tests.zot.test_db.ZOTERO_SCHEMA` plus a `deletedItems` table) and `mk_item(db, *, type_name="journalArticle", title="", doi="", isbn="", date="", extra="") -> int`. + +- [ ] **Step 1: Create the package and the test scaffold** + +`src/zot/fetch/__init__.py`: + +```python +"""Full-text connector for Zotero items (spec 2026-09-09, slice Z1). + +select → resolve → deliver → attach, with a JSONL ledger of attempts. +""" +``` + +`tests/zot/fetch/__init__.py`: empty file. + +`tests/zot/fetch/conftest.py`: + +```python +"""Scratch Zotero DB for the fetch tests (no live DB, no network).""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from tests.zot.test_db import ZOTERO_SCHEMA, _seed_schema_maps +from zot.db import TYPE_MAP, Db + +EXTRA_SCHEMA = """ +CREATE TABLE IF NOT EXISTS deletedItems ( + itemID INTEGER PRIMARY KEY, + dateDeleted TEXT DEFAULT CURRENT_TIMESTAMP NOT NULL +); +""" + + +@pytest.fixture() +def zdb(tmp_path: Path) -> Db: + path = str(tmp_path / "zotero.sqlite") + con = sqlite3.connect(path) + con.executescript(ZOTERO_SCHEMA) + con.executescript(EXTRA_SCHEMA) + _seed_schema_maps(con) + con.commit() + con.close() + d = Db(path) + yield d + d.close() + + +def mk_item( + db: Db, + *, + type_name: str = "journalArticle", + title: str = "", + doi: str = "", + isbn: str = "", + date: str = "", + extra: str = "", +) -> int: + item_id = db.create_item(TYPE_MAP[type_name]) + db.set_fields( + item_id, + {"title": title, "DOI": doi, "ISBN": isbn, "date": date, "extra": extra}, + ) + return item_id +``` + +- [ ] **Step 2: Write the failing ledger tests** + +`tests/zot/fetch/test_ledger.py`: + +```python +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from zot.fetch.ledger import RETRY_AFTER, TERMINAL, Attempt, Ledger + +T0 = datetime(2026, 9, 22, 12, 0, tzinfo=timezone.utc) + + +def test_record_appends_one_json_line(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + att = led.record( + "ABCD2345", outcome="attached", source="libgen", md5="a" * 32, + ids={"doi": "10.1/x"}, now=T0, + ) + assert isinstance(att, Attempt) + lines = (tmp_path / "ledger.jsonl").read_text().splitlines() + assert len(lines) == 1 + row = json.loads(lines[0]) + assert row == { + "item_key": "ABCD2345", "outcome": "attached", "source": "libgen", + "md5": "a" * 32, "ids": {"doi": "10.1/x"}, "note": "", + "ts": "2026-09-22T12:00:00+00:00", + } + + +def test_unknown_item_should_try(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + assert led.should_try("NEW00000") is True + + +def test_attached_item_never_retried(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + led.record("K1", outcome="attached", source="unpaywall", now=T0) + assert led.should_try("K1", now=T0 + timedelta(days=400)) is False + + +def test_terminal_miss_waits_thirty_days(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + for outcome in TERMINAL: + led.record("K1", outcome=outcome, now=T0) + assert led.should_try("K1", now=T0 + RETRY_AFTER - timedelta(seconds=1)) is False + assert led.should_try("K1", now=T0 + RETRY_AFTER) is True + + +def test_retry_flag_overrides_terminal_miss(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + led.record("K1", outcome="no_candidate", now=T0) + assert led.should_try("K1", retry=True, now=T0) is True + + +def test_quota_miss_retries_immediately(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + led.record("K1", outcome="quota", now=T0) + assert led.should_try("K1", now=T0) is True + + +def test_ledger_reloads_last_attempt_per_key(tmp_path: Path): + p = tmp_path / "ledger.jsonl" + Ledger(p).record("K1", outcome="no_candidate", now=T0) + Ledger(p).record("K1", outcome="attached", now=T0) + led = Ledger(p) + assert led.should_try("K1", now=T0 + timedelta(days=60)) is False + + +def test_ledger_skips_corrupt_lines(tmp_path: Path): + p = tmp_path / "ledger.jsonl" + p.write_text('{"item_key": "K1", "outcome": "attached", "ts": "2026-09-22T12:00:00+00:00"}\nnot json\n') + led = Ledger(p) + assert led.should_try("K1") is False + assert led.should_try("K2") is True + + +def test_summary_counts_this_session_only(tmp_path: Path): + p = tmp_path / "ledger.jsonl" + Ledger(p).record("K0", outcome="attached", now=T0) + led = Ledger(p) + led.record("K1", outcome="attached", now=T0) + led.record("K2", outcome="no_candidate", now=T0) + led.record("K3", outcome="no_candidate", now=T0) + assert led.summary() == {"attached": 1, "no_candidate": 2} + + +def test_default_now_is_utc_iso(tmp_path: Path): + led = Ledger(tmp_path / "ledger.jsonl") + att = led.record("K1", outcome="attached") + assert att.ts.endswith("+00:00") +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_ledger.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.ledger'` + +- [ ] **Step 4: Implement the ledger** + +`src/zot/fetch/ledger.py`: + +```python +"""JSONL ledger of fetch attempts — the report the operator reads. + +One line per attempt. ``should_try`` keeps a terminal miss +(``no_candidate`` / ``all_failed``) from being retried for 30 days, +never retries an ``attached`` item, and always retries ``quota``. +""" + +from __future__ import annotations + +import json +from collections import Counter +from dataclasses import asdict, dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path + +TERMINAL = {"no_candidate", "all_failed"} +RETRY_AFTER = timedelta(days=30) + + +@dataclass(frozen=True) +class Attempt: + item_key: str + outcome: str + source: str = "" + md5: str = "" + ids: dict[str, str] = field(default_factory=dict) + note: str = "" + ts: str = "" + + @property + def when(self) -> datetime: + return datetime.fromisoformat(self.ts) + + +def _now_iso(now: datetime | None) -> str: + return (now or datetime.now(timezone.utc)).isoformat() + + +class Ledger: + def __init__(self, path: Path) -> None: + self.path = Path(path) + self._last: dict[str, Attempt] = {} + self._session: list[Attempt] = [] + if self.path.is_file(): + for line in self.path.read_text().splitlines(): + try: + row = json.loads(line) + att = Attempt(**row) + except (ValueError, TypeError): + continue + self._last[att.item_key] = att + + def record( + self, + item_key: str, + *, + outcome: str, + source: str = "", + md5: str = "", + ids: dict[str, str] | None = None, + note: str = "", + now: datetime | None = None, + ) -> Attempt: + att = Attempt( + item_key=item_key, + outcome=outcome, + source=source, + md5=md5, + ids=dict(ids or {}), + note=note, + ts=_now_iso(now), + ) + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(asdict(att), ensure_ascii=False) + "\n") + self._last[item_key] = att + self._session.append(att) + return att + + def should_try( + self, item_key: str, *, retry: bool = False, now: datetime | None = None + ) -> bool: + if retry: + return True + last = self._last.get(item_key) + if last is None: + return True + if last.outcome == "attached": + return False + if last.outcome in TERMINAL: + current = now or datetime.now(timezone.utc) + return current - last.when >= RETRY_AFTER + return True + + def summary(self) -> dict[str, int]: + return dict(Counter(a.outcome for a in self._session)) +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch/test_ledger.py -q` +Expected: 10 passed + +- [ ] **Step 6: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/__init__.py src/zot/fetch/ledger.py tests/zot/fetch/ +git commit -m "feat(zot): fetch ledger — JSONL attempts with the 30-day retry rule (refs spec 2026-09-09 Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 2: Attach + +**Files:** +- Create: `src/zot/fetch/attach.py` +- Test: `tests/zot/fetch/test_attach.py` + +**Interfaces:** +- Consumes: `zot.db.Db` (`add_attachment`, `set_field`, `get_field`, `tag_item`, `generate_key`), `zdb`/`mk_item` from Task 1. +- Produces: `FULL_TEXT_TYPES`, `CONTENT_TYPES`, `has_full_text(db, item_id) -> bool`, `Copier = Callable[[Path, Path], None]`, `sudo_copier(src, dest_dir)`, `attach_file(db, parent_id, path, *, source, storage_dir, md5="", copier=sudo_copier) -> int`, `append_extra(db, item_id, line) -> None`. + +- [ ] **Step 1: Write the failing tests** + +`tests/zot/fetch/test_attach.py`: + +```python +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): # noqa: ARG001 + calls.append(cmd) + + monkeypatch.setattr("zot.fetch.attach.subprocess.run", fake_run) + sudo_copier(tmp_path / "a.pdf", tmp_path / "storage" / "KEY") + assert calls == [ + ["sudo", "mkdir", "-p", str(tmp_path / "storage" / "KEY")], + ["sudo", "cp", str(tmp_path / "a.pdf"), str(tmp_path / "storage" / "KEY" / "a.pdf")], + ["sudo", "chown", "-R", "100999:100999", str(tmp_path / "storage" / "KEY")], + ] +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_attach.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.attach'` + +- [ ] **Step 3: Implement attach** + +`src/zot/fetch/attach.py`: + +```python +"""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:`` on the parent and, when an md5 + is known, an ``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 +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch/test_attach.py -q` +Expected: 11 passed + +- [ ] **Step 5: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/attach.py tests/zot/fetch/test_attach.py +git commit -m "feat(zot): fetch.attach — storage copy, attachment row, source tag and md5 provenance (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 3: Select + +**Files:** +- Create: `src/zot/fetch/select.py` +- Test: `tests/zot/fetch/test_select.py` + +**Interfaces:** +- Consumes: `has_full_text` (Task 2), `zot.db.Db.search_by_tag / search_by_type / search_by_collection / get_field / get_item`, `TYPE_MAP`. +- Produces: `ItemRef(item_id: int, key: str, item_type: str, title: str, doi: str, isbn: str, year: str)` with property `kind -> "book" | "paper"`; `select_items(db, *, collection="", tag="", type_="", item="", all_=False) -> list[ItemRef]`; `normalise_doi(raw) -> str`; `first_isbn(raw) -> str`; `year_of(raw) -> str`; `item_ref(db, item_id) -> ItemRef`. + +- [ ] **Step 1: Write the failing tests** + +`tests/zot/fetch/test_select.py`: + +```python +from __future__ import annotations + +import pytest + +from tests.zot.fetch.conftest import mk_item +from zot.db import Db +from zot.fetch.select import ( + ItemRef, + first_isbn, + item_ref, + normalise_doi, + select_items, + year_of, +) + + +@pytest.mark.parametrize( + "raw, want", + [ + ("10.1056/NEJMoa1000678", "10.1056/NEJMoa1000678"), + (" https://doi.org/10.1056/NEJMoa1000678 ", "10.1056/NEJMoa1000678"), + ("http://dx.doi.org/10.1/ab", "10.1/ab"), + ("doi:10.1/ab", "10.1/ab"), + ("", ""), + ], +) +def test_doi_normalised_from_url_form(raw, want): + assert normalise_doi(raw) == want + + +@pytest.mark.parametrize( + "raw, want", + [ + ("978-1-64016-228-0 1640162283", "9781640162280"), + ("1640162283", "1640162283"), + ("ISBN 0-19-852663-6", "0198526636"), + ("no isbn here", ""), + ("", ""), + ], +) +def test_isbn_prefers_13_digit_and_strips_hyphens(raw, want): + assert first_isbn(raw) == want + + +@pytest.mark.parametrize("raw, want", [("2010-08-19", "2010"), ("August 2010", "2010"), ("", "")]) +def test_year_of(raw, want): + assert year_of(raw) == want + + +def test_item_ref_reads_fields_and_kind(zdb: Db): + paper = mk_item(zdb, title="P", doi="https://doi.org/10.1/x", date="2020-01-01") + book = mk_item(zdb, type_name="book", title="B", isbn="978-1-64016-228-0") + p = item_ref(zdb, paper) + b = item_ref(zdb, book) + assert p == ItemRef(paper, zdb.get_item(paper)["key"], "journalArticle", "P", "10.1/x", "", "2020") + assert p.kind == "paper" + assert b.isbn == "9781640162280" and b.kind == "book" + + +def test_select_requires_exactly_one_primary_selector(zdb: Db): + with pytest.raises(ValueError, match="exactly one"): + select_items(zdb) + with pytest.raises(ValueError, match="exactly one"): + select_items(zdb, tag="t", all_=True) + + +def test_select_by_tag_type_item_collection_all(zdb: Db): + a = mk_item(zdb, title="A", doi="10.1/a") + b = mk_item(zdb, type_name="book", title="B", isbn="9781640162280") + zdb.tag_item(a, "rfi:codeset") + ckey = zdb.ensure_collection("AMA Coding Publications") + zdb.add_to_collection(b, collection_key=ckey) + assert [r.item_id for r in select_items(zdb, tag="rfi:codeset")] == [a] + assert [r.item_id for r in select_items(zdb, type_="book")] == [b] + assert [r.item_id for r in select_items(zdb, item=zdb.get_item(a)["key"])] == [a] + assert [r.item_id for r in select_items(zdb, collection="AMA Coding Publications")] == [b] + assert [r.item_id for r in select_items(zdb, all_=True)] == [a, b] + + +def test_type_filter_combines_with_primary_selector(zdb: Db): + a = mk_item(zdb, title="A") + b = mk_item(zdb, type_name="book", title="B") + ckey = zdb.ensure_collection("C") + zdb.add_to_collection(a, collection_key=ckey) + zdb.add_to_collection(b, collection_key=ckey) + assert [r.item_id for r in select_items(zdb, collection="C", type_="book")] == [b] + + +def test_unknown_collection_or_item_or_type_raises(zdb: Db): + with pytest.raises(ValueError, match="no collection named"): + select_items(zdb, collection="Nope") + with pytest.raises(ValueError, match="no item with key"): + select_items(zdb, item="ZZZZZZZZ") + with pytest.raises(ValueError, match="unknown item type"): + select_items(zdb, type_="novel") + + +def test_select_excludes_full_text_deleted_document_and_children(zdb: Db): + has_pdf = mk_item(zdb, title="has pdf") + zdb.add_attachment(has_pdf, content_type="application/pdf", path="storage:x.pdf") + deleted = mk_item(zdb, title="deleted") + zdb.con.execute("INSERT INTO deletedItems (itemID) VALUES (?)", (deleted,)) + mk_item(zdb, type_name="document", title="reg pdf") + keep = mk_item(zdb, title="keep") + got = [r.item_id for r in select_items(zdb, all_=True)] + assert got == [keep] + + +def test_select_all_excludes_document_even_when_typed(zdb: Db): + mk_item(zdb, type_name="document", title="reg pdf") + assert select_items(zdb, type_="document") == [] +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_select.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.select'` + +- [ ] **Step 3: Implement select** + +`src/zot/fetch/select.py`: + +```python +"""Which Zotero items a fetch run looks at, and their identifiers.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from zot.db import TYPE_MAP, Db +from zot.fetch.attach import has_full_text + +BOOK_TYPES = {"book", "bookSection"} +NEVER = {TYPE_MAP["attachment"], TYPE_MAP["note"], TYPE_MAP["annotation"], TYPE_MAP["document"]} + +_DOI_PREFIX = re.compile(r"^(?:https?://(?:dx\.)?doi\.org/|doi:)\s*", re.I) +_ISBN_TOKEN = re.compile(r"(? str: + return "book" if self.item_type in BOOK_TYPES else "paper" + + @property + def ids(self) -> dict[str, str]: + return {k: v for k, v in (("doi", self.doi), ("isbn", self.isbn)) if v} + + +def normalise_doi(raw: str) -> str: + return _DOI_PREFIX.sub("", (raw or "").strip()).strip() + + +def first_isbn(raw: str) -> str: + """First ISBN in *raw*, digits only; a 13-digit one wins over a 10.""" + flat = (raw or "").replace("-", "").upper() + tokens = _ISBN_TOKEN.findall(flat) + for t in tokens: + if len(t) == 13: + return t + return tokens[0] if tokens else "" + + +def year_of(raw: str) -> str: + m = _YEAR.search(raw or "") + return m.group(0) if m else "" + + +def item_ref(db: Db, item_id: int) -> ItemRef: + item = db.get_item(item_id) + f = item["fields"] + return ItemRef( + item_id=item_id, + key=item["key"], + item_type=item["itemType"], + title=f.get("title", "") or "", + doi=normalise_doi(f.get("DOI", "")), + isbn=first_isbn(f.get("ISBN", "")), + year=year_of(f.get("date", "")), + ) + + +def _collection_keys(db: Db, name: str) -> list[str]: + rows = db.con.execute( + "SELECT key FROM collections WHERE collectionName = ? ORDER BY collectionID", + (name,), + ).fetchall() + return [r[0] for r in rows] + + +def _all_ids(db: Db) -> list[int]: + marks = ",".join("?" * len(NEVER)) + rows = db.con.execute( + f"SELECT itemID FROM items WHERE itemTypeID NOT IN ({marks}) ORDER BY itemID", # noqa: S608 + tuple(NEVER), + ).fetchall() + return [r[0] for r in rows] + + +def _deleted(db: Db) -> set[int]: + return {r[0] for r in db.con.execute("SELECT itemID FROM deletedItems").fetchall()} + + +def select_items( + db: Db, + *, + collection: str = "", + tag: str = "", + type_: str = "", + item: str = "", + all_: bool = False, +) -> list[ItemRef]: + """Items to try, in itemID order. + + Exactly one of *collection* (by name), *tag*, *item* (key) or *all_* + picks the population; *type_* narrows it. Items with a PDF/EPUB + child, deleted items, and ``document`` items are always excluded. + """ + primaries = [bool(collection), bool(tag), bool(item), all_] + if sum(primaries) != 1: + raise ValueError("exactly one of --collection/--tag/--item/--all is required") + if type_ and type_ not in TYPE_MAP: + raise ValueError(f"unknown item type {type_!r}") + + if collection: + keys = _collection_keys(db, collection) + if not keys: + raise ValueError(f"no collection named {collection!r}") + ids: list[int] = sorted({i for k in keys for i in db.search_by_collection(k)}) + elif tag: + ids = db.search_by_tag(tag) + elif item: + found = db.find_item_by_key(item) + if found is None: + raise ValueError(f"no item with key {item!r}") + ids = [found] + else: + ids = _all_ids(db) + + if type_: + wanted = TYPE_MAP[type_] + ids = [i for i in ids if db.get_item_type(i) == wanted] + + deleted = _deleted(db) + out: list[ItemRef] = [] + for i in ids: + if i in deleted or db.get_item_type(i) in NEVER or has_full_text(db, i): + continue + out.append(item_ref(db, i)) + return out +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch/test_select.py -q` +Expected: all passed (20 cases) + +- [ ] **Step 5: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/select.py tests/zot/fetch/test_select.py +git commit -m "feat(zot): fetch.select — item population by collection/tag/type/item/all with the standing exclusions (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 4: Resolve + +**Files:** +- Create: `src/zot/fetch/resolve.py` +- Create: `tests/zot/fetch/fixtures/unpaywall_doi.json`, `tests/zot/fetch/fixtures/libgen_json_doi.json`, `tests/zot/fetch/fixtures/libgen_json_isbn.json`, `tests/zot/fetch/fixtures/libgen_json_file.json`, `tests/zot/fetch/fixtures/libgen_index.html` +- Test: `tests/zot/fetch/test_resolve.py` + +**Interfaces:** +- Consumes: `ItemRef` (Task 3), `prisma.fetch.unpaywall_urls(client, doi, email) -> list[str]` (existing, reused as-is). +- Produces: `BROWSER_UA`, `LIBGEN = "https://libgen.li"`, `Candidate(source, md5, url, ext, size_bytes, title)`, `Pacer(min_interval, *, clock=time.monotonic, sleep=time.sleep)` with `.wait()`, `libgen_editions(client, *, doi="", isbn="", ids=(), pacer) -> dict`, `libgen_files(client, f_ids, pacer) -> dict`, `libgen_candidates(client, *, kind, doi="", isbn="", pacer) -> list[Candidate]`, `title_candidates(client, *, title, year, pacer) -> list[Candidate]`, `title_matches(target, year, entries) -> list[dict]`, `candidates(ref, *, direct, proxied, email, up_pacer, lg_pacer) -> list[Candidate]`. + +- [ ] **Step 1: Write the fixtures** + +These are the shapes captured through the proxy on 2026-09-22 (trimmed). `tests/zot/fetch/fixtures/libgen_json_doi.json`: + +```json +{"27703795": {"title": "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer", "libgen_topic": "a", "type": "a", "author": "Temel, Jennifer S.; Greer, Joseph A.", "year": "2010", "doi": "10.1056/NEJMoa1000678", "files": {"26575841": {"f_id": "27858481", "md5": "a78add1bd62047ead5945134488f470f", "time_added": "2021-07-17 13:59:28"}, "26575842": {"f_id": "27858482", "md5": "b78add1bd62047ead5945134488f470f", "time_added": "2021-07-17 13:59:28"}}, "add": {"68149362": {"key": "904", "value": "20818875", "name_en": "PMID"}}}} +``` + +`tests/zot/fetch/fixtures/libgen_json_isbn.json`: + +```json +{"146962484": {"title": "HCPCS Level II Professional 2023", "libgen_topic": "l", "type": "b", "year": "2023", "edition": "1", "files": {"103379671": {"f_id": "104009164", "md5": "120ddd118103ea2ea927aaa489310dc8"}, "103379672": {"f_id": "104009165", "md5": "220ddd118103ea2ea927aaa489310dc8"}}}} +``` + +`tests/zot/fetch/fixtures/libgen_json_file.json` (the `object=f` answer for the three f_ids above; `27858482` is deliberately absent): + +```json +{"27858481": {"md5": "a78add1bd62047ead5945134488f470f", "filesize": "401662", "extension": "pdf"}, "104009164": {"md5": "120ddd118103ea2ea927aaa489310dc8", "filesize": "9000000", "extension": "pdf"}, "104009165": {"md5": "220ddd118103ea2ea927aaa489310dc8", "filesize": "4000000", "extension": "epub"}} +``` + +`tests/zot/fetch/fixtures/unpaywall_doi.json` (the real API shape; this DOI is closed-access live, so the fixture is a synthetic OA record with the same keys): + +```json +{"doi": "10.1056/nejmoa1000678", "is_oa": true, "best_oa_location": {"url_for_pdf": "https://example.org/best.pdf", "url": "https://example.org/best"}, "oa_locations": [{"url_for_pdf": "https://example.org/best.pdf", "url": "https://example.org/best"}, {"url_for_pdf": null, "url": "https://example.org/landing"}]} +``` + +`tests/zot/fetch/fixtures/libgen_index.html` (minimal results table; the live search page did not return rows through the proxy on 2026-09-22 — the parser only needs the `edition.php?id=` anchors): + +```html + + + +
HCPCS Level II Professional 20232023
HCPCS Level II Professional 20192019
+``` + +- [ ] **Step 2: Write the failing tests** + +`tests/zot/fetch/test_resolve.py`: + +```python +from __future__ import annotations + +import json +from pathlib import Path + +import httpx +import pytest + +from zot.fetch.resolve import ( + BROWSER_UA, + LIBGEN, + Candidate, + Pacer, + candidates, + libgen_candidates, + libgen_editions, + libgen_files, + title_candidates, + title_matches, +) +from zot.fetch.select import ItemRef + +FX = Path(__file__).parent / "fixtures" + + +def _fx(name: str) -> str: + return (FX / name).read_text() + + +def _pacer() -> Pacer: + return Pacer(3.0, clock=lambda: 0.0, sleep=lambda s: None) + + +def _client(handler) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(handler), headers={"User-Agent": BROWSER_UA}) + + +def _libgen_handler(seen: list[str]): + def handler(req: httpx.Request) -> httpx.Response: + seen.append(str(req.url)) + assert req.headers["user-agent"] == BROWSER_UA + q = req.url.params + if req.url.path == "/json.php" and q.get("object") == "e": + if q.get("doi") == "10.1056/NEJMoa1000678": + return httpx.Response(200, text=_fx("libgen_json_doi.json")) + if q.get("isbn") == "9781640162280" or q.get("ids") == "146962484": + return httpx.Response(200, text=_fx("libgen_json_isbn.json")) + return httpx.Response(200, text="[]") + if req.url.path == "/json.php" and q.get("object") == "f": + want = set(q.get("ids", "").split(",")) + data = {k: v for k, v in json.loads(_fx("libgen_json_file.json")).items() if k in want} + return httpx.Response(200, json=data) + if req.url.path == "/index.php": + return httpx.Response(200, text=_fx("libgen_index.html")) + return httpx.Response(404) + + return handler + + +def test_pacer_sleeps_only_the_remaining_interval(): + now = [100.0] + slept: list[float] = [] + p = Pacer(3.0, clock=lambda: now[0], sleep=slept.append) + p.wait() # first call never sleeps + now[0] += 1.0 + p.wait() + assert slept == [2.0] + now[0] += 10.0 + p.wait() + assert slept == [2.0] + + +def test_libgen_editions_doi_and_isbn_and_ids(): + seen: list[str] = [] + c = _client(_libgen_handler(seen)) + e = libgen_editions(c, doi="10.1056/NEJMoa1000678", pacer=_pacer()) + assert list(e) == ["27703795"] + assert list(libgen_editions(c, isbn="9781640162280", pacer=_pacer())) == ["146962484"] + assert list(libgen_editions(c, ids=("146962484",), pacer=_pacer())) == ["146962484"] + assert seen[0].startswith(f"{LIBGEN}/json.php?object=e&addkeys=%2A&doi=") + + +@pytest.mark.parametrize("status, body", [(500, "x"), (200, "nginx"), (200, "[]"), (200, "null")]) +def test_libgen_editions_bad_answers_are_empty(status, body): + c = _client(lambda req: httpx.Response(status, text=body)) + assert libgen_editions(c, doi="10.1/x", pacer=_pacer()) == {} + + +def test_libgen_editions_transport_error_is_empty(): + def boom(req): + raise httpx.ConnectError("down") + + assert libgen_editions(_client(boom), doi="10.1/x", pacer=_pacer()) == {} + + +def test_libgen_files_batches_ids_in_one_call(): + seen: list[str] = [] + c = _client(_libgen_handler(seen)) + f = libgen_files(c, ["27858481", "104009164"], pacer=_pacer()) + assert set(f) == {"27858481", "104009164"} + assert len(seen) == 1 and "ids=27858481%2C104009164" in seen[0] + assert libgen_files(c, [], pacer=_pacer()) == {} + + +def test_libgen_candidates_paper_pdf_only_with_size(): + c = _client(_libgen_handler([])) + got = libgen_candidates(c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()) + assert got == [ + Candidate("libgen", "a78add1bd62047ead5945134488f470f", None, "pdf", 401662, + "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer"), + Candidate("libgen", "b78add1bd62047ead5945134488f470f", None, "pdf", None, + "Early Palliative Care for Patients with Metastatic Non–Small-Cell Lung Cancer"), + ] + + +def test_libgen_candidates_book_epub_before_pdf(): + c = _client(_libgen_handler([])) + got = libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer()) + assert [(g.ext, g.md5[:3]) for g in got] == [("epub", "220"), ("pdf", "120")] + + +def test_libgen_file_lookup_missing_defaults_paper_to_pdf_drops_book(): + def handler(req: httpx.Request) -> httpx.Response: + q = req.url.params + if q.get("object") == "f": + return httpx.Response(503) + if q.get("doi"): + return httpx.Response(200, text=_fx("libgen_json_doi.json")) + return httpx.Response(200, text=_fx("libgen_json_isbn.json")) + + c = _client(handler) + paper = libgen_candidates(c, kind="paper", doi="10.1056/NEJMoa1000678", pacer=_pacer()) + assert [p.ext for p in paper] == ["pdf", "pdf"] and paper[0].size_bytes is None + assert libgen_candidates(c, kind="book", isbn="9781640162280", pacer=_pacer()) == [] + + +def test_libgen_candidates_skips_files_without_md5(): + body = json.dumps({"1": {"title": "T", "files": {"9": {"f_id": "9"}}, "year": "2000"}}) + c = _client(lambda req: httpx.Response(200, text=body)) + assert libgen_candidates(c, kind="paper", doi="10.1/x", pacer=_pacer()) == [] + + +def test_title_matches_exact_then_containment_with_year(): + entries = [ + {"id": "1", "title": "HCPCS Level II Professional 2023", "year": "2023"}, + {"id": "2", "title": "HCPCS Level II Professional 2019", "year": "2019"}, + {"id": "3", "title": "CPT 2023 Professional", "year": "2023"}, + ] + assert [e["id"] for e in title_matches("HCPCS Level II Professional 2023", "", entries)] == ["1"] + assert [e["id"] for e in title_matches("HCPCS Level II Professional", "2019", entries)] == ["2"] + assert [e["id"] for e in title_matches("HCPCS Level II Professional", "", entries)] == ["1", "2"] + assert title_matches("Netter's Atlas", "", entries) == [] + + +def test_title_candidates_parses_index_then_fetches_editions(): + seen: list[str] = [] + c = _client(_libgen_handler(seen)) + got = title_candidates(c, title="HCPCS Level II Professional", year="2023", pacer=_pacer()) + assert [g.ext for g in got] == ["epub", "pdf"] + assert any("/index.php?" in s and "req=HCPCS+Level+II+Professional" in s for s in seen) + assert any("object=e" in s and "ids=146962484" in s for s in seen) + + +def test_title_candidates_no_rows_or_error_is_empty(): + assert title_candidates(_client(lambda r: httpx.Response(200, text="")), + title="X", year="", pacer=_pacer()) == [] + assert title_candidates(_client(lambda r: httpx.Response(500)), + title="X", year="", pacer=_pacer()) == [] + assert title_candidates(_client(lambda r: httpx.Response(200)), title="", year="", pacer=_pacer()) == [] + + +def _direct(oa: bool): + def handler(req: httpx.Request) -> httpx.Response: + assert req.url.host == "api.unpaywall.org" + assert req.url.params["email"] == "dev@fhirworx.io" + if not oa: + return httpx.Response(404) + return httpx.Response(200, text=_fx("unpaywall_doi.json")) + + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def _ref(**kw) -> ItemRef: + base = dict(item_id=1, key="ABCD2345", item_type="journalArticle", title="T", doi="", isbn="", year="") + base.update(kw) + return ItemRef(**base) + + +def test_candidates_doi_unpaywall_first_then_libgen(): + got = candidates( + _ref(doi="10.1056/NEJMoa1000678"), direct=_direct(True), proxied=_client(_libgen_handler([])), + email="dev@fhirworx.io", up_pacer=_pacer(), lg_pacer=_pacer(), + ) + assert [g.source for g in got] == ["unpaywall", "unpaywall", "unpaywall", "libgen", "libgen"] + assert got[0] == Candidate("unpaywall", None, "https://example.org/best.pdf", "pdf", None, "T") + assert got[3].md5 == "a78add1bd62047ead5945134488f470f" + + +def test_candidates_isbn_book_and_title_fallback_and_nothing(): + proxied = _client(_libgen_handler([])) + kw = dict(direct=_direct(False), proxied=proxied, email="e", up_pacer=_pacer(), lg_pacer=_pacer()) + book = candidates(_ref(item_type="book", isbn="9781640162280"), **kw) + assert [b.source for b in book] == ["libgen", "libgen"] + by_title = candidates(_ref(item_type="book", title="HCPCS Level II Professional", year="2023"), **kw) + assert [b.ext for b in by_title] == ["epub", "pdf"] + assert candidates(_ref(title="paper without doi"), **kw) == [] + assert candidates(_ref(item_type="book"), **kw) == [] +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_resolve.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.resolve'` + +- [ ] **Step 4: Implement resolve** + +`src/zot/fetch/resolve.py`: + +```python +"""Identifiers → download candidates. + +DOI: Unpaywall (direct, legal) first, then the libgen JSON index (through +the VPN proxy). ISBN: libgen JSON. A book with no ISBN: libgen title +search → edition ids → JSON. libgen serves a bare nginx page to +non-browser user agents, so the proxied client must send ``BROWSER_UA``. +""" + +from __future__ import annotations + +import re +import time +from collections.abc import Callable, Iterable +from dataclasses import dataclass + +import httpx + +from prisma.fetch import unpaywall_urls +from zot.fetch.select import ItemRef + +LIBGEN = "https://libgen.li" +BROWSER_UA = ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/128.0 Safari/537.36" +) +_EDITION_RE = re.compile(r'edition\.php\?id=(\d+)"[^>]*>([^<]*)<') +_YEAR_RE = re.compile(r"\b(19|20)\d\d\b") +_EXT_ORDER = {"book": ("epub", "pdf"), "paper": ("pdf",)} + + +@dataclass(frozen=True) +class Candidate: + source: str # unpaywall | libgen + md5: str | None + url: str | None + ext: str + size_bytes: int | None + title: str + + +class Pacer: + """Keep calls at least *min_interval* seconds apart.""" + + def __init__( + self, + min_interval: float, + *, + clock: Callable[[], float] = time.monotonic, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + self.min_interval = min_interval + self._clock = clock + self._sleep = sleep + self._last: float | None = None + + def wait(self) -> None: + now = self._clock() + if self._last is not None: + remaining = self.min_interval - (now - self._last) + if remaining > 0: + self._sleep(remaining) + self._last = self._clock() + + +def _get_json(client: httpx.Client, url: str, params: dict[str, str], pacer: Pacer) -> dict: + pacer.wait() + try: + r = client.get(url, params=params, timeout=30) + if r.status_code != 200: + return {} + data = r.json() + except (httpx.HTTPError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def libgen_editions( + client: httpx.Client, + *, + doi: str = "", + isbn: str = "", + ids: Iterable[str] = (), + pacer: Pacer, +) -> dict: + params = {"object": "e", "addkeys": "*"} + if doi: + params["doi"] = doi + elif isbn: + params["isbn"] = isbn + else: + params["ids"] = ",".join(ids) + return _get_json(client, f"{LIBGEN}/json.php", params, pacer) + + +def libgen_files(client: httpx.Client, f_ids: list[str], *, pacer: Pacer) -> dict: + if not f_ids: + return {} + params = {"object": "f", "addkeys": "*", "ids": ",".join(f_ids)} + return _get_json(client, f"{LIBGEN}/json.php", params, pacer) + + +def _candidates_from_editions( + client: httpx.Client, editions: dict, *, kind: str, pacer: Pacer +) -> list[Candidate]: + files: list[tuple[str, str, str]] = [] # (f_id, md5, title) + for ed in editions.values(): + title = (ed.get("title") or "") if isinstance(ed, dict) else "" + raw = ed.get("files") if isinstance(ed, dict) else None + entries = raw.values() if isinstance(raw, dict) else (raw or []) + for f in entries: + md5 = (f.get("md5") or "").lower() + if md5: + files.append((str(f.get("f_id") or ""), md5, title)) + info = libgen_files(client, [fid for fid, _, _ in files if fid], pacer=pacer) + out: list[Candidate] = [] + for fid, md5, title in files: + meta = info.get(fid) or {} + ext = (meta.get("extension") or "").lower() + if not ext and kind == "paper": + ext = "pdf" + if ext not in _EXT_ORDER[kind]: + continue + size = meta.get("filesize") + out.append( + Candidate("libgen", md5, None, ext, int(size) if size else None, title) + ) + order = _EXT_ORDER[kind] + out.sort(key=lambda c: order.index(c.ext)) + return out + + +def libgen_candidates( + client: httpx.Client, *, kind: str, doi: str = "", isbn: str = "", pacer: Pacer +) -> list[Candidate]: + editions = libgen_editions(client, doi=doi, isbn=isbn, pacer=pacer) + if not editions: + return [] + return _candidates_from_editions(client, editions, kind=kind, pacer=pacer) + + +def _norm(s: str) -> str: + return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip() + + +def _year(s: str) -> str: + m = _YEAR_RE.search(s or "") + return m.group(0) if m else "" + + +def title_matches(target: str, year: str, entries: list[dict]) -> list[dict]: + """Exact normalised title, else containment with agreeing years + (the AMA script's folder-matching rule).""" + t = _norm(target) + exact = [e for e in entries if _norm(e.get("title", "")) == t] + if exact: + return exact + tight = t.replace(" ", "") + out: list[dict] = [] + for e in entries: + ey = _year(e.get("title", "")) or str(e.get("year") or "") + if year and ey and year != ey: + continue + cand = _norm(e.get("title", "")).replace(" ", "") + if cand and (cand in tight or tight in cand): + out.append(e) + return out + + +def title_candidates( + client: httpx.Client, *, title: str, year: str, pacer: Pacer +) -> list[Candidate]: + if not title: + return [] + pacer.wait() + try: + r = client.get( + f"{LIBGEN}/index.php", + params={"req": title, "columns[]": "t", "objects[]": "e", "topics[]": "l", "res": "25"}, + timeout=30, + ) + except httpx.HTTPError: + return [] + if r.status_code != 200: + return [] + entries = [{"id": i, "title": t.strip()} for i, t in _EDITION_RE.findall(r.text)] + hits = title_matches(title, year, entries) + if not hits: + return [] + editions = libgen_editions(client, ids=[h["id"] for h in hits], pacer=pacer) + return _candidates_from_editions(client, editions, kind="book", pacer=pacer) + + +def candidates( + ref: ItemRef, + *, + direct: httpx.Client, + proxied: httpx.Client, + email: str, + up_pacer: Pacer, + lg_pacer: Pacer, +) -> list[Candidate]: + """Ordered candidates for one item: Unpaywall URLs, then libgen md5s.""" + out: list[Candidate] = [] + if ref.doi: + up_pacer.wait() + for url in unpaywall_urls(direct, ref.doi, email): + out.append(Candidate("unpaywall", None, url, "pdf", None, ref.title)) + out.extend(libgen_candidates(proxied, kind=ref.kind, doi=ref.doi, pacer=lg_pacer)) + elif ref.isbn: + out.extend(libgen_candidates(proxied, kind=ref.kind, isbn=ref.isbn, pacer=lg_pacer)) + elif ref.kind == "book" and ref.title: + out.extend(title_candidates(proxied, title=ref.title, year=ref.year, pacer=lg_pacer)) + return out +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch/test_resolve.py -q` +Expected: all passed. If `test_candidates_doi_unpaywall_first_then_libgen` reports a different Unpaywall count, check `prisma.fetch.unpaywall_urls`: it returns `url_for_pdf` values first across all locations then `url` values, de-duplicated — the fixture yields `best.pdf`, `best`, `landing` (three). + +- [ ] **Step 6: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/resolve.py tests/zot/fetch/test_resolve.py tests/zot/fetch/fixtures/ +git commit -m "feat(zot): fetch.resolve — Unpaywall + libgen JSON (DOI/ISBN/title) → ordered candidates (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 5: Deliver + +**Files:** +- Create: `src/zot/fetch/deliver.py` +- Create: `tests/zot/fetch/fixtures/libgen_ads.html`, `tests/zot/fetch/fixtures/annas_fast_download.json` +- Test: `tests/zot/fetch/test_deliver.py` + +**Interfaces:** +- Consumes: `Candidate`, `Pacer`, `LIBGEN` (Task 4). +- Produces: `ANNAS = "https://annas-archive.gl"`, `MIN_BYTES = 10_240`, `AnnasAuthError(RuntimeError)`, `Budget(annas_key="", reserve=5, annas_left=None, annas_disabled=False, quota_hit=False)` with `.can_use_annas() -> bool`, `download(client, url, dest, *, ext, headers=None) -> Path | None`, `fetch_libgen(client, cand, dest, *, pacer, sleep) -> Path | None`, `fetch_annas(client, cand, dest, *, budget) -> Path | None`, `fetch(cand, dest, *, direct, proxied, budget, pacer, sleep=time.sleep) -> str | None` (returns the source that delivered: `unpaywall` / `libgen` / `annas`). + +- [ ] **Step 1: Write the fixtures** + +`tests/zot/fetch/fixtures/libgen_ads.html` (the real page is 21 kB; only the anchor matters): + +```html +
GET
+``` + +`tests/zot/fetch/fixtures/annas_fast_download.json` (shape from the API's self-description; `account_fast_download_info` field names as documented on the member page): + +```json +{"download_url": "https://fast.example.org/dl/a78add1bd62047ead5945134488f470f.pdf", "account_fast_download_info": {"downloads_left": 42, "downloads_per_day": 50, "recently_downloaded_md5s": []}} +``` + +- [ ] **Step 2: Write the failing tests** + +`tests/zot/fetch/test_deliver.py`: + +```python +from __future__ import annotations + +from pathlib import Path + +import httpx +import pytest + +from zot.fetch.deliver import ( + ANNAS, + MIN_BYTES, + AnnasAuthError, + Budget, + download, + fetch, + fetch_annas, + fetch_libgen, +) +from zot.fetch.resolve import LIBGEN, Candidate, Pacer + +FX = Path(__file__).parent / "fixtures" +PDF = b"%PDF-1.4\n" + b"0" * MIN_BYTES +EPUB = b"PK\x03\x04" + b"0" * MIN_BYTES +MD5 = "a78add1bd62047ead5945134488f470f" +ADS = (FX / "libgen_ads.html").read_text() +ANNAS_OK = (FX / "annas_fast_download.json").read_text() + + +def _pacer() -> Pacer: + return Pacer(3.0, clock=lambda: 0.0, sleep=lambda s: None) + + +def _client(handler) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def _lg(md5=MD5) -> Candidate: + return Candidate("libgen", md5, None, "pdf", None, "T") + + +def _up() -> Candidate: + return Candidate("unpaywall", None, "https://example.org/x.pdf", "pdf", None, "T") + + +# ── download ────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "status, ctype, body, ext, ok", + [ + (200, "application/pdf", PDF, "pdf", True), + (200, "application/octet-stream", PDF, "pdf", True), + (200, "application/epub+zip", EPUB, "epub", True), + (200, "text/html; charset=utf-8", PDF, "pdf", False), + (200, "application/pdf", b"%PDF tiny", "pdf", False), + (200, "application/pdf", b"" + b"0" * MIN_BYTES, "pdf", False), + (503, "application/pdf", PDF, "pdf", False), + ], +) +def test_download_validates_status_type_size_and_magic(tmp_path, status, ctype, body, ext, ok): + c = _client(lambda r: httpx.Response(status, content=body, headers={"content-type": ctype})) + dest = tmp_path / f"f.{ext}" + got = download(c, "https://x/y", dest, ext=ext) + assert (got == dest) is ok + assert dest.exists() is ok + + +def test_download_transport_error_is_none(tmp_path): + def boom(r): + raise httpx.ReadTimeout("t") + + assert download(_client(boom), "https://x", tmp_path / "f.pdf", ext="pdf") is None + + +def test_download_passes_headers(tmp_path): + seen = {} + + def h(r): + seen.update(r.headers) + return httpx.Response(200, content=PDF, headers={"content-type": "application/pdf"}) + + download(_client(h), "https://x", tmp_path / "f.pdf", ext="pdf", headers={"Referer": "https://r/"}) + assert seen["referer"] == "https://r/" + + +# ── libgen ──────────────────────────────────────────────────────── + + +def _libgen_handler(ads_pages: list[httpx.Response], get_status=200): + calls: list[str] = [] + + def h(r: httpx.Request) -> httpx.Response: + calls.append(r.url.path) + if r.url.path == "/ads.php": + assert r.url.params["md5"] == MD5 + assert r.headers["referer"] == f"{LIBGEN}/" + return ads_pages.pop(0) + if r.url.path == "/get.php": + assert r.url.params["key"] == "AJPW7IO3P1LGRGWA" + return httpx.Response(get_status, content=PDF, headers={"content-type": "application/pdf"}) + raise AssertionError(r.url) + + return h, calls + + +def test_fetch_libgen_ads_then_get(tmp_path): + h, calls = _libgen_handler([httpx.Response(200, text=ADS)]) + got = fetch_libgen(_client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=lambda s: None) + assert got == tmp_path / "f.pdf" and calls == ["/ads.php", "/get.php"] + + +def test_libgen_ads_error_div_retries_once_then_gives_up(tmp_path): + err = httpx.Response(200, text='
Could not connect to the database 3306
') + h, calls = _libgen_handler([err, err]) + slept: list[float] = [] + got = fetch_libgen(_client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=slept.append) + assert got is None and calls == ["/ads.php", "/ads.php"] and slept == [3.0] + + +def test_libgen_ads_error_then_success_on_retry(tmp_path): + err = httpx.Response(500) + h, calls = _libgen_handler([err, httpx.Response(200, text=ADS)]) + got = fetch_libgen(_client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=lambda s: None) + assert got is not None and calls == ["/ads.php", "/ads.php", "/get.php"] + + +def test_libgen_get_503_is_none(tmp_path): + h, _ = _libgen_handler([httpx.Response(200, text=ADS)], get_status=503) + assert fetch_libgen(_client(h), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=lambda s: None) is None + + +def test_libgen_transport_error_is_none(tmp_path): + def boom(r): + raise httpx.ConnectError("x") + + assert fetch_libgen(_client(boom), _lg(), tmp_path / "f.pdf", pacer=_pacer(), sleep=lambda s: None) is None + + +# ── Anna's ──────────────────────────────────────────────────────── + + +def _annas_handler(api: httpx.Response, file_status=200): + def h(r: httpx.Request) -> httpx.Response: + if r.url.host == "annas-archive.gl": + assert r.url.path == "/dyn/api/fast_download.json" + assert r.url.params["md5"] == MD5 and r.url.params["key"] == "SECRET" + return api + return httpx.Response(file_status, content=PDF, headers={"content-type": "application/pdf"}) + + return h + + +def test_fetch_annas_downloads_and_updates_budget(tmp_path): + b = Budget(annas_key="SECRET") + got = fetch_annas(_client(_annas_handler(httpx.Response(200, text=ANNAS_OK))), _lg(), tmp_path / "f.pdf", budget=b) + assert got == tmp_path / "f.pdf" and b.annas_left == 41 + + +@pytest.mark.parametrize("status", [401, 403]) +def test_fetch_annas_auth_errors_stop_the_run(tmp_path, status): + b = Budget(annas_key="SECRET") + with pytest.raises(AnnasAuthError): + fetch_annas(_client(_annas_handler(httpx.Response(status, json={"error": "Not a member"}))), _lg(), tmp_path / "f.pdf", budget=b) + + +def test_fetch_annas_other_error_disables_annas(tmp_path): + b = Budget(annas_key="SECRET") + assert fetch_annas(_client(_annas_handler(httpx.Response(500))), _lg(), tmp_path / "f.pdf", budget=b) is None + assert b.annas_disabled is True and b.can_use_annas() is False + + +def test_fetch_annas_null_download_url_is_none(tmp_path): + b = Budget(annas_key="SECRET") + api = httpx.Response(200, json={"download_url": None, "error": "Invalid md5"}) + assert fetch_annas(_client(_annas_handler(api)), _lg(), tmp_path / "f.pdf", budget=b) is None + + +def test_fetch_annas_bad_json_or_transport_is_none(tmp_path): + b = Budget(annas_key="SECRET") + assert fetch_annas(_client(_annas_handler(httpx.Response(200, text="nope"))), _lg(), tmp_path / "f.pdf", budget=b) is None + + def boom(r): + raise httpx.ConnectError("x") + + assert fetch_annas(_client(boom), _lg(), tmp_path / "f.pdf", budget=b) is None + + +def test_fetch_annas_file_failure_keeps_budget_decrement(tmp_path): + b = Budget(annas_key="SECRET") + got = fetch_annas(_client(_annas_handler(httpx.Response(200, text=ANNAS_OK), file_status=503)), _lg(), tmp_path / "f.pdf", budget=b) + assert got is None and b.annas_left == 41 # the API call itself consumed one + + +def test_budget_reserve_blocks_annas(): + assert Budget(annas_key="").can_use_annas() is False + assert Budget(annas_key="k").can_use_annas() is True # unknown budget → allowed + assert Budget(annas_key="k", annas_left=6, reserve=5).can_use_annas() is True + assert Budget(annas_key="k", annas_left=5, reserve=5).can_use_annas() is False + + +# ── fetch (the cascade) ─────────────────────────────────────────── + + +def test_fetch_unpaywall_uses_direct_client(tmp_path): + direct = _client(lambda r: httpx.Response(200, content=PDF, headers={"content-type": "application/pdf"})) + proxied = _client(lambda r: (_ for _ in ()).throw(AssertionError("proxied used"))) + src = fetch(_up(), tmp_path / "f.pdf", direct=direct, proxied=proxied, budget=Budget(), pacer=_pacer()) + assert src == "unpaywall" + + +def test_fetch_libgen_then_annas_on_failure(tmp_path): + def h(r: httpx.Request) -> httpx.Response: + if r.url.path == "/ads.php": + return httpx.Response(500) + if r.url.host == "annas-archive.gl": + return httpx.Response(200, text=ANNAS_OK) + return httpx.Response(200, content=PDF, headers={"content-type": "application/pdf"}) + + b = Budget(annas_key="SECRET") + src = fetch(_lg(), tmp_path / "f.pdf", direct=_client(h), proxied=_client(h), budget=b, pacer=_pacer(), sleep=lambda s: None) + assert src == "annas" and b.annas_left == 41 + + +def test_fetch_marks_quota_when_annas_blocked_by_reserve(tmp_path): + proxied = _client(lambda r: httpx.Response(500)) + b = Budget(annas_key="SECRET", annas_left=2, reserve=5) + src = fetch(_lg(), tmp_path / "f.pdf", direct=proxied, proxied=proxied, budget=b, pacer=_pacer(), sleep=lambda s: None) + assert src is None and b.quota_hit is True + + +def test_fetch_all_fail_without_annas_is_not_quota(tmp_path): + proxied = _client(lambda r: httpx.Response(500)) + b = Budget() + src = fetch(_lg(), tmp_path / "f.pdf", direct=proxied, proxied=proxied, budget=b, pacer=_pacer(), sleep=lambda s: None) + assert src is None and b.quota_hit is False +``` + +- [ ] **Step 3: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_deliver.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.deliver'` + +- [ ] **Step 4: Implement deliver** + +`src/zot/fetch/deliver.py`: + +```python +"""Candidate → file on disk. + +libgen: ``ads.php?md5=`` (browser UA + Referer; a 200 with an error div +and no ``get.php`` link is a throttle → one retry after 3 s) then the +keyed ``get.php``. Anna's: the member fast-download JSON API by md5, +guarded by a daily reserve. Unpaywall URLs download directly. +""" + +from __future__ import annotations + +import re +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +import httpx + +from zot.fetch.resolve import LIBGEN, Candidate, Pacer + +ANNAS = "https://annas-archive.gl" +MIN_BYTES = 10_240 +MAGIC = {"pdf": b"%PDF", "epub": b"PK"} +_GET_LINK = re.compile(r'get\.php\?md5=([0-9a-f]{32})&(?:amp;)?key=([A-Za-z0-9]+)') +_ADS_RETRY_SECONDS = 3.0 + + +class AnnasAuthError(RuntimeError): + """The Anna's key is wrong or the account is not a member — stop the run.""" + + +@dataclass +class Budget: + annas_key: str = "" + reserve: int = 5 + annas_left: int | None = None + annas_disabled: bool = False + quota_hit: bool = False + + def can_use_annas(self) -> bool: + if not self.annas_key or self.annas_disabled: + return False + return self.annas_left is None or self.annas_left > self.reserve + + +def download( + client: httpx.Client, + url: str, + dest: Path, + *, + ext: str, + headers: dict[str, str] | None = None, +) -> Path | None: + """Stream *url* to *dest*; None unless it is a real file of type *ext*.""" + try: + with client.stream("GET", url, headers=headers, timeout=120, follow_redirects=True) as r: + if r.status_code != 200: + return None + if "text/html" in r.headers.get("content-type", "").lower(): + return None + dest.parent.mkdir(parents=True, exist_ok=True) + with dest.open("wb") as fh: + for chunk in r.iter_bytes(1 << 16): + fh.write(chunk) + except (httpx.HTTPError, OSError): + dest.unlink(missing_ok=True) + return None + try: + size = dest.stat().st_size + with dest.open("rb") as fh: + head = fh.read(4) + except OSError: + return None + if size < MIN_BYTES or not head.startswith(MAGIC[ext]): + dest.unlink(missing_ok=True) + return None + return dest + + +def _ads_key(client: httpx.Client, md5: str, pacer: Pacer) -> str | None: + pacer.wait() + try: + r = client.get( + f"{LIBGEN}/ads.php", params={"md5": md5}, headers={"Referer": f"{LIBGEN}/"}, timeout=30 + ) + except httpx.HTTPError: + return None + if r.status_code != 200: + return None + m = _GET_LINK.search(r.text) + return m.group(2) if m and m.group(1) == md5 else None + + +def fetch_libgen( + client: httpx.Client, + cand: Candidate, + dest: Path, + *, + pacer: Pacer, + sleep: Callable[[float], None], +) -> Path | None: + md5 = cand.md5 or "" + key = _ads_key(client, md5, pacer) + if key is None: + sleep(_ADS_RETRY_SECONDS) + key = _ads_key(client, md5, pacer) + if key is None: + return None + return download( + client, + f"{LIBGEN}/get.php?md5={md5}&key={key}", + dest, + ext=cand.ext, + headers={"Referer": f"{LIBGEN}/ads.php?md5={md5}"}, + ) + + +def fetch_annas( + client: httpx.Client, cand: Candidate, dest: Path, *, budget: Budget +) -> Path | None: + try: + r = client.get( + f"{ANNAS}/dyn/api/fast_download.json", + params={"md5": cand.md5 or "", "key": budget.annas_key}, + timeout=30, + ) + except httpx.HTTPError: + return None + if r.status_code in (401, 403): + raise AnnasAuthError(f"Anna's Archive refused the key ({r.status_code}): {r.text[:200]}") + if r.status_code not in (200, 204): + budget.annas_disabled = True + return None + try: + data = r.json() + except ValueError: + return None + info = data.get("account_fast_download_info") or {} + if isinstance(info.get("downloads_left"), int): + budget.annas_left = info["downloads_left"] + url = data.get("download_url") + if not url: + return None + return download(client, url, dest, ext=cand.ext) + + +def fetch( + cand: Candidate, + dest: Path, + *, + direct: httpx.Client, + proxied: httpx.Client, + budget: Budget, + pacer: Pacer, + sleep: Callable[[float], None] = time.sleep, +) -> str | None: + """Deliver *cand* to *dest*; returns the source that worked, else None. + + Sets ``budget.quota_hit`` when Anna's would have been tried but the + reserve (or an earlier API failure) blocked it. + """ + if cand.source == "unpaywall": + return "unpaywall" if download(direct, cand.url or "", dest, ext=cand.ext) else None + if fetch_libgen(proxied, cand, dest, pacer=pacer, sleep=sleep): + return "libgen" + if budget.can_use_annas(): + return "annas" if fetch_annas(proxied, cand, dest, budget=budget) else None + if budget.annas_key: + budget.quota_hit = True + return None +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch/test_deliver.py -q` +Expected: all passed + +- [ ] **Step 6: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/deliver.py tests/zot/fetch/test_deliver.py tests/zot/fetch/fixtures/libgen_ads.html tests/zot/fetch/fixtures/annas_fast_download.json +git commit -m "feat(zot): fetch.deliver — libgen ads→get with throttle retry, Anna's fast download under a daily reserve (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 6: Run loop + +**Files:** +- Create: `src/zot/fetch/run.py` +- Modify: `src/zot/fetch/__init__.py` +- Test: `tests/zot/fetch/test_run.py` + +**Interfaces:** +- Consumes: everything above — `select_items`, `ItemRef`, `candidates`, `Pacer`, `BROWSER_UA`, `fetch`, `Budget`, `AnnasAuthError`, `attach_file`, `sudo_copier`, `Ledger`. +- Produces: `Clients(direct: httpx.Client, proxied: httpx.Client)`, `make_clients(proxy: str, *, email: str) -> Clients`, `safe_name(title: str, fallback: str) -> str`, `run(db, refs, *, ledger, budget, clients, email, storage_dir, scratch_dir, limit=0, dry_run=False, retry=False, copier=sudo_copier, sleep=time.sleep, out=print) -> dict[str, int]` with stats keys `seen, skipped, attempted, attached, no_candidate, all_failed, quota`. + +- [ ] **Step 1: Write the failing tests** + +`tests/zot/fetch/test_run.py`: + +```python +from __future__ import annotations + +from pathlib import Path + +import httpx +import pytest + +from tests.zot.fetch.conftest import mk_item +from zot.db import Db +from zot.fetch.attach import has_full_text +from zot.fetch.deliver import MIN_BYTES, AnnasAuthError, Budget +from zot.fetch.ledger import Ledger +from zot.fetch.resolve import BROWSER_UA +from zot.fetch.run import Clients, make_clients, run, safe_name +from zot.fetch.select import select_items + +FX = Path(__file__).parent / "fixtures" +PDF = b"%PDF-1.4\n" + b"0" * MIN_BYTES +ADS = (FX / "libgen_ads.html").read_text() + + +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 _clients(direct_handler, proxied_handler) -> Clients: + return Clients( + direct=httpx.Client(transport=httpx.MockTransport(direct_handler)), + proxied=httpx.Client(transport=httpx.MockTransport(proxied_handler), headers={"User-Agent": BROWSER_UA}), + ) + + +def _unpaywall_ok(r: httpx.Request) -> httpx.Response: + if r.url.host == "api.unpaywall.org": + return httpx.Response(200, text=(FX / "unpaywall_doi.json").read_text()) + return httpx.Response(200, content=PDF, headers={"content-type": "application/pdf"}) + + +def _unpaywall_miss(r: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + +def _libgen_ok(r: httpx.Request) -> httpx.Response: + q = r.url.params + if r.url.path == "/json.php" and q.get("object") == "e": + return httpx.Response(200, text=(FX / "libgen_json_doi.json").read_text()) + if r.url.path == "/json.php": + return httpx.Response(200, text=(FX / "libgen_json_file.json").read_text()) + if r.url.path == "/ads.php": + return httpx.Response(200, text=ADS) + if r.url.path == "/get.php": + return httpx.Response(200, content=PDF, headers={"content-type": "application/pdf"}) + return httpx.Response(404) + + +def _libgen_none(r: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="[]") + + +def _libgen_dead(r: httpx.Request) -> httpx.Response: + if r.url.path == "/json.php" and r.url.params.get("object") == "e": + return httpx.Response(200, text=(FX / "libgen_json_doi.json").read_text()) + if r.url.path == "/json.php": + return httpx.Response(200, text=(FX / "libgen_json_file.json").read_text()) + if r.url.host == "annas-archive.gl": + return httpx.Response(200, json={"download_url": None, "error": "nope"}) + return httpx.Response(500) + + +def _setup(zdb: Db, tmp_path: Path, **kw): + return dict( + ledger=Ledger(tmp_path / "ledger.jsonl"), + budget=kw.pop("budget", Budget()), + email="dev@fhirworx.io", + storage_dir=tmp_path / "storage", + scratch_dir=tmp_path / "scratch", + copier=_py_copier, + sleep=lambda s: None, + out=kw.pop("out", lambda *a: None), + **kw, + ) + + +def test_safe_name(): + assert safe_name("Early Palliative Care: NSCLC / 2010", "K") == "Early Palliative Care NSCLC 2010" + assert safe_name("", "ABCD2345") == "ABCD2345" + assert len(safe_name("x" * 200, "K")) == 80 + + +def test_make_clients_sets_proxy_and_agents(): + c = make_clients("http://127.0.0.1:19035", email="dev@fhirworx.io") + assert c.proxied.headers["user-agent"] == BROWSER_UA + assert "dev@fhirworx.io" in c.direct.headers["user-agent"] + assert c.proxied._transport is not c.direct._transport # proxied one carries the proxy + + +def test_run_attaches_from_unpaywall_and_ledgers(zdb: Db, tmp_path: Path): + item = mk_item(zdb, title="Early Palliative Care", doi="10.1056/NEJMoa1000678") + refs = select_items(zdb, all_=True) + lines: list[str] = [] + kw = _setup(zdb, tmp_path, out=lambda *a: lines.append(" ".join(map(str, a)))) + stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), **kw) + assert stats == {"seen": 1, "skipped": 0, "attempted": 1, "attached": 1, "no_candidate": 0, "all_failed": 0, "quota": 0} + assert has_full_text(zdb, item) + assert "source:unpaywall" in zdb.get_tags(item) + att = zdb.con.execute("SELECT path FROM itemAttachments WHERE parentItemID = ?", (item,)).fetchone()[0] + assert att == "storage:Early Palliative Care.pdf" + assert kw["ledger"].summary() == {"attached": 1} + assert any("attached" in ln for ln in lines) + + +def test_run_falls_through_to_libgen_with_md5_provenance(zdb: Db, tmp_path: Path): + item = mk_item(zdb, title="T", doi="10.1056/NEJMoa1000678") + kw = _setup(zdb, tmp_path) + stats = run(zdb, select_items(zdb, all_=True), clients=_clients(_unpaywall_miss, _libgen_ok), **kw) + assert stats["attached"] == 1 + assert "source:libgen" in zdb.get_tags(item) + assert zdb.get_field(item, "extra") == "md5: a78add1bd62047ead5945134488f470f" + + +def test_run_no_candidate_and_all_failed_outcomes(zdb: Db, tmp_path: Path): + mk_item(zdb, title="no ids") + mk_item(zdb, title="dead", doi="10.1056/NEJMoa1000678") + kw = _setup(zdb, tmp_path) + stats = run(zdb, select_items(zdb, all_=True), clients=_clients(_unpaywall_miss, _libgen_dead), **kw) + assert stats["no_candidate"] == 1 and stats["all_failed"] == 1 and stats["attached"] == 0 + assert kw["ledger"].summary() == {"no_candidate": 1, "all_failed": 1} + + +def test_run_ledgers_quota_when_only_annas_was_left(zdb: Db, tmp_path: Path): + mk_item(zdb, title="dead", doi="10.1056/NEJMoa1000678") + kw = _setup(zdb, tmp_path, budget=Budget(annas_key="k", annas_left=1, reserve=5)) + stats = run(zdb, select_items(zdb, all_=True), clients=_clients(_unpaywall_miss, _libgen_dead), **kw) + assert stats["quota"] == 1 and stats["all_failed"] == 0 + assert kw["ledger"].summary() == {"quota": 1} + + +def test_run_skips_ledgered_items_and_honours_limit(zdb: Db, tmp_path: Path): + a = mk_item(zdb, title="A", doi="10.1056/NEJMoa1000678") + mk_item(zdb, title="B", doi="10.1056/NEJMoa1000678") + mk_item(zdb, title="C", doi="10.1056/NEJMoa1000678") + refs = select_items(zdb, all_=True) + kw = _setup(zdb, tmp_path) + kw["ledger"].record(refs[0].key, outcome="no_candidate") + stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), limit=1, **kw) + assert stats["skipped"] == 1 and stats["attempted"] == 1 and stats["attached"] == 1 + assert not has_full_text(zdb, a) + + +def test_run_retry_flag_reprocesses_ledgered_items(zdb: Db, tmp_path: Path): + mk_item(zdb, title="A", doi="10.1056/NEJMoa1000678") + refs = select_items(zdb, all_=True) + kw = _setup(zdb, tmp_path) + kw["ledger"].record(refs[0].key, outcome="all_failed") + stats = run(zdb, refs, clients=_clients(_unpaywall_ok, _libgen_none), retry=True, **kw) + assert stats["skipped"] == 0 and stats["attached"] == 1 + + +def test_run_dry_run_prints_candidates_and_writes_nothing(zdb: Db, tmp_path: Path): + item = mk_item(zdb, title="T", doi="10.1056/NEJMoa1000678") + lines: list[str] = [] + kw = _setup(zdb, tmp_path, out=lambda *a: lines.append(" ".join(map(str, a)))) + stats = run(zdb, select_items(zdb, all_=True), clients=_clients(_unpaywall_ok, _libgen_ok), dry_run=True, **kw) + assert stats["attempted"] == 1 and stats["attached"] == 0 + assert not has_full_text(zdb, item) + assert not (tmp_path / "ledger.jsonl").exists() + assert any("unpaywall" in ln for ln in lines) and any("libgen" in ln for ln in lines) + + +def test_run_annas_auth_error_aborts(zdb: Db, tmp_path: Path): + mk_item(zdb, title="T", doi="10.1056/NEJMoa1000678") + + def proxied(r: httpx.Request) -> httpx.Response: + if r.url.host == "annas-archive.gl": + return httpx.Response(401) + return _libgen_dead(r) + + kw = _setup(zdb, tmp_path, budget=Budget(annas_key="bad")) + with pytest.raises(AnnasAuthError): + run(zdb, select_items(zdb, all_=True), clients=_clients(_unpaywall_miss, proxied), **kw) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/zot/fetch/test_run.py -q` +Expected: FAIL with `ModuleNotFoundError: No module named 'zot.fetch.run'` + +- [ ] **Step 3: Implement run** + +`src/zot/fetch/run.py`: + +```python +"""The per-item loop: ledger gate → candidates → deliver → attach → ledger.""" + +from __future__ import annotations + +import hashlib +import re +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +import httpx + +from zot.db import Db +from zot.fetch.attach import Copier, attach_file, sudo_copier +from zot.fetch.deliver import Budget, fetch +from zot.fetch.ledger import Ledger +from zot.fetch.resolve import BROWSER_UA, Pacer, candidates +from zot.fetch.select import ItemRef + +STATS = ("seen", "skipped", "attempted", "attached", "no_candidate", "all_failed", "quota") +_UNSAFE = re.compile(r"[^A-Za-z0-9 ._-]+") + + +@dataclass +class Clients: + direct: httpx.Client + proxied: httpx.Client + + +def make_clients(proxy: str, *, email: str) -> Clients: + direct = httpx.Client( + headers={"User-Agent": f"stack-zot-fetch/1 (mailto:{email})"}, follow_redirects=True + ) + proxied = httpx.Client( + proxy=proxy, headers={"User-Agent": BROWSER_UA}, follow_redirects=True + ) + return Clients(direct=direct, proxied=proxied) + + +def safe_name(title: str, fallback: str) -> str: + cleaned = " ".join(_UNSAFE.sub(" ", title or "").split())[:80].strip() + return cleaned or fallback + + +def _scratch_path(scratch_dir: Path, ref: ItemRef, cand_key: str, ext: str) -> Path: + tag = hashlib.sha1(cand_key.encode()).hexdigest()[:10] # noqa: S324 — not security + return scratch_dir / f"{ref.key}-{tag}.{ext}" + + +def run( + db: Db, + refs: list[ItemRef], + *, + ledger: Ledger, + budget: Budget, + clients: Clients, + email: str, + storage_dir: Path, + scratch_dir: Path, + limit: int = 0, + dry_run: bool = False, + retry: bool = False, + copier: Copier = sudo_copier, + sleep: Callable[[float], None] = time.sleep, + out: Callable[..., None] = print, +) -> dict[str, int]: + stats = dict.fromkeys(STATS, 0) + up_pacer = Pacer(1.0, sleep=sleep) + lg_pacer = Pacer(3.0, sleep=sleep) + scratch_dir.mkdir(parents=True, exist_ok=True) + + for ref in refs: + stats["seen"] += 1 + if not ledger.should_try(ref.key, retry=retry): + stats["skipped"] += 1 + continue + if limit and stats["attempted"] >= limit: + break + stats["attempted"] += 1 + cands = candidates( + ref, direct=clients.direct, proxied=clients.proxied, email=email, + up_pacer=up_pacer, lg_pacer=lg_pacer, + ) + if dry_run: + out(f"{ref.key} {ref.title[:60]} ids={ref.ids}") + for c in cands: + out(f" {c.source:<9} {c.ext:<4} {c.md5 or c.url}") + if not cands: + out(" (no candidates)") + continue + if not cands: + ledger.record(ref.key, outcome="no_candidate", ids=ref.ids) + stats["no_candidate"] += 1 + out(f"{ref.key} no_candidate {ref.title[:60]}") + continue + + budget.quota_hit = False + delivered: tuple[str, Path, str] | None = None + for c in cands: + dest = _scratch_path(scratch_dir, ref, c.md5 or c.url or "", c.ext) + source = fetch( + c, dest, direct=clients.direct, proxied=clients.proxied, + budget=budget, pacer=lg_pacer, sleep=sleep, + ) + if source: + delivered = (source, dest, c.md5 or "") + break + if delivered is None: + outcome = "quota" if budget.quota_hit else "all_failed" + ledger.record(ref.key, outcome=outcome, ids=ref.ids) + stats[outcome] += 1 + out(f"{ref.key} {outcome} {ref.title[:60]}") + continue + + source, path, md5 = delivered + final = path.with_name(f"{safe_name(ref.title, ref.key)}.{path.suffix.lstrip('.')}") + path.replace(final) + attach_file( + db, ref.item_id, final, source=source, md5=md5, + storage_dir=storage_dir, copier=copier, + ) + db.commit() + final.unlink(missing_ok=True) + ledger.record(ref.key, outcome="attached", source=source, md5=md5, ids=ref.ids) + stats["attached"] += 1 + out(f"{ref.key} attached {source} {final.name}") + return stats +``` + +Update `src/zot/fetch/__init__.py`: + +```python +"""Full-text connector for Zotero items (spec 2026-09-09, slice Z1). + +select → resolve → deliver → attach, with a JSONL ledger of attempts. +""" + +from zot.fetch.deliver import AnnasAuthError, Budget +from zot.fetch.ledger import Ledger +from zot.fetch.resolve import Candidate +from zot.fetch.run import Clients, make_clients, run +from zot.fetch.select import ItemRef, select_items + +__all__ = [ + "AnnasAuthError", "Budget", "Candidate", "Clients", "ItemRef", "Ledger", + "make_clients", "run", "select_items", +] +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/zot/fetch -q` +Expected: all passed + +- [ ] **Step 5: Coverage check on the package** + +Run: `uv run pytest tests/zot/fetch --cov=src/zot/fetch --cov-report=term-missing -q` +Expected: every module at 100%. Any uncovered line gets a test in the owning test file before you continue (the CI gate is 99% across all of `src`, and this package must not spend that margin). + +- [ ] **Step 6: Lint and commit** + +```bash +uv run ruff check src/zot/fetch tests/zot/fetch && uv run ruff format src/zot/fetch tests/zot/fetch +git status --short +git add src/zot/fetch/run.py src/zot/fetch/__init__.py tests/zot/fetch/test_run.py +git commit -m "feat(zot): fetch.run — the per-item loop with dry-run, limit, retry and quota outcomes (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 7: CLI verb, env, docs + +**Files:** +- Modify: `src/cli/zot.py` (append the `fetch` command after `verify-parity`) +- Test: `tests/cli/test_zot_fetch.py` +- Generate: `docs/docs/cli/zot-fetch.md`, `docs/docs/cli/zot.md` +- Modify: `docs/superpowers/specs/2026-09-09-zotero-annas-connector-design.md` (status line) +- Local only, never committed: `.env` gains `VPN_HTTP_PROXY` and `ANNAS_ARCHIVE_KEY` + +**Interfaces:** +- Consumes: `select_items`, `make_clients`, `run`, `Ledger`, `Budget`, `AnnasAuthError`, `conf.path("storage.zotero")`, `_hold_zotero`, `_resolve`. +- Produces: `stack zot fetch [--collection NAME] [--tag TAG] [--type TYPE] [--item KEY] [--all] [--limit N] [--dry-run] [--retry] [--db PATH] [--ledger PATH] [--hold/--no-hold]`. + +- [ ] **Step 1: Write the failing CLI tests** + +`tests/cli/test_zot_fetch.py`: + +```python +"""stack zot fetch — CLI wiring (no network, no docker).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from cli.zot import app +from tests.zot.fetch.conftest import mk_item +from zot.db import Db +from zot.fetch.deliver import AnnasAuthError + +runner = CliRunner() + + +@pytest.fixture() +def env(monkeypatch, tmp_path: Path): + monkeypatch.setenv("VPN_HTTP_PROXY", "http://127.0.0.1:19035") + monkeypatch.setenv("ANNAS_ARCHIVE_KEY", "k") + monkeypatch.setenv("ANNAS_DAILY_RESERVE", "7") + monkeypatch.setenv("UNPAYWALL_EMAIL", "dev@fhirworx.io") + monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn()) + return tmp_path + + +def _db_with_item(zdb: Db) -> str: + mk_item(zdb, title="T", doi="10.1/x") + zdb.commit() + return zdb.path + + +def test_fetch_requires_proxy(monkeypatch, zdb: Db): + monkeypatch.delenv("VPN_HTTP_PROXY", raising=False) + monkeypatch.setattr("cli.zot._hold_zotero", lambda fn, hold: fn()) + r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"]) + assert r.exit_code != 0 + assert "VPN_HTTP_PROXY" in r.output and "gluetun" in r.output + + +def test_fetch_dry_run_calls_run_with_selection(env, zdb: Db, monkeypatch): + captured: dict = {} + + def fake_run(db, refs, **kw): + captured["refs"] = refs + captured["kw"] = kw + return {"seen": 1, "skipped": 0, "attempted": 1, "attached": 0, "no_candidate": 0, "all_failed": 0, "quota": 0} + + monkeypatch.setattr("cli.zot.run", fake_run) + r = runner.invoke( + app, + ["fetch", "--all", "--dry-run", "--limit", "3", "--db", _db_with_item(zdb), + "--ledger", str(env / "ledger.jsonl"), "--no-hold"], + ) + assert r.exit_code == 0, r.output + assert [x.doi for x in captured["refs"]] == ["10.1/x"] + kw = captured["kw"] + assert kw["dry_run"] is True and kw["limit"] == 3 and kw["retry"] is False + assert kw["budget"].annas_key == "k" and kw["budget"].reserve == 7 + assert kw["email"] == "dev@fhirworx.io" + assert kw["ledger"].path == env / "ledger.jsonl" + assert "seen=1" in r.output and "attempted=1" in r.output + + +def test_fetch_selection_error_is_a_usage_error(env, zdb: Db): + r = runner.invoke(app, ["fetch", "--db", _db_with_item(zdb), "--no-hold"]) + assert r.exit_code != 0 and "exactly one" in r.output + + +def test_fetch_annas_auth_error_exits_nonzero_with_message(env, zdb: Db, monkeypatch): + def fake_run(db, refs, **kw): + raise AnnasAuthError("Anna's Archive refused the key (403)") + + monkeypatch.setattr("cli.zot.run", fake_run) + r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"]) + assert r.exit_code != 0 and "refused the key" in r.output + + +def test_fetch_without_annas_key_prints_note(env, zdb: Db, monkeypatch): + monkeypatch.delenv("ANNAS_ARCHIVE_KEY") + monkeypatch.setattr("cli.zot.run", lambda db, refs, **kw: dict.fromkeys( + ("seen", "skipped", "attempted", "attached", "no_candidate", "all_failed", "quota"), 0)) + r = runner.invoke(app, ["fetch", "--all", "--db", _db_with_item(zdb), "--no-hold"]) + assert r.exit_code == 0 and "ANNAS_ARCHIVE_KEY not set" in r.output +``` + +Add a `zdb` fixture import for the CLI tests by creating `tests/cli/conftest.py` if it does not exist, or appending to it: + +```python +from tests.zot.fetch.conftest import zdb # noqa: F401 — re-export the scratch Zotero DB +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `uv run pytest tests/cli/test_zot_fetch.py -q` +Expected: FAIL — `fetch` is not a command (`No such command 'fetch'`) and `cli.zot.run` does not exist. + +- [ ] **Step 3: Add the command** + +Append to `src/cli/zot.py` (after `verify_parity`). Add `import os` and the fetch imports at the top with the other imports: + +```python +import os +``` + +and, below `import typer`: + +```python +from zot.fetch.deliver import AnnasAuthError, Budget +from zot.fetch.ledger import Ledger +from zot.fetch.run import make_clients, run +from zot.fetch.select import select_items +``` + +The command: + +```python +_DEFAULT_LEDGER = Path("data/zotero/fetch_ledger.jsonl") +_PROXY_HELP = ( + "VPN_HTTP_PROXY is not set. libgen and Anna's must go through the gluetun " + "HTTP proxy published by corwins.media (docker compose up gluetun → " + "127.0.0.1:19035): add VPN_HTTP_PROXY=http://127.0.0.1:19035 to .env and " + "source it (set -a; . ./.env; set +a) before running." +) + + +@app.command(name="fetch") +def fetch( + collection: str = typer.Option("", "--collection", help="Collection name."), + tag: str = typer.Option("", "--tag", help="Tag name."), + type_: str = typer.Option("", "--type", help="Item type, e.g. journalArticle, book."), + item: str = typer.Option("", "--item", help="One item key."), + all_: bool = typer.Option(False, "--all", help="Every eligible item."), + limit: int = typer.Option(0, "--limit", help="Stop after N attempts (0 = all)."), + dry_run: bool = typer.Option(False, "--dry-run", help="List candidates, write nothing."), + retry: bool = typer.Option(False, "--retry", help="Retry items the ledger marked as misses."), + db_path: Path | None = typer.Option(None, "--db"), + ledger_path: Path = typer.Option(_DEFAULT_LEDGER, "--ledger"), + hold: bool = typer.Option(True, "--hold/--no-hold"), +) -> None: + """Attach full text (PDF/EPUB) to items that lack it. + + DOI → Unpaywall, then the libgen index; ISBN → libgen; delivery by + libgen or Anna's fast download (member key). Items with a PDF/EPUB, + deleted items and ``document`` items are always skipped. Every + attempt is a line in the ledger; misses wait 30 days unless --retry. + """ + from conf import path as conf_path + + proxy = os.environ.get("VPN_HTTP_PROXY", "") + if not proxy: + raise typer.BadParameter(_PROXY_HELP) + annas_key = os.environ.get("ANNAS_ARCHIVE_KEY", "") + if not annas_key: + typer.echo("ANNAS_ARCHIVE_KEY not set — Anna's fast download disabled for this run.") + email = os.environ.get("UNPAYWALL_EMAIL", "dev@fhirworx.io") + reserve = int(os.environ.get("ANNAS_DAILY_RESERVE", "5")) + path = _resolve(db_path) + + def _go() -> dict[str, int]: + with Db(str(path)) as db: + try: + refs = select_items( + db, collection=collection, tag=tag, type_=type_, item=item, all_=all_ + ) + except ValueError as exc: + raise typer.BadParameter(str(exc)) from exc + typer.echo(f"{len(refs)} eligible item(s)") + clients = make_clients(proxy, email=email) + try: + return run( + db, + refs, + ledger=Ledger(ledger_path), + budget=Budget(annas_key=annas_key, reserve=reserve), + clients=clients, + email=email, + storage_dir=conf_path("storage.zotero"), + scratch_dir=Path(".state/zot/fetch"), + limit=limit, + dry_run=dry_run, + retry=retry, + out=typer.echo, + ) + except AnnasAuthError as exc: + typer.echo(f"error: {exc}", err=True) + raise typer.Exit(2) from exc + finally: + clients.direct.close() + clients.proxied.close() + + stats = _hold_zotero(_go, hold) + typer.echo("fetch: " + " ".join(f"{k}={v}" for k, v in stats.items())) +``` + +`src/cli/zot.py` imports `zot.db` lazily inside each command today (verified 2026-09-22: no top-level `import os` or `from zot.db import Db`). Add both at module top with the imports above; the lazy imports elsewhere in the file keep working. + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `uv run pytest tests/cli/test_zot_fetch.py tests/cli/test_zot.py -q` +Expected: all passed. `CliRunner` mixes stderr into `output` by default, so the auth-error assertion reads it there. + +- [ ] **Step 5: Regenerate the CLI docs** + +```bash +uv run python docs/scripts/extract_cli.py +git status --short docs/docs/cli +``` + +Expected: `docs/docs/cli/zot-fetch.md` created, `docs/docs/cli/zot.md` updated with the new command row; any other changed doc files are regeneration noise from unrelated commands — leave them unstaged unless the diff is only whitespace from this run. + +- [ ] **Step 6: Mark the spec as in build** + +In `docs/superpowers/specs/2026-09-09-zotero-annas-connector-design.md`, replace the status line: + +``` +**Status:** draft for operator approval, 2026-09-09. +``` + +with + +``` +**Status:** Z1 approved and built 2026-09-22 (plan `docs/superpowers/plans/2026-09-22-zotero-fetch-z1.md`); Z2–Z4 pending. +``` + +- [ ] **Step 7: Full test run and lint** + +```bash +uv run ruff check src tests && uv run ruff format --check src tests +uv run pytest tests/zot tests/cli/test_zot.py tests/cli/test_zot_fetch.py -q +``` + +Expected: all passed, ruff clean. Then the whole suite with the CI gate, since the coverage threshold is global: + +```bash +uv run pytest tests/ -x --cov=src --cov-report=term-missing --cov-fail-under=99 -q -n auto +``` + +Expected: passes the 99% gate. + +- [ ] **Step 8: Commit** + +```bash +git status --short +git add src/cli/zot.py tests/cli/test_zot_fetch.py tests/cli/conftest.py docs/docs/cli/zot-fetch.md docs/docs/cli/zot.md docs/superpowers/specs/2026-09-09-zotero-annas-connector-design.md +git commit -m "feat(zot): stack zot fetch — full-text connector CLI under the --hold guard, with docs (refs Z1) + +Co-Authored-By: Claude Fable 5.1 " +``` + +--- + +### Task 8: Live acceptance (from `/home/kert/stack`, after merge to main) + +**Files:** none committed. Local `.env` edits and a ledger under `data/zotero/`. + +**Interfaces:** the two "Done when" criteria in the spec's Z1 slice. + +- [ ] **Step 1: Put the two env keys in the stack `.env`** (both files belong to the operator; neither is committed) + +```bash +cd /home/kert/stack +grep -q '^VPN_HTTP_PROXY=' .env || echo 'VPN_HTTP_PROXY=http://127.0.0.1:19035' >> .env +grep -q '^ANNAS_ARCHIVE_KEY=' .env || grep -h '^ANNAS_ARCHIVE_KEY=' /home/kert/corwins.media/.env >> .env +grep -cE '^(VPN_HTTP_PROXY|ANNAS_ARCHIVE_KEY)=' .env # expect 2 +``` + +- [ ] **Step 2: Confirm the proxy and libgen answer** + +```bash +curl -s -m 20 -x http://127.0.0.1:19035 -A "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0 Safari/537.36" \ + "https://libgen.li/json.php?object=e&addkeys=*&doi=10.1056/NEJMoa1000678" | head -c 200 +``` + +Expected: JSON starting `{"27703795":{"title":"Early Palliative Care …`. A `Welcome to nginx!` page means the user agent was dropped; a timeout means the corwins gluetun service is down (`docker compose -f /home/kert/corwins.media/compose.yml up -d gluetun`). + +- [ ] **Step 3: Dry run on the AMA collection** (criterion 1) + +```bash +set -a; . ./.env; set +a +uv run stack zot fetch --collection "AMA Coding Publications" --dry-run +``` + +Expected: a line per eligible book with `libgen epub|pdf ` candidate lines beneath; books that already carry an EPUB/PDF are absent (excluded by `has_full_text`). Zotero stops and restarts around the run (`--hold`). + +- [ ] **Step 4: Real run on twenty journal articles** (criterion 2) + +```bash +set -a; . ./.env; set +a +uv run stack zot fetch --type journalArticle --limit 20 +tail -n 20 data/zotero/fetch_ledger.jsonl +``` + +Expected: the summary line shows `attached ≥ 1` with at least one `source=unpaywall` and one `source=libgen` or `source=annas` line in the ledger, and every other attempted item has a `no_candidate` / `all_failed` / `quota` line. Open Zotero and confirm the attachment renders for one attached item and that its tags gained `source:` and nothing else changed. + +If the harness's auto-mode classifier blocks the Anna's leg of the run, hand the exact command to the operator to run with `!` rather than retrying (memory: `palliative_rfi_project`). + +- [ ] **Step 5: Record the outcome** + +Comment on the Gitea tracker (a new issue for Z1 under a "P51: Zotero full-text connector" milestone, or the corwins.media #707 thread if the operator prefers) with the summary line, the ledger counts by outcome and source, and the Anna's `downloads_left` after the run. Update the memory file `zotero_fetch_connector.md` with the live findings (the browser-UA requirement, the DB-3306 throttle page, the proxy's exit host). + +--- + +## Self-review + +**Spec coverage.** Decisions 1–9 map to: resolver/attach unification (Tasks 2, 4), package + CLI verb under `--hold` (Tasks 1–7), explicit selection with the standing exclusions (Task 3), quota reserve + libgen/Unpaywall pacing (Tasks 4–5), ledger + 30-day rule (Task 1), provenance tags/extra (Task 2), proxy required with the compose step named (Task 7), secrets in `.env` (Task 8), no timer (nothing scheduled). Error handling bullets: per-item isolation (Task 6), Unpaywall 404 not an error (`unpaywall_urls` returns `[]`, Task 4), ads 500 → one retry (Task 5), get 503 → next (Task 5), Anna's 401/403 stop and other non-200 skip (Task 5), DB-locked with `--no-hold` — `Db.__init__` raises `sqlite3.OperationalError` before any download because `select_items` runs first inside `_go` (Task 7). Testing section: MockTransport everywhere, fixtures pinned (Tasks 4–5), truth table for `has_full_text` (Task 2), 30-day rule (Task 1), CLI dry-run on a scratch DB (Task 7). Z2 (folding the AMA script) is deliberately not here — its `_attach` keeps working until Z2 replaces it. + +**Deviations from the spec, stated:** (a) the budget is read from Anna's on the first *successful* download call rather than "at run start", because the API only returns `account_fast_download_info` alongside a `download_url`; until then Anna's is allowed. (b) `attach.py` uses `linkMode 0` (imported file), matching the AMA script rather than `prisma.fetch.attach_pdf`'s `1`. (c) `prisma.fetch.unpaywall_urls` is reused instead of re-implemented; it already returns every OA location, best first. + +**Type consistency.** `Candidate(source, md5, url, ext, size_bytes, title)` is constructed positionally in tests and by keyword nowhere else; `ItemRef` field order matches `item_ref`; `fetch()` returns `str | None` and `run` treats the string as the source; `Budget.quota_hit` is reset per item in `run` and set only in `fetch`. + +**Review Focus.** All five lines have their named tests in Tasks 3, 4, 5 and 6.