agent_search_gateway.providers.academic.core

  1"""CORE works search API discovery adapter."""
  2
  3from __future__ import annotations
  4
  5from ...academic.normalization import normalize_core_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.core.ac.uk/v3"
 21
 22
 23class CoreProvider:
 24    name = "core"
 25
 26    def __init__(
 27        self,
 28        executor: AcademicHttpExecutor,
 29        *,
 30        api_key: SecretValue,
 31        api_url: str = _DEFAULT_API_URL,
 32    ) -> None:
 33        self._executor = executor
 34        self._api_url = api_url
 35        self._credential = api_key
 36
 37    async def search(self, query: str) -> list[PaperSearchHit]:
 38        payload = await self._executor.request_json(
 39            "GET",
 40            join_url(self._api_url, "search/works/"),
 41            stage="paper_search",
 42            headers=self._request_headers(),
 43            params={"q": query, "limit": 10, "offset": 0},
 44        )
 45        envelope = as_mapping(payload)
 46        results = as_list(envelope.get("results")) if envelope is not None else None
 47        if results is None:
 48            raise protocol_failure(self.name, "response results envelope was invalid")
 49        hits: list[PaperSearchHit] = []
 50        for item in results:
 51            mapped = self._map_item(item)
 52            if mapped is None:
 53                reject_item(self.name)
 54            else:
 55                hits.append(mapped)
 56        return hits
 57
 58    def _request_headers(self) -> dict[str, str]:
 59        reveal = self._credential.reveal
 60        return {"Authorization": f"Bearer {reveal()}"}
 61
 62    def _map_item(self, value: object) -> PaperSearchHit | None:
 63        item = as_mapping(value)
 64        if item is None:
 65            return None
 66        raw_id = item.get("id")
 67        source_id = normalize_core_id(str(raw_id)) if isinstance(raw_id, str | int) else None
 68        title = text(item.get("title"))
 69        if source_id is None or not title:
 70            return None
 71        repository = as_mapping(item.get("repository")) or {}
 72        return PaperSearchHit(
 73            source=self.name,
 74            source_id=source_id,
 75            title=title,
 76            authors=self._authors(item.get("authors")),
 77            abstract=text(item.get("abstract")),
 78            doi=text(item.get("doi")),
 79            published_date=parse_iso_date(item.get("publishedDate")),
 80            url=f"https://core.ac.uk/works/{source_id}",
 81            pdf_url=self._pdf_url(item),
 82            venue=text(repository.get("name")),
 83            topics=self._topics(item),
 84            citation_count=nonnegative_int(item.get("citationCount")),
 85        )
 86
 87    @staticmethod
 88    def _authors(value: object) -> tuple[str, ...]:
 89        authors: list[str] = []
 90        for raw_author in as_list(value) or []:
 91            if isinstance(raw_author, str):
 92                name = raw_author.strip()
 93            else:
 94                author = as_mapping(raw_author)
 95                name = text(author.get("name")) if author is not None else ""
 96            if name:
 97                authors.append(name)
 98        return tuple(authors)
 99
100    @staticmethod
101    def _pdf_url(item: object) -> str:
102        mapping = as_mapping(item)
103        if mapping is None:
104            return ""
105        direct = text(mapping.get("downloadUrl"))
106        if direct:
107            return direct
108        for raw_url in as_list(mapping.get("fullTextUrls")) or []:
109            candidate = text(raw_url)
110            if candidate:
111                return candidate
112        return ""
113
114    @staticmethod
115    def _topics(item: object) -> tuple[str, ...]:
116        mapping = as_mapping(item)
117        if mapping is None:
118            return ()
119        topics: list[str] = []
120        seen: set[str] = set()
121        for key in ("subjects", "tags"):
122            for raw_topic in as_list(mapping.get(key)) or []:
123                topic = text(raw_topic)
124                folded = topic.casefold()
125                if topic and folded not in seen:
126                    seen.add(folded)
127                    topics.append(topic)
128        return tuple(topics)