agent_search_gateway.providers.academic.openalex

  1"""OpenAlex works API discovery adapter."""
  2
  3from __future__ import annotations
  4
  5from ...academic.normalization import normalize_openalex_id
  6from ...observability import SecretValue
  7from ..contracts import PaperSearchHit
  8from .common import (
  9    AcademicHttpExecutor,
 10    as_list,
 11    as_mapping,
 12    join_url,
 13    nonnegative_int,
 14    parse_iso_date,
 15    protocol_failure,
 16    reject_item,
 17    text,
 18)
 19
 20_DEFAULT_API_URL = "https://api.openalex.org"
 21
 22
 23def reconstruct_abstract(value: object) -> str:
 24    """Reconstruct OpenAlex's inverted abstract index by numeric position."""
 25
 26    if not isinstance(value, dict):
 27        return ""
 28    positioned: list[tuple[int, str]] = []
 29    for raw_word, raw_positions in value.items():
 30        if not isinstance(raw_word, str) or not isinstance(raw_positions, list):
 31            continue
 32        for raw_position in raw_positions:
 33            if (
 34                isinstance(raw_position, bool)
 35                or not isinstance(raw_position, int)
 36                or raw_position < 0
 37            ):
 38                continue
 39            positioned.append((raw_position, raw_word))
 40    positioned.sort(key=lambda item: (item[0], item[1]))
 41    return " ".join(word for _, word in positioned)
 42
 43
 44class OpenAlexProvider:
 45    name = "openalex"
 46
 47    def __init__(
 48        self,
 49        executor: AcademicHttpExecutor,
 50        *,
 51        api_url: str = _DEFAULT_API_URL,
 52        contact_email: SecretValue | None = None,
 53    ) -> None:
 54        self._executor = executor
 55        self._api_url = api_url
 56        self._contact = contact_email
 57
 58    async def search(self, query: str) -> list[PaperSearchHit]:
 59        params: dict[str, object] = {"search": query, "per_page": 10}
 60        if self._contact is not None:
 61            reveal = self._contact.reveal
 62            params["mailto"] = reveal()
 63        payload = await self._executor.request_json(
 64            "GET",
 65            join_url(self._api_url, "works"),
 66            stage="paper_search",
 67            params=params,
 68        )
 69        envelope = as_mapping(payload)
 70        results = as_list(envelope.get("results")) if envelope is not None else None
 71        if results is None:
 72            raise protocol_failure(self.name, "response results envelope was invalid")
 73        hits: list[PaperSearchHit] = []
 74        for item in results:
 75            mapped = self._map_item(item)
 76            if mapped is None:
 77                reject_item(self.name)
 78            else:
 79                hits.append(mapped)
 80        return hits
 81
 82    def _map_item(self, value: object) -> PaperSearchHit | None:
 83        item = as_mapping(value)
 84        if item is None:
 85            return None
 86        source_id = normalize_openalex_id(text(item.get("id")))
 87        title = text(item.get("title"))
 88        if source_id is None or not title:
 89            return None
 90        primary = as_mapping(item.get("primary_location")) or {}
 91        oa = as_mapping(item.get("open_access")) or {}
 92        landing = text(primary.get("landing_page_url")) or f"https://openalex.org/{source_id}"
 93        pdf_url = text(primary.get("pdf_url"))
 94        authors = self._authors(item.get("authorships"))
 95        topics = self._concepts(item.get("concepts"))
 96        raw_is_oa = oa.get("is_oa")
 97        is_oa = raw_is_oa if isinstance(raw_is_oa, bool) else None
 98        return PaperSearchHit(
 99            source=self.name,
100            source_id=source_id,
101            title=title,
102            authors=authors,
103            abstract=reconstruct_abstract(item.get("abstract_inverted_index")),
104            doi=text(item.get("doi")),
105            published_date=parse_iso_date(item.get("publication_date")),
106            url=landing,
107            pdf_url=pdf_url,
108            topics=topics,
109            citation_count=nonnegative_int(item.get("cited_by_count")),
110            is_open_access=is_oa,
111            oa_status=text(oa.get("oa_status")),
112            license=text(primary.get("license")),
113        )
114
115    @staticmethod
116    def _authors(value: object) -> tuple[str, ...]:
117        result: list[str] = []
118        for raw_authorship in as_list(value) or []:
119            authorship = as_mapping(raw_authorship)
120            author = as_mapping(authorship.get("author")) if authorship is not None else None
121            name = text(author.get("display_name")) if author is not None else ""
122            if name:
123                result.append(name)
124        return tuple(result)
125
126    @staticmethod
127    def _concepts(value: object) -> tuple[str, ...]:
128        result: list[str] = []
129        for raw_concept in as_list(value) or []:
130            concept = as_mapping(raw_concept)
131            name = text(concept.get("display_name")) if concept is not None else ""
132            if name:
133                result.append(name)
134        return tuple(result)