Skills › Scientific Agent Skills › paper-lookup › paginate.py
paginate.py
521 lines · skills/paper-lookup/scripts/paginate.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded, rate-limited, count-reconciling pagination for this skill's APIs.
4The five database adapters here paginate differently -- absolute record offsets
5or opaque cursors -- and each reports totals its
6own way. Re-deriving the walk per query is how records get silently dropped. The
7worst case is bioRxiv: `cursor` is an absolute offset, `/details/` returns 30 per
8page but `/pubs/` returns 100, and an out-of-step cursor returns **HTTP 200**, so
9stepping by 100 skips records 30-99 of every hundred and looks successful.
11Every walk here:
13- steps by the page size the response actually reported, never an assumed one
14- stops on this API's terminator (Europe PMC may omit the next cursor or echo it
15 on an empty page; bioRxiv also stops at its reported total)
16- reconciles retrieved against the expected total and **exits 4 on a shortfall**
17- refuses to exceed --max-records / --max-calls, and says so rather than
18 truncating quietly
20 python3 paginate.py --api biorxiv --query 2024-01-01/2024-01-03
21 python3 paginate.py --api europepmc --query 'SRC:"PPR" AND "organoid"' --max-records 200
22 python3 paginate.py --api openalex --query 'filter=publication_year:2024' --dry-run
24Needs network access. No credentials required for bioRxiv, medRxiv, Europe PMC,
25Crossref, or OpenAlex; OPENALEX_API_KEY raises the OpenAlex daily budget.
26"""
28from __future__ import annotations
30import argparse
31import json
32import os
33import sys
34import time
35import urllib.error
36import urllib.parse
37import urllib.request
38from dataclasses import dataclass
39from pathlib import Path
40from typing import Any, Callable
42sys.path.insert(0, str(Path(__file__).resolve().parent))
44from _common import Reconciliation, emit, fail, redact_url # noqa: E402
46USER_AGENT = "paper-lookup-skill/2.4 (+https://agentskills.io)"
47DEFAULT_MAX_RECORDS = 1000
48DEFAULT_MAX_CALLS = 50
49REQUEST_TIMEOUT = 60
52@dataclass
53class Page:
54 """One response, normalized."""
56 records: list[Any]
57 total: int | None = None
58 #: The next cursor/token/offset, or None when this API says it is done.
59 next_state: Any = None
60 #: Anything the caller must be told that is not a record.
61 notes: list[str] | None = None
64@dataclass
65class Api:
66 name: str
67 #: Seconds to wait between requests. Serialized: never parallelize one host.
68 delay: float
69 build_url: Callable[[str, Any, int], str]
70 parse: Callable[[Any, Any], Page]
71 initial_state: Any = 0
72 note: str = ""
75def fetch(url: str, *, headers: dict[str, str] | None = None) -> Any:
76 request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, **(headers or {})})
77 try:
78 with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
79 body = response.read().decode("utf-8", errors="replace")
80 except urllib.error.HTTPError as error:
81 # Servers may echo credentials in their error bodies. Do not print them.
82 raise RuntimeError(f"HTTP {error.code} from {redact_url(url)}") from error
83 except urllib.error.URLError as error:
84 raise RuntimeError(f"could not reach {redact_url(url)}") from error
85 try:
86 return json.loads(body)
87 except json.JSONDecodeError as error:
88 raise RuntimeError(f"response from {redact_url(url)} was not JSON") from error
91# --- bioRxiv / medRxiv ------------------------------------------------------
92#
93# `cursor` is an absolute record offset. The page size is 30 on /details/ and 100
94# on /pubs/, and an out-of-step cursor is accepted with HTTP 200 -- so the step
95# comes from the response's own `count`, never from a constant.
98def _rxiv_url(server: str) -> Callable[[str, Any, int], str]:
99 def build(query: str, state: Any, _limit: int) -> str:
100 endpoint = "pubs" if query.startswith("pubs:") else "details"
101 interval = query[5:] if query.startswith("pubs:") else query
102 if interval.startswith("10."):
103 return f"https://api.biorxiv.org/{endpoint}/{server}/{urllib.parse.quote(interval, safe='/')}/na/json"
104 if interval.isdigit():
105 return f"https://api.biorxiv.org/{endpoint}/{server}/{interval}"
106 suffix = "/json" if endpoint == "details" else ""
107 return f"https://api.biorxiv.org/{endpoint}/{server}/{interval}/{int(state)}{suffix}"
109 return build
112def _rxiv_parse(payload: Any, state: Any) -> Page:
113 if not isinstance(payload, dict):
114 raise RuntimeError(f"expected a JSON object, got {type(payload).__name__}")
116 if not isinstance(payload.get("collection"), list) or not isinstance(payload.get("messages"), list):
117 raise RuntimeError("bioRxiv/medRxiv response lacks messages/collection arrays")
118 messages = payload.get("messages") or [{}]
119 message = messages[0] if isinstance(messages[0], dict) else {}
120 status = message.get("status")
121 records = payload.get("collection") or []
122 notes: list[str] = []
124 if status and status != "ok":
125 if status != "no articles found":
126 raise RuntimeError("bioRxiv/medRxiv returned a non-success status")
127 # "no articles found" arrives with HTTP 200 and an empty collection, which
128 # is indistinguishable from a genuine no-match unless status is read.
129 notes.append(f"server status: {status!r} (HTTP 200 with an empty collection)")
130 return Page(records=[], total=None, next_state=None, notes=notes)
132 total = message.get("total")
133 total = int(total) if total is not None and str(total).isdigit() else None
135 new_papers = message.get("count_new_papers")
136 if new_papers is not None:
137 notes.append(
138 f"count_new_papers={new_papers} counts distinct first-posting preprints while "
139 f"total={total} counts every version record; deduplicate by DOI to compare against "
140 "count_new_papers"
141 )
143 reported = message.get("count")
144 step = int(reported) if isinstance(reported, int) and reported > 0 else len(records)
145 if not records:
146 return Page(records=[], total=total, next_state=None, notes=notes)
147 if total is None:
148 notes.append("endpoint reports no total or continuation; retained one response only")
149 return Page(records=records, total=None, next_state=None, notes=notes)
151 if step != len(records):
152 notes.append(f"response reported count={step} but returned {len(records)} records")
153 step = len(records)
155 next_state = int(state) + step
156 if total is not None and next_state >= total:
157 next_state = None
158 return Page(records=records, total=total, next_state=next_state, notes=notes)
161# --- Europe PMC ------------------------------------------------------------
162#
163# cursorMark. Exhaustion may omit the next cursor or return an empty result list
164# with the cursor echoed back. Accept both forms.
167def _europepmc_url(query: str, state: Any, limit: int) -> str:
168 params = {
169 "query": query,
170 "format": "json",
171 "pageSize": str(min(limit, 1000)),
172 "cursorMark": str(state),
173 "resultType": "lite",
174 }
175 return "https://www.ebi.ac.uk/europepmc/webservices/rest/search?" + urllib.parse.urlencode(params)
178def _europepmc_parse(payload: Any, state: Any) -> Page:
179 if not isinstance(payload, dict):
180 raise RuntimeError(f"expected a JSON object, got {type(payload).__name__}")
181 # Europe PMC reports errors with HTTP 200 and an errCode in the body.
182 if "errCode" in payload:
183 raise RuntimeError(
184 f"Europe PMC errCode {payload['errCode']}: {payload.get('errMsg', 'no message')}"
185 )
186 if not isinstance(payload.get("resultList"), dict) or not isinstance(payload.get("hitCount"), int):
187 raise RuntimeError("Europe PMC response lacks resultList/hitCount")
189 total = payload.get("hitCount")
190 records = (payload.get("resultList") or {}).get("result") or []
191 next_cursor = payload.get("nextCursorMark")
192 notes: list[str] = []
194 echoed = (payload.get("request") or {}).get("queryString")
195 if echoed:
196 notes.append(f"query as parsed by Europe PMC: {echoed!r}")
198 if not records or next_cursor in (None, state):
199 next_cursor = None
200 return Page(
201 records=records,
202 total=int(total) if isinstance(total, int) else None,
203 next_state=next_cursor,
204 notes=notes,
205 )
208# --- OpenAlex --------------------------------------------------------------
211def _openalex_url(query: str, state: Any, limit: int) -> str:
212 # `query` is a raw parameter string, e.g. `search=crispr` or
213 # `filter=publication_year:2024`, so both forms work without a second flag.
214 base = "https://api.openalex.org/works?"
215 params = dict(urllib.parse.parse_qsl(query, keep_blank_values=True))
216 params.update({"per_page": str(min(limit, 100)), "cursor": str(state)})
217 params.pop("per-page", None)
218 params.pop("page", None)
219 key = os.environ.get("OPENALEX_API_KEY")
220 if key:
221 params["api_key"] = key
222 return base + urllib.parse.urlencode(params)
225def _openalex_parse(payload: Any, _state: Any) -> Page:
226 if not isinstance(payload, dict):
227 raise RuntimeError(f"expected a JSON object, got {type(payload).__name__}")
228 if not isinstance(payload.get("meta"), dict) or not isinstance(payload.get("results"), list):
229 raise RuntimeError("OpenAlex response lacks meta/results; it may be an API error")
230 meta = payload.get("meta") or {}
231 records = payload.get("results") or []
232 notes = []
233 if meta.get("cost_usd") is not None:
234 notes.append(f"OpenAlex reported cost_usd={meta['cost_usd']} for this call")
235 next_cursor = meta.get("next_cursor")
236 if not records:
237 next_cursor = None
238 return Page(
239 records=records,
240 total=meta.get("count") if isinstance(meta.get("count"), int) else None,
241 next_state=next_cursor,
242 notes=notes,
243 )
246# --- Crossref --------------------------------------------------------------
249def _crossref_url(query: str, state: Any, limit: int) -> str:
250 params = dict(urllib.parse.parse_qsl(query, keep_blank_values=True))
251 params.update({"rows": str(min(limit, 1000)), "cursor": str(state)})
252 params.pop("offset", None)
253 if "sample" in params:
254 raise RuntimeError("Crossref sample cannot be combined with cursor pagination")
255 mail = os.environ.get("CROSSREF_MAILTO")
256 if mail:
257 params["mailto"] = mail
258 return "https://api.crossref.org/works?" + urllib.parse.urlencode(params)
261def _crossref_parse(payload: Any, _state: Any) -> Page:
262 if not isinstance(payload, dict):
263 raise RuntimeError(f"expected a JSON object, got {type(payload).__name__}")
264 message = payload.get("message") or {}
265 if not isinstance(message, dict) or not isinstance(message.get("items"), list):
266 raise RuntimeError("Crossref response lacks message.items; it may be an API error")
267 records = message.get("items") or []
268 next_cursor = message.get("next-cursor")
269 requested = message.get("items-per-page")
270 if not records or (isinstance(requested, int) and len(records) < requested):
271 next_cursor = None
272 total = message.get("total-results")
273 notes = ["Crossref cursors expire after 5 minutes; a long walk must keep moving"]
274 return Page(
275 records=records,
276 total=int(total) if isinstance(total, int) else None,
277 next_state=next_cursor,
278 notes=notes,
279 )
282APIS: dict[str, Api] = {
283 "biorxiv": Api(
284 name="biorxiv",
285 delay=1.0,
286 build_url=_rxiv_url("biorxiv"),
287 parse=_rxiv_parse,
288 initial_state=0,
289 note=(
290 "query is an interval (2024-01-01/2024-01-03), Nd, N, or a DOI. "
291 "Prefix with 'pubs:' to walk /pubs/ instead of /details/."
292 ),
293 ),
294 "medrxiv": Api(
295 name="medrxiv",
296 delay=1.0,
297 build_url=_rxiv_url("medrxiv"),
298 parse=_rxiv_parse,
299 initial_state=0,
300 note="same as biorxiv; always via api.biorxiv.org, never api.medrxiv.org",
301 ),
302 "europepmc": Api(
303 name="europepmc",
304 delay=0.5,
305 build_url=_europepmc_url,
306 parse=_europepmc_parse,
307 initial_state="*",
308 note="query is Europe PMC query syntax, e.g. 'SRC:\"PPR\" AND \"organoid\"'",
309 ),
310 "openalex": Api(
311 name="openalex",
312 delay=0.2,
313 build_url=_openalex_url,
314 parse=_openalex_parse,
315 initial_state="*",
316 note="query is a raw parameter string, e.g. 'search=crispr' or 'filter=publication_year:2024'",
317 ),
318 "crossref": Api(
319 name="crossref",
320 delay=1.0,
321 build_url=_crossref_url,
322 parse=_crossref_parse,
323 initial_state="*",
324 note="query is a raw parameter string, e.g. 'query.bibliographic=attention+is+all+you+need'",
325 ),
326}
329def walk(
330 api: Api,
331 query: str,
332 *,
333 page_size: int,
334 max_records: int,
335 max_calls: int,
336 verbose: bool,
337) -> tuple[list[Any], Reconciliation, list[str]]:
338 records: list[Any] = []
339 reconciliation = Reconciliation()
340 urls: list[str] = []
341 state = api.initial_state
342 seen_notes: set[str] = set()
344 while True:
345 if reconciliation.pages >= max_calls:
346 reconciliation.stopped_at_limit = True
347 reconciliation.note(
348 f"stopped at the --max-calls limit of {max_calls}; the walk is INCOMPLETE"
349 )
350 break
351 if len(records) >= max_records:
352 reconciliation.stopped_at_limit = True
353 reconciliation.note(
354 f"stopped at the --max-records limit of {max_records}; the walk is INCOMPLETE"
355 )
356 break
358 remaining = max_records - len(records)
359 try:
360 url = api.build_url(query, state, min(page_size, remaining))
361 except RuntimeError as error:
362 fail(str(error))
363 # Record and log the redacted form only. OpenAlex and Crossref authenticate
364 # by query string, so the fetched URL carries the credential and this
365 # provenance list is printed to the user.
366 safe_url = redact_url(url)
367 urls.append(safe_url)
368 if verbose:
369 sys.stderr.write(f" page {reconciliation.pages + 1}: {safe_url}\n")
371 try:
372 page = api.parse(fetch(url), state)
373 except RuntimeError as error:
374 fail(str(error))
376 reconciliation.pages += 1
377 if page.total is not None and reconciliation.expected is None:
378 reconciliation.expected = page.total
379 for note in page.notes or []:
380 if note not in seen_notes:
381 seen_notes.add(note)
382 reconciliation.note(note)
384 records.extend(page.records)
385 if page.next_state is None:
386 break
387 state = page.next_state
388 time.sleep(api.delay)
390 # Trim only after the walk, so the reported page count stays truthful.
391 if len(records) > max_records:
392 reconciliation.stopped_at_limit = True
393 reconciliation.note(
394 f"last page overshot --max-records; kept the first {max_records} of {len(records)}"
395 )
396 records = records[:max_records]
398 reconciliation.retrieved = len(records)
399 return records, reconciliation, urls
402def build_parser() -> argparse.ArgumentParser:
403 parser = argparse.ArgumentParser(
404 description=(
405 "Paginate one of this skill's APIs with the correct step, the correct stop "
406 "condition, and count reconciliation. Exits 4 on a reconciliation shortfall."
407 ),
408 epilog="python3 %(prog)s --api biorxiv --query 2024-01-01/2024-01-03",
409 )
410 # Not `required=True`: --list-apis is the flag you reach for when you do not yet
411 # know what to pass for either of these.
412 parser.add_argument("--api", choices=sorted(APIS), help="which API to walk")
413 parser.add_argument("--query", help="see --list-apis for the per-API format")
414 parser.add_argument("--page-size", type=int, default=100, help="requested page size (default 100)")
415 parser.add_argument(
416 "--max-records",
417 type=int,
418 default=DEFAULT_MAX_RECORDS,
419 help=f"stop after this many records (default {DEFAULT_MAX_RECORDS})",
420 )
421 parser.add_argument(
422 "--max-calls",
423 type=int,
424 default=DEFAULT_MAX_CALLS,
425 help=f"stop after this many requests (default {DEFAULT_MAX_CALLS})",
426 )
427 parser.add_argument("-o", "--output", help="write JSON here instead of stdout")
428 parser.add_argument(
429 "--dry-run",
430 action="store_true",
431 help="print the first URL that would be requested and exit without fetching",
432 )
433 parser.add_argument("--list-apis", action="store_true", help="describe each API's query format")
434 parser.add_argument("-v", "--verbose", action="store_true", help="log each URL to stderr")
435 return parser
438def main(argv: list[str] | None = None) -> int:
439 args = build_parser().parse_args(argv)
441 if args.list_apis:
442 emit(
443 {
444 name: {"delay_seconds": api.delay, "query_format": api.note}
445 for name, api in sorted(APIS.items())
446 },
447 args.output,
448 )
449 return 0
451 if not args.api or not args.query:
452 fail("--api and --query are both required (use --list-apis to see the query format)", code=2)
453 if args.page_size < 1:
454 fail("--page-size must be at least 1")
455 if args.max_records < 1:
456 fail("--max-records must be at least 1")
457 if args.max_calls < 1:
458 fail("--max-calls must be at least 1")
460 api = APIS[args.api]
462 if args.dry_run:
463 try:
464 first_url = api.build_url(args.query, api.initial_state, args.page_size)
465 except RuntimeError as error:
466 fail(str(error))
467 emit(
468 {
469 "api": api.name,
470 "first_url": redact_url(first_url),
471 "delay_seconds": api.delay,
472 "query_format": api.note,
473 },
474 args.output,
475 )
476 return 0
478 records, reconciliation, urls = walk(
479 api,
480 args.query,
481 page_size=args.page_size,
482 max_records=args.max_records,
483 max_calls=args.max_calls,
484 verbose=args.verbose,
485 )
487 emit(
488 {
489 "api": api.name,
490 "query": (
491 urllib.parse.urlsplit(redact_url("https://provenance.invalid/?" + args.query)).query
492 if args.api in {"openalex", "crossref"} else args.query
493 ),
494 "provenance": {"urls": urls, "delay_seconds": api.delay},
495 "reconciliation": reconciliation.as_dict(),
496 "records": records,
497 },
498 args.output,
499 )
501 if not reconciliation.ok:
502 # Exit 4 is reserved for the unexplained case: the walk terminated on its
503 # own and still came up short, which means records went missing. A bound
504 # the caller set is not a failure and exits 0 with the partiality recorded.
505 fail(
506 f"reconciliation failed: the walk ended on its own but retrieved "
507 f"{reconciliation.retrieved} of {reconciliation.expected}. Records are missing -- "
508 "say so before drawing any conclusion from this result.",
509 code=4,
510 )
511 if reconciliation.stopped_at_limit:
512 sys.stderr.write(
513 f"note: stopped at a caller-set bound with {reconciliation.retrieved}"
514 f"{f' of {reconciliation.expected}' if reconciliation.expected is not None else ''} "
515 "records. Report this result as partial.\n"
516 )
517 return 0
520if __name__ == "__main__":
521 raise SystemExit(main())