feat: resolve library accessions against the GNPS2 library server

metabolomics_spectrum_resolver/parsing.py

@@ -357,6 +357,9 @@ def _parse_gnps2(usi: str) -> Tuple[sus.MsmsSpectrum, str]:
     ms_run = match.group(2)
     if ms_run.lower().startswith("task"):
         return _parse_gnps2_task(usi)
+    elif match.group(3).lower() == "accession":
+        # A GNPS2 library accession (e.g. GNPS2LIB...).
+        return _parse_gnps2_library(usi)
     else:
         # We are likely dealing with a dataset on the GNPS2 side
         return _parse_gnps2_dataset(usi)
@@ -561,7 +564,49 @@ def _parse_tinymass(usi: str) -> Tuple[sus.MsmsSpectrum, str]:
     except (requests.exceptions.HTTPError, json.decoder.JSONDecodeError):
         raise UsiError("Unknown Tiny Mass task USI", 404)
 
-# Parse GNPS library.
+# Fetch a gnpsspectrum-shaped record from the first responding URL and build the
+# spectrum. library.gnps2.org and external.gnps2.org both serve this shape.
+# Raises UsiError(404) if none of the URLs return a usable spectrum.
+def _fetch_gnps_library_spectrum(
+    usi: str, request_urls
+) -> sus.MsmsSpectrum:
+    for request_url in request_urls:
+        try:
+            lookup_request = requests.get(request_url, timeout=timeout)
+            lookup_request.raise_for_status()
+            spectrum_dict = lookup_request.json()
+            if spectrum_dict["spectruminfo"]["peaks_json"] == "null":
+                continue
+            mz, intensity = zip(
+                *json.loads(spectrum_dict["spectruminfo"]["peaks_json"])
+            )
+            # Use the most up-to-date spectrum annotation.
+            annotations = sorted(
+                spectrum_dict["annotations"],
+                key=lambda annotation: datetime.datetime.strptime(
+                    annotation["create_time"], "%Y-%m-%d %H:%M:%S.%f"
+                ),
+                reverse=True,
+            )[0]
+            return sus.MsmsSpectrum(
+                usi,
+                float(annotations["Precursor_MZ"]),
+                int(annotations["Charge"]),
+                mz,
+                intensity,
+            )
+        except (
+            requests.exceptions.RequestException,
+            json.decoder.JSONDecodeError,
+            KeyError,
+            ValueError,
+        ):
+            continue
+    raise UsiError("Unknown GNPS library USI", 404)
+
+
+# Parse GNPS library (legacy CCMSLIB accessions). Resolve against our GNPS2
+# library server first, then fall back to external.gnps2.org.
 def _parse_gnps_library(usi: str) -> Tuple[sus.MsmsSpectrum, str]:
     match = _match_usi(usi)
     index_flag = match.group(3)
@@ -570,42 +615,34 @@ def _parse_gnps_library(usi: str) -> Tuple[sus.MsmsSpectrum, str]:
             "Currently supported GNPS library index flags: accession", 400
         )
     index = match.group(4)
-    try:
-        request_url = (
-            f"https://external.gnps2.org/"
-            f"gnpsspectrum?SpectrumID={index}"
-        )
-        lookup_request = requests.get(request_url, timeout=timeout)
-        lookup_request.raise_for_status()
-        spectrum_dict = lookup_request.json()
-        if spectrum_dict["spectruminfo"]["peaks_json"] == "null":
-            raise UsiError("Unknown GNPS library USI", 404)
-        mz, intensity = zip(
-            *json.loads(spectrum_dict["spectruminfo"]["peaks_json"])
-        )
-        source_link = (
-            f"https://gnps.ucsd.edu/ProteoSAFe/"
-            f"gnpslibraryspectrum.jsp?SpectrumID={index}"
-        )
+    spectrum = _fetch_gnps_library_spectrum(
+        usi,
+        [
+            f"https://library.gnps2.org/gnpsspectrum?SpectrumID={index}",
+            f"https://external.gnps2.org/gnpsspectrum?SpectrumID={index}",
+        ],
+    )
+    source_link = (
+        f"https://gnps.ucsd.edu/ProteoSAFe/"
+        f"gnpslibraryspectrum.jsp?SpectrumID={index}"
+    )
+    return spectrum, source_link
 
-        # Use the most up-to-date spectrum annotation.
-        annotations = sorted(
-            spectrum_dict["annotations"],
-            key=lambda annotation: datetime.datetime.strptime(
-                annotation["create_time"], "%Y-%m-%d %H:%M:%S.%f"
-            ),
-            reverse=True,
-        )[0]
-        spectrum = sus.MsmsSpectrum(
-            usi,
-            float(annotations["Precursor_MZ"]),
-            int(annotations["Charge"]),
-            mz,
-            intensity,
+
+# Parse GNPS2 library (our minted GNPS2LIB accessions). Resolved only against our
+# GNPS2 library server — no legacy fallback for these ids.
+def _parse_gnps2_library(usi: str) -> Tuple[sus.MsmsSpectrum, str]:
+    match = _match_usi(usi)
+    index_flag = match.group(3)
+    if index_flag.lower() != "accession":
+        raise UsiError(
+            "Currently supported GNPS2 library index flags: accession", 400
         )
-        return spectrum, source_link
-    except requests.exceptions.HTTPError:
-        raise UsiError("Unknown GNPS library USI", 404)
+    index = match.group(4)
+    spectrum = _fetch_gnps_library_spectrum(
+        usi, [f"https://library.gnps2.org/gnpsspectrum?SpectrumID={index}"]
+    )
+    return spectrum, "https://library.gnps2.org"
 
 
 # Parse MassBank entry.

test/test_unit.py

@@ -138,6 +138,96 @@ def test_parse_gnps_library():
     assert spectrum.precursor_mz == pytest.approx(981.54)
 
 
+# --- GNPS2 library server routing (library.gnps2.org) --------------------------
+
+_GNPS_LIBRARY_DICT = {
+    "spectruminfo": {"peaks_json": "[[100.0, 50.0], [200.0, 80.0]]"},
+    "annotations": [
+        {
+            "create_time": "2020-01-01 00:00:00.0",
+            "Precursor_MZ": "123.45",
+            "Charge": "1",
+        }
+    ],
+}
+
+
+class _FakeResponse:
+    def __init__(self, payload, status_code=200):
+        self._payload = payload
+        self.status_code = status_code
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise requests.exceptions.HTTPError(response=self)
+
+    def json(self):
+        return self._payload
+
+
+def test_parse_gnps2_library_uses_our_endpoint_only():
+    calls = []
+
+    def fake_get(url, timeout=None):
+        calls.append(url)
+        return _FakeResponse(_GNPS_LIBRARY_DICT)
+
+    with unittest.mock.patch.object(parsing.requests, "get", side_effect=fake_get):
+        spectrum, source_link, _ = parsing.parse_usi(
+            "mzspec:GNPS2:GNPS2-LIBRARY:accession:GNPS2LIB00000000001"
+        )
+    assert calls and all("library.gnps2.org/gnpsspectrum" in u for u in calls), calls
+    assert source_link == "https://library.gnps2.org"
+    assert spectrum.precursor_mz == pytest.approx(123.45)
+
+
+def test_parse_gnps2_library_no_fallback():
+    def fake_get(url, timeout=None):
+        raise requests.exceptions.ConnectionError("down")
+
+    with unittest.mock.patch.object(parsing.requests, "get", side_effect=fake_get):
+        with pytest.raises(UsiError) as exc_info:
+            parsing.parse_usi(
+                "mzspec:GNPS2:GNPS2-LIBRARY:accession:GNPS2LIB00000000001"
+            )
+    assert exc_info.value.error_code == 404
+
+
+def test_parse_gnps_library_prefers_our_endpoint():
+    calls = []
+
+    def fake_get(url, timeout=None):
+        calls.append(url)
+        return _FakeResponse(_GNPS_LIBRARY_DICT)
+
+    with unittest.mock.patch.object(parsing.requests, "get", side_effect=fake_get):
+        spectrum, source_link, _ = parsing.parse_usi(
+            "mzspec:GNPS:GNPS-LIBRARY:accession:CCMSLIB00000001547"
+        )
+    assert "library.gnps2.org/gnpsspectrum" in calls[0]
+    assert not any("external.gnps2.org" in u for u in calls)  # ours worked
+    assert "gnpslibraryspectrum.jsp" in source_link
+    assert spectrum.precursor_mz == pytest.approx(123.45)
+
+
+def test_parse_gnps_library_falls_back_to_external():
+    calls = []
+
+    def fake_get(url, timeout=None):
+        calls.append(url)
+        if "library.gnps2.org" in url:
+            raise requests.exceptions.ConnectionError("ours down")
+        return _FakeResponse(_GNPS_LIBRARY_DICT)
+
+    with unittest.mock.patch.object(parsing.requests, "get", side_effect=fake_get):
+        spectrum, source_link, _ = parsing.parse_usi(
+            "mzspec:GNPS:GNPS-LIBRARY:accession:CCMSLIB00000001547"
+        )
+    assert any("library.gnps2.org" in u for u in calls)  # tried ours first
+    assert any("external.gnps2.org/gnpsspectrum" in u for u in calls)  # fell back
+    assert spectrum.precursor_mz == pytest.approx(123.45)
+
+
 def test_parse_massbank():
     usi = "mzspec:MASSBANK::accession:SM858102"
     spectrum, _, splash_key = parsing.parse_usi(usi)