agent_search_gateway.providers.academic.arxiv

  1"""arXiv Atom API discovery adapter."""
  2
  3from __future__ import annotations
  4
  5import xml.etree.ElementTree as ET
  6
  7from ...academic.normalization import normalize_arxiv_id, normalize_doi
  8from ..contracts import PaperSearchHit
  9from .common import AcademicHttpExecutor, parse_iso_date, protocol_failure, reject_item
 10
 11_ATOM = "http://www.w3.org/2005/Atom"
 12_ARXIV = "http://arxiv.org/schemas/atom"
 13_DEFAULT_API_URL = "https://export.arxiv.org/api/query"
 14
 15
 16class ArxivProvider:
 17    name = "arxiv"
 18
 19    def __init__(
 20        self,
 21        executor: AcademicHttpExecutor,
 22        *,
 23        api_url: str = _DEFAULT_API_URL,
 24    ) -> None:
 25        self._executor = executor
 26        self._api_url = api_url
 27
 28    async def search(self, query: str) -> list[PaperSearchHit]:
 29        body = await self._executor.request_text(
 30            "GET",
 31            self._api_url,
 32            stage="paper_search",
 33            params={
 34                "search_query": f"all:{query}",
 35                "max_results": 10,
 36                "sortBy": "relevance",
 37                "sortOrder": "descending",
 38            },
 39        )
 40        try:
 41            root = ET.fromstring(body)
 42        except ET.ParseError as exc:
 43            raise protocol_failure(self.name, "response was not valid Atom XML") from exc
 44        if root.tag != f"{{{_ATOM}}}feed":
 45            raise protocol_failure(self.name, "response Atom envelope was invalid")
 46
 47        hits: list[PaperSearchHit] = []
 48        for entry in root.findall(f"{{{_ATOM}}}entry"):
 49            mapped = self._map_entry(entry)
 50            if mapped is None:
 51                reject_item(self.name)
 52                continue
 53            hits.append(mapped)
 54        return hits
 55
 56    def _map_entry(self, entry: ET.Element) -> PaperSearchHit | None:
 57        raw_id = self._child_text(entry, _ATOM, "id")
 58        source_id = normalize_arxiv_id(raw_id)
 59        title = self._clean(self._child_text(entry, _ATOM, "title"))
 60        if source_id is None or not title:
 61            return None
 62
 63        landing_url = raw_id
 64        pdf_url = ""
 65        for link in entry.findall(f"{{{_ATOM}}}link"):
 66            href = (link.get("href") or "").strip()
 67            if not href:
 68                continue
 69            if link.get("rel") == "alternate":
 70                landing_url = href
 71            if link.get("title") == "pdf" or link.get("type") == "application/pdf":
 72                pdf_url = href
 73        if not landing_url:
 74            return None
 75
 76        authors = tuple(
 77            name
 78            for author in entry.findall(f"{{{_ATOM}}}author")
 79            if (name := self._clean(self._child_text(author, _ATOM, "name")))
 80        )
 81        topics = tuple(
 82            term
 83            for category in entry.findall(f"{{{_ATOM}}}category")
 84            if (term := (category.get("term") or "").strip())
 85        )
 86        return PaperSearchHit(
 87            source=self.name,
 88            source_id=source_id,
 89            title=title,
 90            authors=authors,
 91            abstract=self._clean(self._child_text(entry, _ATOM, "summary")),
 92            doi=normalize_doi(self._child_text(entry, _ARXIV, "doi")) or "",
 93            arxiv_id=source_id,
 94            published_date=parse_iso_date(self._child_text(entry, _ATOM, "published")),
 95            updated_date=parse_iso_date(self._child_text(entry, _ATOM, "updated")),
 96            url=landing_url,
 97            pdf_url=pdf_url,
 98            topics=topics,
 99        )
100
101    @staticmethod
102    def _child_text(parent: ET.Element, namespace: str, name: str) -> str:
103        child = parent.find(f"{{{namespace}}}{name}")
104        return child.text if child is not None and child.text is not None else ""
105
106    @staticmethod
107    def _clean(value: str) -> str:
108        return " ".join(value.split())