bee5f57893
Attaches an open-access PDF to Zotero items that lack a full text. Built on the selection, ledger and attach machinery from PR #826 (feat/zot-fetch-z1, left untouched on its branch); delivery is the stack's legal open-access cascade only — every Unpaywall location → Semantic Scholar openAccessPdf → PMC S3 → PMC OA service → Europe PMC (prisma.fetch.fetch_one with no proxied client, so its mirror tier never runs). No DOI? a Crossref title match first. Downloads must be real PDFs before they are attached; the run ledgers every attempt (attached / no_candidate / error; misses wait 30 days unless --retry). From the #828 review list: - Lazy selection: select_ids() is one cheap query (deleted and full-text items excluded in SQL for --all) and iter_refs() reads each item only when the run reaches it, so --limit bites before ~167k file-less items are materialised while the desktop is held down. - A locked database (--no-hold with the desktop writing) aborts the run with a usage error that names the Zotero desktop, instead of a raw sqlite3.OperationalError — or one error per item. - --dry-run reads only (checked against the running desktop), so it no longer stops/starts zotero; UNPAYWALL_EMAIL is documented in --help. Tests: 66 (scratch Zotero DB, no network), 100% of src/zot/fetch.
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
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="unpaywall",
|
|
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": "unpaywall",
|
|
"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)
|
|
early = T0 + RETRY_AFTER - timedelta(seconds=1)
|
|
assert led.should_try("K1", now=early) 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"
|
|
good = (
|
|
'{"item_key": "K1", "outcome": "attached", "ts": "2026-09-22T12:00:00+00:00"}'
|
|
)
|
|
p.write_text(good + "\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")
|
|
|
|
|
|
def test_error_outcome_retries_immediately(tmp_path: Path):
|
|
led = Ledger(tmp_path / "ledger.jsonl")
|
|
led.record(
|
|
"K1", outcome="error", note="ConnectError: unpaywall unreachable", now=T0
|
|
)
|
|
assert led.should_try("K1", now=T0) is True
|