d1ffea1c1c
Review-Gate finding: GROBID emits type="arXiv" (camel-case), not "arxiv". XPath literal match silently returned empty strings for all real responses. Fix: iterate idno elements and compare .lower() == "arxiv". Test fixture updated to reflect real GROBID output (type="arXiv"). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
135 lines
3.8 KiB
Python
135 lines
3.8 KiB
Python
"""GROBID integration.
|
|
|
|
Extracts structured reference lists and full-text TEI XML from PDFs by
|
|
calling a self-hosted GROBID HTTP server. No DB access, no embedding,
|
|
no network fetching of papers happens here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import httpx
|
|
|
|
from codex.config import Settings
|
|
|
|
_TEI_NS = "{http://www.tei-c.org/ns/1.0}"
|
|
|
|
|
|
def _text(element: ET.Element | None) -> str:
|
|
"""Return element text or empty string if element is None."""
|
|
if element is None:
|
|
return ""
|
|
return (element.text or "").strip()
|
|
|
|
|
|
def extract_references(
|
|
pdf_path: str,
|
|
grobid_url: str | None = None,
|
|
) -> list[dict[str, str]]:
|
|
"""Extract a structured reference list from a PDF via GROBID.
|
|
|
|
Parameters
|
|
----------
|
|
pdf_path:
|
|
Path to the PDF file on disk.
|
|
grobid_url:
|
|
Base URL of the GROBID server. Defaults to ``Settings().grobid_url``.
|
|
|
|
Returns
|
|
-------
|
|
list[dict[str, str]]
|
|
One dict per reference. All dicts contain the keys
|
|
``title``, ``authors``, ``year``, ``doi``, ``arxiv_id``
|
|
(missing values are empty strings).
|
|
"""
|
|
if grobid_url is None:
|
|
grobid_url = Settings().grobid_url
|
|
|
|
with open(pdf_path, "rb") as fh, httpx.Client(timeout=60.0) as client:
|
|
response = client.post(
|
|
f"{grobid_url}/api/processReferences",
|
|
files={"input": (pdf_path, fh, "application/pdf")},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
root = ET.fromstring(response.text)
|
|
results: list[dict[str, str]] = []
|
|
|
|
for bib in root.iter(f"{_TEI_NS}biblStruct"):
|
|
# Title
|
|
title_el = bib.find(f".//{_TEI_NS}title[@level='a']")
|
|
title = _text(title_el)
|
|
|
|
# Authors: collect all persName elements
|
|
authors_parts: list[str] = []
|
|
for person in bib.iter(f"{_TEI_NS}persName"):
|
|
forename_el = person.find(f"{_TEI_NS}forename")
|
|
surname_el = person.find(f"{_TEI_NS}surname")
|
|
forename = _text(forename_el)
|
|
surname = _text(surname_el)
|
|
full = " ".join(p for p in (forename, surname) if p)
|
|
if full:
|
|
authors_parts.append(full)
|
|
authors = "; ".join(authors_parts)
|
|
|
|
# Year
|
|
date_el = bib.find(f".//{_TEI_NS}date[@type='published']")
|
|
year = ""
|
|
if date_el is not None:
|
|
when = date_el.get("when", "")
|
|
year = when[:4] if when else ""
|
|
|
|
# DOI
|
|
doi_el = bib.find(f".//{_TEI_NS}idno[@type='DOI']")
|
|
doi = _text(doi_el)
|
|
|
|
# arXiv ID — GROBID emits type="arXiv" (camel-case); match case-insensitively
|
|
arxiv_id = ""
|
|
for idno_el in bib.iter(f"{_TEI_NS}idno"):
|
|
if idno_el.get("type", "").lower() == "arxiv":
|
|
arxiv_id = (idno_el.text or "").strip()
|
|
break
|
|
|
|
results.append(
|
|
{
|
|
"title": title,
|
|
"authors": authors,
|
|
"year": year,
|
|
"doi": doi,
|
|
"arxiv_id": arxiv_id,
|
|
}
|
|
)
|
|
|
|
return results
|
|
|
|
|
|
def extract_structure(
|
|
pdf_path: str,
|
|
grobid_url: str | None = None,
|
|
) -> str:
|
|
"""Extract full-text TEI XML from a PDF via GROBID.
|
|
|
|
Parameters
|
|
----------
|
|
pdf_path:
|
|
Path to the PDF file on disk.
|
|
grobid_url:
|
|
Base URL of the GROBID server. Defaults to ``Settings().grobid_url``.
|
|
|
|
Returns
|
|
-------
|
|
str
|
|
Raw TEI XML response text from GROBID.
|
|
"""
|
|
if grobid_url is None:
|
|
grobid_url = Settings().grobid_url
|
|
|
|
with open(pdf_path, "rb") as fh, httpx.Client(timeout=120.0) as client:
|
|
response = client.post(
|
|
f"{grobid_url}/api/processFulltextDocument",
|
|
files={"input": (pdf_path, fh, "application/pdf")},
|
|
)
|
|
response.raise_for_status()
|
|
return response.text
|