agent_search_gateway.providers.academic.dblp

  1"""dblp publication search API adapter."""
  2
  3from __future__ import annotations
  4
  5import xml.etree.ElementTree as ET
  6from datetime import date
  7
  8from ...academic.normalization import normalize_dblp_key, normalize_doi
  9from ..contracts import PaperSearchHit
 10from .common import AcademicHttpExecutor, protocol_failure, reject_item
 11
 12_DEFAULT_API_URL = "https://dblp.org/search/publ/api"
 13
 14
 15class DblpProvider:
 16    name = "dblp"
 17
 18    def __init__(
 19        self,
 20        executor: AcademicHttpExecutor,
 21        *,
 22        api_url: str = _DEFAULT_API_URL,
 23    ) -> None:
 24        self._executor = executor
 25        self._api_url = api_url
 26
 27    async def search(self, query: str) -> list[PaperSearchHit]:
 28        body = await self._executor.request_text(
 29            "GET",
 30            self._api_url,
 31            stage="paper_search",
 32            params={"q": query, "format": "xml", "h": 10},
 33        )
 34        try:
 35            root = ET.fromstring(body)
 36        except ET.ParseError as exc:
 37            raise protocol_failure(self.name, "response was not valid XML") from exc
 38        if root.tag != "result" or root.find("hits") is None:
 39            raise protocol_failure(self.name, "response XML envelope was invalid")
 40        hits: list[PaperSearchHit] = []
 41        for hit in root.findall("./hits/hit"):
 42            mapped = self._map_hit(hit)
 43            if mapped is None:
 44                reject_item(self.name)
 45            else:
 46                hits.append(mapped)
 47        return hits
 48
 49    def _map_hit(self, hit: ET.Element) -> PaperSearchHit | None:
 50        info = hit.find("info")
 51        if info is None:
 52            return None
 53        title_element = info.find("title")
 54        title = self._element_text(title_element)
 55        url = self._child_text(info, "url")
 56        source_id = normalize_dblp_key((info.get("key") or "").strip())
 57        if source_id is None:
 58            source_id = normalize_dblp_key(url)
 59        if source_id is None or not title or not url:
 60            return None
 61        authors = tuple(
 62            author
 63            for element in info.findall("./authors/author")
 64            if (author := self._element_text(element))
 65        )
 66        doi = ""
 67        for edition in info.findall("ee"):
 68            candidate = normalize_doi(self._element_text(edition))
 69            if candidate is not None:
 70                doi = candidate
 71                break
 72        published_date = self._year_date(self._child_text(info, "year"))
 73        return PaperSearchHit(
 74            source=self.name,
 75            source_id=source_id,
 76            title=title,
 77            authors=authors,
 78            doi=doi,
 79            published_date=published_date,
 80            url=url,
 81            venue=self._child_text(info, "venue"),
 82        )
 83
 84    @staticmethod
 85    def _element_text(element: ET.Element | None) -> str:
 86        if element is None:
 87            return ""
 88        return " ".join("".join(element.itertext()).split())
 89
 90    @classmethod
 91    def _child_text(cls, parent: ET.Element, name: str) -> str:
 92        return cls._element_text(parent.find(name))
 93
 94    @staticmethod
 95    def _year_date(raw_year: str) -> date | None:
 96        try:
 97            year = int(raw_year)
 98            if year < 1:
 99                return None
100            return date(year, 1, 1)
101        except (TypeError, ValueError):
102            return None