agent_search_gateway.providers.academic.unpaywall

  1"""Unpaywall DOI resolver used only for post-deduplication OA enrichment."""
  2
  3from __future__ import annotations
  4
  5from collections.abc import Mapping
  6from urllib.parse import quote
  7
  8from ...academic.normalization import normalize_doi
  9from ...errors import InputFailure
 10from ...models import OAResolution
 11from ...observability import SecretValue
 12from ...url_normalization import NormalizedURL, normalize_url
 13from ..http import HttpStatusFailure
 14from .common import AcademicHttpExecutor, as_list, as_mapping, join_url, protocol_failure, text
 15
 16_DEFAULT_API_URL = "https://api.unpaywall.org/v2"
 17
 18
 19class UnpaywallResolver:
 20    name = "unpaywall"
 21
 22    def __init__(
 23        self,
 24        executor: AcademicHttpExecutor,
 25        *,
 26        contact_email: SecretValue,
 27        api_url: str = _DEFAULT_API_URL,
 28    ) -> None:
 29        self._executor = executor
 30        self._contact = contact_email
 31        self._api_url = api_url
 32
 33    async def resolve(self, doi: str) -> OAResolution | None:
 34        canonical_doi = normalize_doi(doi)
 35        if canonical_doi is None:
 36            raise protocol_failure(
 37                self.name,
 38                "invalid DOI supplied to resolver",
 39                stage="oa_resolve",
 40            )
 41        reveal = self._contact.reveal
 42        encoded_doi = quote(canonical_doi, safe="")
 43        try:
 44            payload = await self._executor.request_json(
 45                "GET",
 46                join_url(self._api_url, encoded_doi),
 47                stage="oa_resolve",
 48                params={"email": reveal()},
 49            )
 50        except HttpStatusFailure as exc:
 51            if exc.status_code == 404:
 52                return None
 53            raise
 54        return self._map_payload(payload)
 55
 56    def _map_payload(self, payload: object) -> OAResolution:
 57        envelope = as_mapping(payload)
 58        raw_is_open_access = envelope.get("is_oa") if envelope is not None else None
 59        if envelope is None or not isinstance(raw_is_open_access, bool):
 60            raise protocol_failure(
 61                self.name,
 62                "response OA envelope was invalid",
 63                stage="oa_resolve",
 64            )
 65        is_open_access = raw_is_open_access
 66        oa_status = text(envelope.get("oa_status"))
 67        best_raw = envelope.get("best_oa_location")
 68        if best_raw is not None and not isinstance(best_raw, dict):
 69            raise protocol_failure(
 70                self.name,
 71                "response best OA location was invalid",
 72                stage="oa_resolve",
 73            )
 74        alternates_raw = envelope.get("oa_locations", [])
 75        alternates = as_list(alternates_raw)
 76        if alternates is None:
 77            raise protocol_failure(
 78                self.name,
 79                "response OA locations were invalid",
 80                stage="oa_resolve",
 81            )
 82        best = as_mapping(best_raw)
 83        chosen = self._choose_location(best, alternates)
 84        landing_url = self._location_url(chosen, landing=True)
 85        pdf_url = self._location_url(chosen, landing=False)
 86        license_value = text(chosen.get("license")) if chosen is not None else ""
 87        return OAResolution(
 88            landing_url=landing_url,
 89            pdf_url=pdf_url,
 90            is_open_access=is_open_access,
 91            oa_status=oa_status,
 92            license=license_value,
 93        )
 94
 95    def _choose_location(
 96        self,
 97        best: Mapping[str, object] | None,
 98        alternates: list[object],
 99    ) -> Mapping[str, object] | None:
100        if best is not None and self._location_has_url(best):
101            return best
102        candidates = [
103            mapping
104            for value in alternates
105            if (mapping := as_mapping(value)) is not None and self._location_has_url(mapping)
106        ]
107        candidates.sort(key=self._location_sort_key)
108        return candidates[0] if candidates else best
109
110    @classmethod
111    def _location_has_url(cls, location: Mapping[str, object]) -> bool:
112        return (
113            cls._location_url(location, landing=True) is not None
114            or cls._location_url(location, landing=False) is not None
115        )
116
117    @classmethod
118    def _location_sort_key(cls, location: Mapping[str, object]) -> tuple[int, str, str]:
119        pdf = cls._location_url(location, landing=False)
120        landing = cls._location_url(location, landing=True)
121        return (
122            0 if pdf is not None else 1,
123            str(pdf) if pdf is not None else "",
124            str(landing) if landing is not None else "",
125        )
126
127    @classmethod
128    def _location_url(
129        cls,
130        location: Mapping[str, object] | None,
131        *,
132        landing: bool,
133    ) -> NormalizedURL | None:
134        if location is None:
135            return None
136        raw_values = (
137            (
138                text(location.get("url_for_landing_page")),
139                text(location.get("url")),
140            )
141            if landing
142            else (text(location.get("url_for_pdf")),)
143        )
144        for raw in raw_values:
145            if not raw:
146                continue
147            try:
148                return normalize_url(raw)
149            except InputFailure:
150                continue
151        return None