agent_search_gateway.providers.academic.semantic_scholar

  1"""Semantic Scholar Graph API discovery adapter."""
  2
  3from __future__ import annotations
  4
  5from ...observability import SecretValue
  6from ..contracts import PaperSearchHit
  7from .common import (
  8    AcademicHttpExecutor,
  9    as_list,
 10    as_mapping,
 11    join_url,
 12    nonnegative_int,
 13    parse_iso_date,
 14    protocol_failure,
 15    reject_item,
 16    text,
 17)
 18
 19_DEFAULT_API_URL = "https://api.semanticscholar.org/graph/v1"
 20_FIELDS = (
 21    "title,abstract,citationCount,authors,url,publicationDate,"
 22    "externalIds,fieldsOfStudy,openAccessPdf"
 23)
 24
 25
 26class SemanticScholarProvider:
 27    name = "semantic_scholar"
 28
 29    def __init__(
 30        self,
 31        executor: AcademicHttpExecutor,
 32        *,
 33        api_url: str = _DEFAULT_API_URL,
 34        api_key: SecretValue | None = None,
 35    ) -> None:
 36        self._executor = executor
 37        self._api_url = api_url
 38        self._credential = api_key
 39
 40    async def search(self, query: str) -> list[PaperSearchHit]:
 41        payload = await self._executor.request_json(
 42            "GET",
 43            join_url(self._api_url, "paper/search"),
 44            stage="paper_search",
 45            headers=self._request_headers(),
 46            params={"query": query, "limit": 10, "fields": _FIELDS},
 47        )
 48        envelope = as_mapping(payload)
 49        if envelope is None:
 50            reason = "invalid_data_envelope"
 51            raise protocol_failure(self.name, reason, reason=reason)
 52        if "data" not in envelope:
 53            reason = "missing_data_envelope"
 54            raise protocol_failure(self.name, reason, reason=reason)
 55        data = as_list(envelope.get("data"))
 56        if data is None:
 57            reason = "invalid_data_envelope"
 58            raise protocol_failure(self.name, reason, reason=reason)
 59        hits: list[PaperSearchHit] = []
 60        for item in data:
 61            mapped = self._map_item(item)
 62            if mapped is None:
 63                reject_item(self.name)
 64            else:
 65                hits.append(mapped)
 66        return hits
 67
 68    def _request_headers(self) -> dict[str, str] | None:
 69        if self._credential is None:
 70            return None
 71        reveal = self._credential.reveal
 72        return {"x-api-key": reveal()}
 73
 74    def _map_item(self, value: object) -> PaperSearchHit | None:
 75        item = as_mapping(value)
 76        if item is None:
 77            return None
 78        paper_id = text(item.get("paperId"))
 79        title = text(item.get("title"))
 80        url = text(item.get("url"))
 81        if not paper_id or not title or not url:
 82            return None
 83        external = as_mapping(item.get("externalIds")) or {}
 84        authors_raw = as_list(item.get("authors")) or []
 85        authors = tuple(
 86            name
 87            for author in authors_raw
 88            if (mapping := as_mapping(author)) is not None and (name := text(mapping.get("name")))
 89        )
 90        topics = tuple(
 91            topic
 92            for raw_topic in (as_list(item.get("fieldsOfStudy")) or [])
 93            if (topic := text(raw_topic))
 94        )
 95        oa = as_mapping(item.get("openAccessPdf"))
 96        pdf_url = text(oa.get("url")) if oa is not None else ""
 97        return PaperSearchHit(
 98            source=self.name,
 99            source_id=paper_id,
100            title=title,
101            authors=authors,
102            abstract=text(item.get("abstract")),
103            doi=text(external.get("DOI")),
104            arxiv_id=text(external.get("ArXiv")),
105            published_date=parse_iso_date(item.get("publicationDate")),
106            url=url,
107            pdf_url=pdf_url,
108            topics=topics,
109            citation_count=nonnegative_int(item.get("citationCount")),
110            is_open_access=True if pdf_url else None,
111        )