feat: more robust OA resolution (all OA locations + arXiv DOIs)

instsci/fetcher.py

@@ -356,6 +356,11 @@ def _try_open_access(self, doi: str) -> Paper | None:
         Priority: arXiv PDF > OA PDF > OA HTML.
         If HTML extraction is too short, attempt PDF fallback from HTML page.
         """
+        arxiv_id = self._arxiv_id_from_doi(doi)
+        if arxiv_id:
+            logger.info("arXiv DOI detected (%s); routing to arXiv source.", doi)
+            return self._fetch_arxiv(arxiv_id, Paper(doi=doi))
+
         logger.info("Checking Unpaywall for OA version of %s...", doi)
         oa = unpaywall.check_oa(doi, email=self.config.email)
 
@@ -379,13 +384,17 @@ def _try_open_access(self, doi: str) -> Paper | None:
         if arxiv_id:
             return self._fetch_arxiv(arxiv_id, paper)
 
-        # Try direct OA PDF download FIRST (always prefer PDF over HTML)
-        if oa.pdf_url:
-            logger.info("Downloading OA PDF: %s", oa.pdf_url)
+        # Try direct OA PDF download FIRST (always prefer PDF over HTML).
+        # Iterate every candidate location so a forbidden/broken publisher
+        # link (e.g. a 403 on the publisher copy) can fall back to a
+        # repository copy rather than escalating prematurely.
+        pdf_candidates = oa.pdf_urls or ([oa.pdf_url] if oa.pdf_url else [])
+        for pdf_url in pdf_candidates:
+            logger.info("Downloading OA PDF: %s", pdf_url)
             paper.source = "open_access"
             self._rate_limit()
             try:
-                resp = request_with_retry("GET", oa.pdf_url, timeout=60, stream=True)
+                resp = request_with_retry("GET", pdf_url, timeout=60, stream=True)
                 resp.raise_for_status()
                 ct = resp.headers.get("content-type", "").lower()
                 if "pdf" in ct:
@@ -474,6 +483,20 @@ def _fetch_arxiv(self, arxiv_id: str, paper: Paper) -> Paper:
 
         return paper
 
+    @staticmethod
+    def _arxiv_id_from_doi(doi: str) -> str | None:
+        """Return the arXiv ID when the DOI is an arXiv DataCite DOI.
+
+        arXiv registers DOIs of the form ``10.48550/arXiv.<id>`` that Unpaywall
+        does not index, so they must be routed to the arXiv source directly
+        instead of falling through to institutional access.
+        """
+        low = (doi or "").strip().lower()
+        prefix = "10.48550/arxiv."
+        if low.startswith(prefix):
+            return doi.strip()[len(prefix):] or None
+        return None
+
     @staticmethod
     def _build_publisher_pdf_url(doi: str, resolved_url: str) -> str | None:
         """Construct a direct PDF URL from known publisher patterns.

instsci/sources/unpaywall.py

@@ -1,7 +1,7 @@
 """Open Access detection via Unpaywall API."""
 
 import logging
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 import requests
 
@@ -24,6 +24,23 @@ class OAResult:
     authors: list[str] | None = None
     journal: str = ""
     year: int | None = None
+    pdf_urls: list[str] = field(default_factory=list)
+
+
+def collect_pdf_urls(best_oa: dict, oa_locations: list[dict]) -> list[str]:
+    """Return ordered, de-duplicated candidate PDF URLs.
+
+    The best OA location comes first, followed by every other OA location, so a
+    caller can fall back to a repository copy when the preferred (often
+    publisher) link is forbidden or broken.
+    """
+    urls: list[str] = []
+    candidates = [best_oa, *oa_locations] if best_oa else list(oa_locations)
+    for loc in candidates:
+        pdf = (loc.get("url_for_pdf") or "").strip()
+        if pdf and pdf not in urls:
+            urls.append(pdf)
+    return urls
 
 
 def check_oa(doi: str, email: str = "instsci@example.com") -> OAResult:
@@ -88,12 +105,10 @@ def check_oa(doi: str, email: str = "instsci@example.com") -> OAResult:
         else:
             result.source = "repository"
 
-    # If no PDF from best location, scan all locations
-    if not result.pdf_url:
-        for loc in oa_locations:
-            pdf = loc.get("url_for_pdf", "") or ""
-            if pdf:
-                result.pdf_url = pdf
-                break
+    # Collect every candidate PDF URL (best first, then all locations) so a
+    # forbidden or broken link can fall back to a repository copy.
+    result.pdf_urls = collect_pdf_urls(best_oa, oa_locations)
+    if not result.pdf_url and result.pdf_urls:
+        result.pdf_url = result.pdf_urls[0]
 
     return result

tests/test_arxiv_doi_routing.py

@@ -0,0 +1,38 @@
+import unittest
+
+from instsci.fetcher import PaperFetcher
+
+
+class ArxivDoiRoutingTests(unittest.TestCase):
+    """arXiv DataCite DOIs (10.48550/arXiv.*) must route to the arXiv source.
+
+    Unpaywall does not index these, so without direct routing they escalate to
+    institutional access and never download.
+    """
+
+    def test_arxiv_datacite_doi(self):
+        self.assertEqual(
+            PaperFetcher._arxiv_id_from_doi("10.48550/arXiv.2504.06435"),
+            "2504.06435",
+        )
+
+    def test_case_insensitive_prefix(self):
+        self.assertEqual(
+            PaperFetcher._arxiv_id_from_doi("10.48550/ARXIV.2602.11522"),
+            "2602.11522",
+        )
+
+    def test_non_arxiv_doi_returns_none(self):
+        self.assertIsNone(
+            PaperFetcher._arxiv_id_from_doi("10.1080/10447318.2023.2301250")
+        )
+        self.assertIsNone(
+            PaperFetcher._arxiv_id_from_doi("10.1177/0018720810376055")
+        )
+
+    def test_blank(self):
+        self.assertIsNone(PaperFetcher._arxiv_id_from_doi(""))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_unpaywall_oa_locations.py

@@ -0,0 +1,38 @@
+import unittest
+
+from instsci.sources.unpaywall import collect_pdf_urls
+
+
+class CollectPdfUrlsTests(unittest.TestCase):
+    """Every OA location should be offered as a fallback candidate.
+
+    Regression: a forbidden publisher link must not hide an available
+    repository copy (best location first, then the rest).
+    """
+
+    def test_best_first_then_other_locations_deduped(self):
+        best = {"url_for_pdf": "https://pub.example/best.pdf", "host_type": "publisher"}
+        locs = [
+            {"url_for_pdf": "https://pub.example/best.pdf"},          # dup of best
+            {"url_for_pdf": "https://repo.example/copy.pdf"},         # repository fallback
+            {"url_for_pdf": ""},                                     # empty -> ignored
+            {"url_for_landing_page": "https://repo.example/landing"},  # no pdf -> ignored
+        ]
+        self.assertEqual(
+            collect_pdf_urls(best, locs),
+            ["https://pub.example/best.pdf", "https://repo.example/copy.pdf"],
+        )
+
+    def test_no_best_location(self):
+        locs = [{"url_for_pdf": "https://repo.example/a.pdf"}]
+        self.assertEqual(collect_pdf_urls({}, locs), ["https://repo.example/a.pdf"])
+
+    def test_empty(self):
+        self.assertEqual(collect_pdf_urls({}, []), [])
+
+    def test_strips_whitespace(self):
+        self.assertEqual(collect_pdf_urls({"url_for_pdf": "  x.pdf  "}, []), ["x.pdf"])
+
+
+if __name__ == "__main__":
+    unittest.main()