Skills › Scientific Agent Skills › paper-lookup › jats_to_text.py
jats_to_text.py
324 lines · skills/paper-lookup/scripts/jats_to_text.py at 5f2bc03
1#!/usr/bin/env python3
2"""Turn PMC / Europe PMC JATS XML into sectioned text, refusing metadata-only XML.
4The failure this exists to stop: NCBI eFetch returns **HTTP 200** and a
5well-formed `<pmc-articleset>` for articles whose publisher forbids XML
6redistribution -- containing `<front>` metadata, no `<body>`, and the reason in
7an XML *comment* that every standard parser discards. An agent that fetches,
8parses, and reports "full text retrieved" has retrieved the title and author
9list. See the hazard section of references/pmc.md.
11So this script exits **2** when there is no `<body>`, and surfaces the discarded
12comment as the explanation. Metadata is still emitted, clearly labelled as
13metadata, so the caller can fall back to Europe PMC (a clean 404), Unpaywall, or
14the abstract without a second fetch.
16Handles both wrappers: eFetch's `<pmc-articleset><article>` and Europe PMC's bare
17`<article>`.
19 curl -s ".../efetch.fcgi?db=pmc&id=7029759&retmode=xml" | python3 jats_to_text.py -
20 python3 jats_to_text.py article.xml --sections METHODS,RESULTS --text-only
21"""
23from __future__ import annotations
25import argparse
26import re
27import sys
28import xml.etree.ElementTree as ET
29from pathlib import Path
30from typing import Any
32sys.path.insert(0, str(Path(__file__).resolve().parent))
34from _common import InputError, collapse_ws, emit, fail, read_input # noqa: E402
36#: Tags whose text is not part of the reading order. `xref` and `label` are
37#: excluded so citation markers and "Figure 1" labels do not land mid-sentence.
38SKIP_TAGS = frozenset({"xref", "label", "table-wrap", "graphic", "media", "inline-formula"})
40#: Block-level tags that must not run into their neighbours.
41#:
42#: NCBI serializes JATS with no whitespace between elements, so
43#: `<title>Introduction</title><p>A mysterious illness...` concatenates to
44#: "IntroductionA mysterious illness" unless a separator is inserted at these
45#: boundaries. Inline tags (`italic`, `sup`, `xref`) are deliberately absent --
46#: separating those would break words apart instead.
47BLOCK_TAGS = frozenset(
48 {
49 "abstract",
50 "body",
51 "caption",
52 "def-item",
53 "disp-quote",
54 "list-item",
55 "p",
56 "sec",
57 "statement",
58 "td",
59 "th",
60 "title",
61 "tr",
62 }
63)
65#: Publisher restriction notices arrive only as XML comments.
66RESTRICTION_HINT = re.compile(r"does not allow|not allow downloading|restricted", re.IGNORECASE)
69def parse(xml_text: str) -> ET.Element:
70 try:
71 return ET.fromstring(xml_text)
72 except ET.ParseError as error:
73 raise InputError(f"not parseable as XML: {error}") from error
76def find_article(root: ET.Element) -> ET.Element:
77 """The `<article>` element, whichever wrapper it arrived in."""
78 if root.tag == "article":
79 return root
80 article = root.find(".//article")
81 if article is None:
82 raise InputError(
83 f"no <article> element (root was <{root.tag}>); "
84 "this is not a JATS document -- check whether the response was an error page"
85 )
86 return article
89def comments_in(xml_text: str) -> list[str]:
90 """XML comments, which ElementTree drops.
92 Read from the raw text on purpose: the publisher-restriction notice that
93 explains a missing `<body>` exists *only* as a comment, so parsing it away is
94 what makes the failure silent.
95 """
96 return [collapse_ws(match) for match in re.findall(r"<!--(.*?)-->", xml_text, re.DOTALL)]
99def element_text(element: ET.Element) -> str:
100 """Flattened text of an element, skipping non-reading-order tags."""
101 parts: list[str] = []
103 def walk(node: ET.Element) -> None:
104 if node.tag in SKIP_TAGS:
105 # Keep the tail: text following an <xref> continues the sentence.
106 if node.tail:
107 parts.append(node.tail)
108 return
109 block = node.tag in BLOCK_TAGS
110 if block:
111 parts.append(" ")
112 if node.text:
113 parts.append(node.text)
114 for child in node:
115 walk(child)
116 if block:
117 parts.append(" ")
118 if node.tail:
119 parts.append(node.tail)
121 walk(element)
122 return collapse_ws("".join(parts))
125def section_title(section: ET.Element) -> str:
126 title = section.find("title")
127 return element_text(title) if title is not None else ""
130def collect_sections(body: ET.Element) -> list[dict[str, Any]]:
131 """Top-level `<sec>` blocks, each with its nested subsection text inlined."""
132 sections: list[dict[str, Any]] = []
134 top_level = body.findall("sec")
135 if not top_level:
136 # Some articles put paragraphs straight under <body> with no sections.
137 text = element_text(body)
138 return [{"title": "", "sec_type": None, "text": text}] if text else []
140 for section in top_level:
141 sections.append(
142 {
143 "title": section_title(section),
144 "sec_type": section.get("sec-type"),
145 "text": element_text(section),
146 }
147 )
148 return sections
151#: JATS `pub-id-type` values, mapped to the field names this script emits.
152#: PMC tags its own accession as `pmcid` (the bare `pmc` form appears in older
153#: documents), alongside `pmcid-ver`/`pmcaid`/`pmcaiid` variants that are not the
154#: canonical PMCID and must not be mistaken for it.
155ID_TYPES = {"pmid": "pmid", "pmcid": "pmcid", "pmc": "pmcid", "doi": "doi"}
158def extract_metadata(article: ET.Element) -> dict[str, Any]:
159 front = article.find("front")
160 metadata: dict[str, Any] = {
161 "title": None,
162 "journal": None,
163 "pmid": None,
164 "pmcid": None,
165 "doi": None,
166 "authors": [],
167 "abstract": None,
168 }
169 if front is None:
170 return metadata
172 title = front.find(".//title-group/article-title")
173 if title is not None:
174 metadata["title"] = element_text(title)
176 journal = front.find(".//journal-title")
177 if journal is not None:
178 metadata["journal"] = element_text(journal)
180 # First occurrence per type wins. F1000Research and similar journals nest peer
181 # review reports as sub-articles with their own DOIs; last-wins would report a
182 # review's DOI as the article's.
183 for article_id in front.findall(".//article-id"):
184 field_name = ID_TYPES.get(article_id.get("pub-id-type") or "")
185 if field_name and metadata[field_name] is None:
186 metadata[field_name] = collapse_ws(article_id.text)
188 for contributor in front.findall(".//contrib"):
189 surname = contributor.find(".//surname")
190 given = contributor.find(".//given-names")
191 name = " ".join(
192 part
193 for part in (
194 element_text(given) if given is not None else "",
195 element_text(surname) if surname is not None else "",
196 )
197 if part
198 )
199 if name:
200 metadata["authors"].append(name)
202 abstract = front.find(".//abstract")
203 if abstract is not None:
204 metadata["abstract"] = element_text(abstract)
206 return metadata
209def build_parser() -> argparse.ArgumentParser:
210 parser = argparse.ArgumentParser(
211 description=(
212 "Convert PMC/Europe PMC JATS XML to sectioned text. Exits 2 when the document "
213 "carries no <body>, which is how eFetch signals a non-open-access article "
214 "while still returning HTTP 200."
215 ),
216 epilog="python3 %(prog)s article.xml --sections METHODS,RESULTS",
217 )
218 parser.add_argument("source", help="path to a JATS XML file, or - for stdin")
219 parser.add_argument("-o", "--output", help="write here instead of stdout")
220 parser.add_argument(
221 "--sections",
222 help=(
223 "comma-separated section filter, matched case-insensitively against the "
224 "section title and sec-type (e.g. METHODS,RESULTS)"
225 ),
226 )
227 parser.add_argument(
228 "--text-only",
229 action="store_true",
230 help="emit plain text rather than JSON",
231 )
232 parser.add_argument(
233 "--allow-metadata-only",
234 action="store_true",
235 help=(
236 "exit 0 instead of 2 when there is no <body>. Only for callers that have "
237 "explicitly decided metadata is enough -- the default refusal is the point."
238 ),
239 )
240 return parser
243def matches(section: dict[str, Any], wanted: list[str]) -> bool:
244 haystack = f"{section['title']} {section['sec_type'] or ''}".lower()
245 return any(want in haystack for want in wanted)
248def main(argv: list[str] | None = None) -> int:
249 args = build_parser().parse_args(argv)
251 try:
252 xml_text = read_input(args.source)
253 root = parse(xml_text)
254 article = find_article(root)
255 except InputError as error:
256 fail(str(error))
258 metadata = extract_metadata(article)
259 body = article.find("body")
261 if body is None:
262 comments = comments_in(xml_text)
263 restriction = next((c for c in comments if RESTRICTION_HINT.search(c)), None)
264 payload = {
265 "full_text_available": False,
266 "reason": restriction
267 or "document has no <body> element and gave no stated reason",
268 "metadata": metadata,
269 "guidance": (
270 "This is metadata only, not full text. Do not present it as the article. "
271 "Try Europe PMC /{PMCID}/fullTextXML (returns 404 rather than a bodyless 200), "
272 "consult PMC Article Datasets for supported bulk access, or fall back to "
273 "Unpaywall/CORE for an open-access copy."
274 ),
275 }
276 if comments:
277 payload["xml_comments"] = comments
278 emit(payload, args.output)
279 if args.allow_metadata_only:
280 return 0
281 fail(
282 "no <body> element: this document is metadata only, not full text"
283 + (f" -- {restriction}" if restriction else ""),
284 code=2,
285 )
287 sections = collect_sections(body)
288 if args.sections:
289 wanted = [part.strip().lower() for part in args.sections.split(",") if part.strip()]
290 selected = [section for section in sections if matches(section, wanted)]
291 if not selected:
292 available = ", ".join(s["title"] or s["sec_type"] or "(untitled)" for s in sections)
293 fail(f"no section matched {args.sections!r}; available: {available}")
294 sections = selected
296 if args.text_only:
297 blocks = []
298 if metadata["title"]:
299 blocks.append(metadata["title"])
300 for section in sections:
301 heading = section["title"]
302 blocks.append(f"{heading}\n{section['text']}" if heading else section["text"])
303 text = "\n\n".join(blocks) + "\n"
304 if args.output:
305 Path(args.output).write_text(text, encoding="utf-8")
306 else:
307 sys.stdout.write(text)
308 return 0
310 emit(
311 {
312 "full_text_available": True,
313 "metadata": metadata,
314 "section_count": len(sections),
315 "word_count": sum(len(section["text"].split()) for section in sections),
316 "sections": sections,
317 },
318 args.output,
319 )
320 return 0
323if __name__ == "__main__":
324 raise SystemExit(main())