Skills › Scientific Agent Skills › paper-lookup › arxiv_atom.py
arxiv_atom.py
205 lines · skills/paper-lookup/scripts/arxiv_atom.py at 5f2bc03
1#!/usr/bin/env python3
2"""Parse arXiv Atom XML into JSON records, catching arXiv's HTTP-200 failures.
4arXiv has no JSON output, and its Atom feed has four traps that make hand-rolled
5parsing quietly wrong (all documented in references/arxiv.md):
7- The feed carries its own `<link>` before the first entry, so "the first link"
8 is the query URL, not a paper.
9- A malformed parameter returns HTTP 200, `totalResults` **1**, and a single
10 entry titled `Error` -- which reads as a successful one-hit search.
11- `<id>` is now `https://` and carries a version suffix (`1706.03762v7`).
12- `<title>` and `<summary>` arrive hard-wrapped mid-sentence.
14This script handles all four, exits **3** on the Error entry, and exits **5** when arXiv is
15throttling -- which it signals with the bare plain-text body `Rate exceeded.`, not a feed.
17 curl -s "https://export.arxiv.org/api/query?id_list=1706.03762" | python3 arxiv_atom.py -
18 python3 arxiv_atom.py feed.xml --ids-only
19"""
21from __future__ import annotations
23import argparse
24import sys
25import xml.etree.ElementTree as ET
26from pathlib import Path
27from typing import Any
29sys.path.insert(0, str(Path(__file__).resolve().parent))
31from _common import InputError, collapse_ws, emit, fail, read_input # noqa: E402
33NS = {
34 "atom": "http://www.w3.org/2005/Atom",
35 "arxiv": "http://arxiv.org/schemas/atom",
36 "opensearch": "http://a9.com/-/spec/opensearch/1.1/",
37}
40def split_version(arxiv_id: str) -> tuple[str, str | None]:
41 """`1706.03762v7` -> `("1706.03762", "7")`.
43 The bare ID is what Semantic Scholar's `ARXIV:` prefix, a DOI, and a
44 user-supplied ID all use, so comparing the versioned form against any of them
45 fails. Both are returned rather than choosing one.
46 """
47 base, separator, version = arxiv_id.rpartition("v")
48 if separator and version.isdigit() and base:
49 return base, version
50 return arxiv_id, None
53def id_from_url(url: str) -> str:
54 """The arXiv ID from an `<id>` URL, scheme-agnostically.
56 Preserve the entire path after `/abs/`: old identifiers such as
57 `hep-th/9901001` contain a slash.
58 """
59 from urllib.parse import urlsplit
61 path = urlsplit(url).path.rstrip("/")
62 return path.split("/abs/", 1)[1] if "/abs/" in path else path.rsplit("/", 1)[-1]
65def link_for(entry: ET.Element, *, rel: str, mime: str | None = None) -> str | None:
66 for link in entry.findall("atom:link", NS):
67 if link.get("rel") != rel:
68 continue
69 if mime and link.get("type") != mime:
70 continue
71 return link.get("href")
72 return None
75def parse_entry(entry: ET.Element) -> dict[str, Any]:
76 raw_id = collapse_ws(entry.findtext("atom:id", namespaces=NS))
77 versioned = id_from_url(raw_id) if raw_id else ""
78 arxiv_id, version = split_version(versioned)
80 categories = [
81 term
82 for term in (category.get("term") for category in entry.findall("atom:category", NS))
83 if term
84 ]
85 primary = entry.find("arxiv:primary_category", NS)
87 return {
88 "arxiv_id": arxiv_id,
89 "arxiv_id_versioned": versioned or None,
90 "version": version,
91 "title": collapse_ws(entry.findtext("atom:title", namespaces=NS)),
92 "abstract": collapse_ws(entry.findtext("atom:summary", namespaces=NS)),
93 "authors": [
94 collapse_ws(name)
95 for name in (
96 author.findtext("atom:name", namespaces=NS)
97 for author in entry.findall("atom:author", NS)
98 )
99 if collapse_ws(name)
100 ],
101 "published": collapse_ws(entry.findtext("atom:published", namespaces=NS)) or None,
102 "updated": collapse_ws(entry.findtext("atom:updated", namespaces=NS)) or None,
103 "primary_category": primary.get("term") if primary is not None else None,
104 "categories": categories,
105 "doi": collapse_ws(entry.findtext("arxiv:doi", namespaces=NS)) or None,
106 "journal_ref": collapse_ws(entry.findtext("arxiv:journal_ref", namespaces=NS)) or None,
107 "comment": collapse_ws(entry.findtext("arxiv:comment", namespaces=NS)) or None,
108 # Selected by rel/type, never by position: the feed's own <link> precedes
109 # the entries and would otherwise be picked up as a paper URL.
110 "abstract_url": link_for(entry, rel="alternate", mime="text/html"),
111 "pdf_url": link_for(entry, rel="related", mime="application/pdf"),
112 }
115def build_parser() -> argparse.ArgumentParser:
116 parser = argparse.ArgumentParser(
117 description=(
118 "Parse an arXiv Atom feed into JSON records. Exits 3 when the feed is an "
119 "arXiv error response, which arrives as HTTP 200 with totalResults 1 and a "
120 "single entry titled 'Error'."
121 ),
122 epilog='curl -s "https://export.arxiv.org/api/query?id_list=1706.03762" | %(prog)s -',
123 )
124 parser.add_argument("source", help="path to an arXiv Atom XML file, or - for stdin")
125 parser.add_argument("-o", "--output", help="write JSON here instead of stdout")
126 parser.add_argument(
127 "--ids-only",
128 action="store_true",
129 help="print one bare arXiv ID per line (version suffix stripped)",
130 )
131 return parser
134def main(argv: list[str] | None = None) -> int:
135 args = build_parser().parse_args(argv)
137 try:
138 xml_text = read_input(args.source)
139 except InputError as error:
140 fail(str(error))
142 # Throttling is not XML. arXiv answers a rate-limited caller with the bare
143 # plain-text body "Rate exceeded." -- 14 bytes, no feed, no status to key on --
144 # so an XML parse error here is usually a pacing problem, not a bad response.
145 if xml_text.strip().startswith("Rate exceeded"):
146 fail(
147 "arXiv is throttling: it returned the plain-text body 'Rate exceeded.' instead of a "
148 "feed. Its limit is one request per three seconds; wait and retry, and serialize "
149 "arXiv calls rather than running them alongside other work.",
150 code=5,
151 )
153 try:
154 feed = ET.fromstring(xml_text)
155 except ET.ParseError as error:
156 fail(
157 f"not parseable as XML: {error}. The first 100 bytes were: "
158 f"{xml_text[:100]!r} -- arXiv returns plain text rather than a feed for "
159 "throttling and some gateway errors."
160 )
162 if feed.tag != f"{{{NS['atom']}}}feed":
163 fail("not an arXiv Atom feed; check for an HTML/XML error response")
164 entries = feed.findall("atom:entry", NS)
166 # The error check must precede any other interpretation: the error feed is a
167 # structurally valid one-hit search result.
168 for entry in entries:
169 if collapse_ws(entry.findtext("atom:title", namespaces=NS)) == "Error":
170 reason = collapse_ws(entry.findtext("atom:summary", namespaces=NS))
171 fail(f"arXiv returned an error feed: {reason or 'no reason given'}", code=3)
173 total = collapse_ws(feed.findtext("opensearch:totalResults", namespaces=NS))
174 # The feed <title> echoes the query as arXiv actually ran it. An unrecognized
175 # field prefix is silently rewritten to `all:`, so this is the only way to see
176 # that the executed query is not the one that was sent.
177 echoed_query = collapse_ws(feed.findtext("atom:title", namespaces=NS))
179 records = [parse_entry(entry) for entry in entries]
181 if args.ids_only:
182 text = "".join(f"{record['arxiv_id']}\n" for record in records if record["arxiv_id"])
183 if args.output:
184 Path(args.output).write_text(text, encoding="utf-8")
185 else:
186 sys.stdout.write(text)
187 return 0
189 payload: dict[str, Any] = {
190 "total_results": int(total) if total.isdigit() else None,
191 "returned": len(records),
192 "query_as_executed": echoed_query or None,
193 "entries": records,
194 }
195 if not records:
196 payload["note"] = (
197 "zero entries with HTTP 200 is arXiv's genuine no-match response, including for "
198 "an unknown ID in id_list; report it as 'not found in arXiv', not as a failure"
199 )
200 emit(payload, args.output)
201 return 0
204if __name__ == "__main__":
205 raise SystemExit(main())