agent_search_gateway.providers.academic.crossref

  1"""Crossref works API discovery adapter."""
  2
  3from __future__ import annotations
  4
  5import html
  6import re
  7from datetime import date
  8
  9from ...academic.normalization import normalize_doi
 10from ...observability import SecretValue
 11from ..contracts import PaperSearchHit
 12from .common import (
 13    AcademicHttpExecutor,
 14    as_list,
 15    as_mapping,
 16    join_url,
 17    nonnegative_int,
 18    protocol_failure,
 19    reject_item,
 20    text,
 21)
 22
 23_DEFAULT_API_URL = "https://api.crossref.org"
 24_TAG_RE = re.compile(r"<[^>]+>")
 25_DATE_KEYS = ("published-print", "published-online", "published", "issued", "created")
 26
 27
 28class CrossrefProvider:
 29    name = "crossref"
 30
 31    def __init__(
 32        self,
 33        executor: AcademicHttpExecutor,
 34        *,
 35        api_url: str = _DEFAULT_API_URL,
 36        contact_email: SecretValue | None = None,
 37    ) -> None:
 38        self._executor = executor
 39        self._api_url = api_url
 40        self._contact = contact_email
 41
 42    async def search(self, query: str) -> list[PaperSearchHit]:
 43        params: dict[str, object] = {
 44            "query": query,
 45            "rows": 10,
 46            "sort": "relevance",
 47            "order": "desc",
 48        }
 49        if self._contact is not None:
 50            reveal = self._contact.reveal
 51            params["mailto"] = reveal()
 52        payload = await self._executor.request_json(
 53            "GET",
 54            join_url(self._api_url, "works"),
 55            stage="paper_search",
 56            params=params,
 57        )
 58        envelope = as_mapping(payload)
 59        message = as_mapping(envelope.get("message")) if envelope is not None else None
 60        items = as_list(message.get("items")) if message is not None else None
 61        if items is None:
 62            raise protocol_failure(self.name, "response message.items envelope was invalid")
 63        hits: list[PaperSearchHit] = []
 64        for item in items:
 65            mapped = self._map_item(item)
 66            if mapped is None:
 67                reject_item(self.name)
 68            else:
 69                hits.append(mapped)
 70        return hits
 71
 72    def _map_item(self, value: object) -> PaperSearchHit | None:
 73        item = as_mapping(value)
 74        if item is None:
 75            return None
 76        doi = normalize_doi(text(item.get("DOI")))
 77        title = self._first_string(item.get("title"))
 78        if doi is None or not title:
 79            return None
 80        authors = self._authors(item.get("author"))
 81        venue = self._first_string(item.get("container-title"))
 82        url = text(item.get("URL")) or f"https://doi.org/{doi}"
 83        return PaperSearchHit(
 84            source=self.name,
 85            source_id=doi,
 86            title=title,
 87            authors=authors,
 88            abstract=self._abstract(item.get("abstract")),
 89            doi=doi,
 90            published_date=self._date(item),
 91            url=url,
 92            pdf_url=self._pdf_url(item.get("link")),
 93            venue=venue,
 94            citation_count=nonnegative_int(item.get("is-referenced-by-count")),
 95        )
 96
 97    @staticmethod
 98    def _first_string(value: object) -> str:
 99        values = as_list(value) or []
100        for raw in values:
101            candidate = text(raw)
102            if candidate:
103                return candidate
104        return ""
105
106    @staticmethod
107    def _authors(value: object) -> tuple[str, ...]:
108        authors: list[str] = []
109        for raw_author in as_list(value) or []:
110            author = as_mapping(raw_author)
111            if author is None:
112                continue
113            given = text(author.get("given"))
114            family = text(author.get("family"))
115            name = " ".join(part for part in (given, family) if part)
116            if name:
117                authors.append(name)
118        return tuple(authors)
119
120    @staticmethod
121    def _abstract(value: object) -> str:
122        raw = text(value)
123        if not raw:
124            return ""
125        return " ".join(html.unescape(_TAG_RE.sub(" ", raw)).split())
126
127    @classmethod
128    def _date(cls, item: object) -> date | None:
129        mapping = as_mapping(item)
130        if mapping is None:
131            return None
132        for key in _DATE_KEYS:
133            date_mapping = as_mapping(mapping.get(key))
134            parts_rows = (
135                as_list(date_mapping.get("date-parts")) if date_mapping is not None else None
136            )
137            if not parts_rows:
138                continue
139            parts = as_list(parts_rows[0])
140            if not parts or isinstance(parts[0], bool) or not isinstance(parts[0], int):
141                continue
142            year = parts[0]
143            month = (
144                parts[1]
145                if len(parts) > 1 and isinstance(parts[1], int) and not isinstance(parts[1], bool)
146                else 1
147            )
148            day = (
149                parts[2]
150                if len(parts) > 2 and isinstance(parts[2], int) and not isinstance(parts[2], bool)
151                else 1
152            )
153            try:
154                return date(year, month, day)
155            except ValueError:
156                continue
157        return None
158
159    @staticmethod
160    def _pdf_url(value: object) -> str:
161        for raw_link in as_list(value) or []:
162            link = as_mapping(raw_link)
163            if link is None:
164                continue
165            candidate = text(link.get("URL"))
166            content_type = text(link.get("content-type")).casefold()
167            if candidate and content_type == "application/pdf":
168                return candidate
169        return ""