Spaces:
Sleeping
Sleeping
| """Open-access full text for a single paper, fetched and sectioned at read time. | |
| An abstract states what a paper claims; only the body states what it actually measured, | |
| in which system, at which dose, and what the numbers were. Novelty assessment reads the | |
| body, so this module resolves a work to Europe PMC and returns its sections. | |
| """ | |
| import asyncio | |
| import logging | |
| import re | |
| from dataclasses import dataclass, field | |
| from xml.etree import ElementTree | |
| import httpx | |
| logger = logging.getLogger(__name__) | |
| EUROPE_PMC = "https://www.ebi.ac.uk/europepmc/webservices/rest" | |
| RESULT_SECTIONS = ("results", "discussion", "conclusion") | |
| # Front and back matter carries no evidence and would crowd the read budget. | |
| SKIP_SECTIONS = ( | |
| "reference", "acknowledg", "competing", "author contribution", "funding", "footnote", | |
| "contributor", "data availability", "supplementary", "associated data", "ethics", | |
| "abbreviation", "conflict", | |
| ) | |
| # Headings vary between journals; match the informative ones by their stem. | |
| SECTION_KINDS = { | |
| "abstract": ("abstract", "summary"), | |
| "methods": ("method", "material", "experimental", "procedure"), | |
| "results": ("result", "finding"), | |
| "discussion": ("discussion", "conclusion", "interpretation"), | |
| } | |
| class FullText: | |
| """One paper's retrieved body, or the abstract when no open full text exists.""" | |
| paper_id: str | |
| availability: str # full_text | abstract_only | unavailable | |
| source: str = "" | |
| sections: dict[str, str] = field(default_factory=dict) | |
| # As the publisher record states them, which the index can contradict. | |
| title: str = "" | |
| year: int | None = None | |
| venue: str = "" | |
| def chars(self) -> int: | |
| return sum(len(text) for text in self.sections.values()) | |
| def evidence_text(self, limit: int = 24000) -> str: | |
| """The part a reader checks for what was found, results first, then the rest.""" | |
| order = [*RESULT_SECTIONS, "abstract", "methods"] | |
| ordered = sorted( | |
| self.sections.items(), | |
| key=lambda item: next( | |
| (i for i, name in enumerate(order) if name in item[0]), len(order) | |
| ), | |
| ) | |
| out: list[str] = [] | |
| remaining = limit | |
| for name, text in ordered: | |
| if remaining <= 0: | |
| break | |
| body = text[:remaining] | |
| remaining -= len(body) + len(name) | |
| out.append(f"## {name}\n{body}") | |
| return "\n\n".join(out) | |
| def _clean(text: str) -> str: | |
| return re.sub(r"\s+", " ", text).strip() | |
| def _kind(heading: str) -> str: | |
| lowered = heading.lower() | |
| if any(needle in lowered for needle in SKIP_SECTIONS): | |
| return "" | |
| for kind, needles in SECTION_KINDS.items(): | |
| if any(needle in lowered for needle in needles): | |
| return kind | |
| return heading.lower()[:60] or "body" | |
| def parse_jats(xml: str) -> dict[str, str]: | |
| """JATS body into `{section kind: text}`; unrecognized headings keep their own name.""" | |
| try: | |
| root = ElementTree.fromstring(xml) | |
| except ElementTree.ParseError as exc: | |
| logger.warning("Full text is not parseable XML: %s", exc) | |
| return {} | |
| sections: dict[str, str] = {} | |
| def add(name: str, text: str) -> None: | |
| text = _clean(text) | |
| if not name: | |
| return | |
| if len(text) < 40: | |
| return | |
| sections[name] = f"{sections[name]}\n{text}" if name in sections else text | |
| for abstract in root.iter("abstract"): | |
| add("abstract", " ".join(abstract.itertext())) | |
| body = root.find(".//body") | |
| if body is not None: | |
| for section in body.findall("sec"): | |
| title = section.find("title") | |
| heading = _clean("".join(title.itertext())) if title is not None else "body" | |
| add(_kind(heading), " ".join(section.itertext())) | |
| if not sections.keys() - {"abstract"}: | |
| add("body", " ".join(body.itertext())) | |
| return sections | |
| class FullTextClient: | |
| """Europe PMC lookup: DOI or PMID to an open-access body, cached per process.""" | |
| def __init__(self, client: httpx.AsyncClient | None = None, timeout: float = 30.0): | |
| self._client = client | |
| self._timeout = timeout | |
| self._cache: dict[str, FullText] = {} | |
| self._locks: dict[str, asyncio.Lock] = {} | |
| async def _http(self) -> httpx.AsyncClient: | |
| if self._client is None: | |
| self._client = httpx.AsyncClient(timeout=self._timeout) | |
| return self._client | |
| async def close(self) -> None: | |
| if self._client is not None: | |
| await self._client.aclose() | |
| self._client = None | |
| async def locate(self, doi: str = "", pmid: str = "") -> dict: | |
| """The Europe PMC record for a work, which carries its PMCID when one exists.""" | |
| doi = (doi or "").replace("https://doi.org/", "").strip() | |
| query = f'DOI:"{doi}"' if doi else f"EXT_ID:{pmid} AND SRC:MED" if pmid else "" | |
| if not query: | |
| return {} | |
| client = await self._http() | |
| response = await client.get( | |
| f"{EUROPE_PMC}/search", | |
| params={"query": query, "format": "json", "resultType": "core", "pageSize": 1}, | |
| ) | |
| response.raise_for_status() | |
| results = response.json().get("resultList", {}).get("result", []) | |
| return results[0] if results else {} | |
| async def fetch(self, paper_id: str, doi: str = "", pmid: str = "", abstract: str = "") -> FullText: | |
| lock = self._locks.setdefault(paper_id, asyncio.Lock()) | |
| async with lock: | |
| if paper_id in self._cache: | |
| return self._cache[paper_id] | |
| result = await self._fetch(paper_id, doi, pmid, abstract) | |
| self._cache[paper_id] = result | |
| return result | |
| async def _fetch(self, paper_id: str, doi: str, pmid: str, abstract: str) -> FullText: | |
| fallback = FullText( | |
| paper_id, | |
| "abstract_only" if abstract else "unavailable", | |
| sections={"abstract": abstract} if abstract else {}, | |
| ) | |
| try: | |
| record = await self.locate(doi, pmid) | |
| except (httpx.HTTPError, ValueError) as exc: | |
| logger.warning("Full-text lookup failed for %s: %s", paper_id, type(exc).__name__) | |
| return fallback | |
| # Europe PMC titles keep inline markup such as <sub> from the publisher record. | |
| fallback.title = _clean(re.sub(r"<[^>]+>", "", str(record.get("title") or ""))) | |
| fallback.year = int(year) if (year := str(record.get("pubYear") or "")).isdigit() else None | |
| journal = record.get("journalInfo") | |
| title = journal.get("journal", {}).get("title") if isinstance(journal, dict) else "" | |
| fallback.venue = str(title or record.get("journalTitle") or "") | |
| pmcid = record.get("pmcid") | |
| if not pmcid or record.get("isOpenAccess") != "Y": | |
| return fallback | |
| client = await self._http() | |
| try: | |
| response = await client.get(f"{EUROPE_PMC}/{pmcid}/fullTextXML") | |
| response.raise_for_status() | |
| except httpx.HTTPError as exc: | |
| logger.info("No open body for %s (%s)", pmcid, type(exc).__name__) | |
| return fallback | |
| sections = parse_jats(response.text) | |
| if not sections: | |
| return fallback | |
| if abstract and "abstract" not in sections: | |
| sections["abstract"] = abstract | |
| return FullText( | |
| paper_id, | |
| "full_text", | |
| source=f"europepmc:{pmcid}", | |
| sections=sections, | |
| title=fallback.title, | |
| year=fallback.year, | |
| venue=fallback.venue, | |
| ) | |