Source code for tooluniverse.metabolite_tool

"""
Metabolite tools for ToolUniverse.

Replaces the broken HMDB direct API (blocked by Cloudflare).
Uses PubChem as the primary compound data source and CTD for
disease associations.

Broken HMDB API archived at: src/tooluniverse/data/broken_apis/hmdb_rest.json
"""

import re
import requests
from typing import Any, Dict, List, Optional

from .base_tool import BaseTool
from .tool_registry import register_tool

PUBCHEM_API = "https://pubchem.ncbi.nlm.nih.gov/rest/pug"
CTD_API = "https://ctdbase.org/tools/batchQuery.go"

# Regex to strip common stereochemistry prefixes so CTD can match the parent compound.
# Example: "Beta-D-Glucose" → "Glucose", "L-Alanine" → "Alanine", "L-Lactic Acid" → "Lactic Acid"
#
# Each stereo descriptor must be followed by its own separator ([-\s]+). This
# prevents over-stripping when the compound name itself starts with a stereo
# letter: "L-Lactic Acid" must strip only "L-" (→ "Lactic Acid"), NOT "L-L"
# (→ "actic Acid"). The trailing optional group only matches a *second*
# descriptor that is itself separator-terminated (e.g. the "D-" in "Beta-D-").
_STEREO_DESC = r"(?:alpha|beta|D|L|R|S|cis|trans|endo|exo)"
_STEREO_PREFIX = re.compile(
    rf"^{_STEREO_DESC}[-\s]+(?:{_STEREO_DESC}[-\s]+)?",
    re.IGNORECASE,
)

# CAS Registry Number pattern (e.g. 50-99-7)
_CAS_RE = re.compile(r"^\d{2,7}-\d{2}-\d$")


def _strip_stereo(name: str) -> str:
    """Remove leading stereochemistry descriptors from a compound name."""
    stripped = _STEREO_PREFIX.sub("", name).strip()
    return stripped if stripped and stripped != name else ""


def _cas_from_synonyms(synonyms: List[str]) -> Optional[str]:
    """Return the first CAS-style number (e.g. 50-99-7) from a synonyms list."""
    for s in synonyms:
        if _CAS_RE.match(s.strip()):
            return s.strip()
    return None




[docs]
class CTDBackendUnavailable(RuntimeError):
    """The CTD mirror could not be reached or is no longer served.

    Raised so that "the backend is gone" is never reported as "this compound has
    no disease associations" -- the two are indistinguishable to a caller once
    an empty list is returned as a success.
    """






[docs]
@register_tool("MetaboliteTool")
class MetaboliteTool(BaseTool):
    """
    Tool for querying metabolite data via PubChem (compound info) and
    CTD (disease associations).  Accepts HMDB IDs, compound names, and
    PubChem CIDs.
    """



[docs]
    def __init__(self, tool_config: Dict[str, Any]):
        super().__init__(tool_config)
        self.timeout: int = tool_config.get("timeout", 30)





[docs]
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        operation = arguments.get("operation", "")
        if not operation:
            operation = self.get_schema_const_operation()

        if operation == "get_info":
            return self._get_info(arguments)
        elif operation == "search":
            return self._search(arguments)
        elif operation == "get_diseases":
            return self._get_diseases(arguments)
        return {
            "status": "error",
            "error": f"Unknown operation: {operation}. Supported: get_info, search, get_diseases",
        }



    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------



[docs]
    def _extract_identifier(self, arguments: Dict[str, Any]) -> str:
        """Extract identifier from hmdb_id, compound_name, or pubchem_cid arguments."""
        identifier = (
            arguments.get("hmdb_id")
            or arguments.get("compound_name")
            or arguments.get("pubchem_cid")
            or ""
        )
        return str(identifier) if identifier else ""





[docs]
    def _resolve_to_cid(self, identifier: str) -> Optional[int]:
        """
        Resolve an HMDB ID, compound name, or PubChem CID string to a CID integer.
        Returns None if resolution fails.
        """
        if identifier.lstrip("-").isdigit():
            return int(identifier)

        # HMDB ID → PubChem via RegistryID cross-reference
        if identifier.upper().startswith("HMDB"):
            hmdb_id = identifier.upper()
            if not re.match(r"^HMDB\d+$", hmdb_id):
                hmdb_id = f"HMDB{hmdb_id[4:].zfill(7)}"
            resp = requests.get(
                f"{PUBCHEM_API}/compound/xref/RegistryID/{hmdb_id}/JSON",
                timeout=self.timeout,
            )
            if resp.status_code == 200:
                compounds = resp.json().get("PC_Compounds", [])
                if compounds:
                    return compounds[0].get("id", {}).get("id", {}).get("cid")
            return None

        # Compound name → PubChem CID (exact match first, then autocomplete fallback)
        resp = requests.get(
            f"{PUBCHEM_API}/compound/name/{requests.utils.quote(identifier)}/cids/JSON",
            timeout=self.timeout,
        )
        if resp.status_code == 200:
            cids = resp.json().get("IdentifierList", {}).get("CID", [])
            return cids[0] if cids else None
        # Autocomplete fallback for lipid classes and inexact names
        ac_resp = requests.get(
            f"https://pubchem.ncbi.nlm.nih.gov/rest/autocomplete/compound/{requests.utils.quote(identifier)}/json?limit=5",
            timeout=self.timeout,
        )
        if ac_resp.status_code == 200:
            suggestions = ac_resp.json().get("dictionary_terms", {}).get("compound", [])
            for suggestion in suggestions[:3]:
                cid_resp = requests.get(
                    f"{PUBCHEM_API}/compound/name/{requests.utils.quote(suggestion)}/cids/JSON",
                    timeout=self.timeout,
                )
                if cid_resp.status_code == 200:
                    cids = cid_resp.json().get("IdentifierList", {}).get("CID", [])
                    if cids:
                        return cids[0]
        return None





[docs]
    def _get_properties(self, cid: int) -> Dict[str, Any]:
        """Fetch Title, IUPAC name, formula, weight, SMILES, InChIKey from PubChem."""
        resp = requests.get(
            f"{PUBCHEM_API}/compound/cid/{cid}/property/"
            "Title,MolecularFormula,MolecularWeight,IsomericSMILES,InChIKey,IUPACName/JSON",
            timeout=self.timeout,
        )
        if resp.status_code == 200:
            props = resp.json().get("PropertyTable", {}).get("Properties", [])
            return props[0] if props else {}
        return {}





[docs]
    def _get_synonyms(self, cid: int) -> List[str]:
        """Fetch all synonyms for a PubChem CID."""
        resp = requests.get(
            f"{PUBCHEM_API}/compound/cid/{cid}/synonyms/JSON",
            timeout=self.timeout,
        )
        if resp.status_code == 200:
            info = resp.json().get("InformationList", {}).get("Information", [])
            return info[0].get("Synonym", []) if info else []
        return []





[docs]
    def _ctd_diseases(self, chemical_term: str) -> List[Dict[str, Any]]:
        """Query mydisease.info's cached CTD chemical-disease curation.

        CTD's native batchQuery.go is altcha-CAPTCHA-blocked since early
        2026, and the RENCI Automat mirror this method used as a fallback
        has since been fully decommissioned (its registry no longer lists a
        'ctd' backend at all -- see ctd_tool.py for the full history). This
        queries the BioThings mydisease.info API instead, which still serves
        CTD's chemical-disease edges under `ctd.chemical_related_to_disease`
        on each disease document -- see ctd_tool.py's `_chemical_to_diseases`
        for the same approach applied to the standalone CTD_* tools.
        """
        safe = chemical_term.replace('"', "").replace("\\", "")
        # `_resolve_ctd_term` tries a CAS Registry Number candidate (from
        # PubChem synonyms) alongside plain names -- match it against the
        # right nested field rather than always searching by name.
        match_field = "cas_registry_number" if _CAS_RE.match(chemical_term) else "chemical_name"
        try:
            r = requests.get(
                "https://mydisease.info/v1/query",
                params={
                    "q": f'ctd.chemical_related_to_disease.{match_field}:"{safe}"',
                    "fields": "mondo.label,ctd.chemical_related_to_disease",
                    "size": 200,
                },
                headers={"Accept": "application/json"},
                timeout=self.timeout,
            )
            r.raise_for_status()
        except requests.RequestException as exc:
            raise CTDBackendUnavailable(
                f"mydisease.info is unavailable: {exc}"
            ) from exc
        try:
            hits = r.json().get("hits") or []
        except ValueError:
            return []

        rows = []
        for hit in hits:
            disease_id = hit.get("_id")
            mondo = hit.get("mondo")
            disease_name = mondo.get("label") if isinstance(mondo, dict) else None
            # BioThings occasionally merges more than one source record onto
            # the same disease document, in which case `ctd` itself is a
            # list of sub-documents instead of a single dict -- see
            # ctd_tool.py's `_ctd_disease_entries` for the same handling.
            ctd_field = hit.get("ctd") or []
            ctd_docs = ctd_field if isinstance(ctd_field, list) else [ctd_field]
            entries = []
            for ctd_doc in ctd_docs:
                if not isinstance(ctd_doc, dict):
                    continue
                related = ctd_doc.get("chemical_related_to_disease") or []
                entries.extend(related if isinstance(related, list) else [related])
            for entry in entries:
                if not isinstance(entry, dict):
                    continue
                if str(entry.get(match_field, "")).strip().lower() != chemical_term.strip().lower():
                    continue
                pubmed = entry.get("pubmed")
                pubmed_ids = pubmed if isinstance(pubmed, list) else ([pubmed] if pubmed else [])
                rows.append(
                    {
                        "DiseaseName": disease_name,
                        "DiseaseID": disease_id,
                        "DirectEvidence": entry.get("direct_evidence"),
                        "PubMedIDs": [str(p) for p in pubmed_ids],
                    }
                )
        return rows





[docs]
    def _resolve_ctd_term(
        self, title: str, synonyms: List[str]
    ) -> tuple[str, List[Dict[str, Any]]]:
        """
        Try multiple name variants to find a CTD match.
        Returns (term_used, disease_rows).
        """
        candidates = [title]
        stripped = _strip_stereo(title)
        if stripped:
            candidates.append(stripped)
        cas = _cas_from_synonyms(synonyms)
        if cas:
            candidates.append(cas)
        # Also try common synonyms (e.g. "glucosylceramide" when title is "GlcCer(d18:1/...)")
        for syn in synonyms[:10]:
            if syn and syn not in candidates and len(syn) < 50 and not syn[0].isdigit():
                candidates.append(syn)

        for term in candidates:
            rows = self._ctd_diseases(term)
            if rows:
                return term, rows
        return candidates[0], []



    # ------------------------------------------------------------------
    # Operations
    # ------------------------------------------------------------------



[docs]
    def _get_info(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Get compound info for a metabolite by HMDB ID, compound name, or PubChem CID.
        Returns common name, IUPAC name, formula, weight, SMILES, InChIKey.
        """
        identifier = self._extract_identifier(arguments)
        if not identifier:
            return {
                "status": "error",
                "error": "Provide hmdb_id, compound_name, or pubchem_cid.",
            }
        try:
            cid = self._resolve_to_cid(identifier)
            if cid is None:
                return {
                    "status": "error",
                    "error": f"Could not resolve '{identifier}' to a PubChem compound.",
                }
            props = self._get_properties(cid)
            return {
                "status": "success",
                "data": {
                    "pubchem_cid": cid,
                    "name": props.get("Title"),
                    "iupac_name": props.get("IUPACName"),
                    "formula": props.get("MolecularFormula"),
                    "molecular_weight": props.get("MolecularWeight"),
                    "smiles": props.get("SMILES") or props.get("IsomericSMILES"),
                    "inchikey": props.get("InChIKey"),
                },
                "metadata": {
                    "source": "PubChem",
                    "pubchem_url": f"https://pubchem.ncbi.nlm.nih.gov/compound/{cid}",
                },
            }
        except requests.exceptions.RequestException as e:
            return {"status": "error", "error": f"Request failed: {str(e)}"}








[docs]
    def _get_diseases(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Get curated disease associations for a metabolite.

        Accepts HMDB ID, compound name, or PubChem CID. Resolves to a
        PubChem compound, then queries CTD with multiple name variants
        (title, stereo-stripped, CAS number) to maximise CTD match rate.
        """
        identifier = self._extract_identifier(arguments)
        if not identifier:
            return {
                "status": "error",
                "error": "Provide hmdb_id, compound_name, or pubchem_cid.",
            }
        limit = int(arguments.get("limit", 50))

        try:
            cid = self._resolve_to_cid(identifier)
            if cid is None:
                return {
                    "status": "error",
                    "error": f"Could not resolve '{identifier}' to a PubChem compound.",
                }

            props = self._get_properties(cid)
            title = props.get("Title") or props.get("IUPACName", identifier)
            synonyms = self._get_synonyms(cid)

            # Prepend the original user input as the first candidate for CTD resolution
            # (e.g. "glucocerebroside" is recognized by CTD even if PubChem title is "GlcCer(d18:1/...)")
            if (
                identifier
                and not identifier.upper().startswith("HMDB")
                and not identifier.isdigit()
                and identifier != title
            ):
                synonyms = [identifier] + list(synonyms)
            term_used, rows = self._resolve_ctd_term(title, synonyms)
            diseases = [
                {
                    "disease_name": r.get("DiseaseName"),
                    "disease_id": r.get("DiseaseID"),
                    "disease_categories": r.get("DiseaseCategories"),
                    "direct_evidence": r.get("DirectEvidence"),
                    "pubmed_ids": (
                        r["PubMedIDs"].split("|")
                        if isinstance(r.get("PubMedIDs"), str)
                        else (r.get("PubMedIDs") or [])
                    ),
                }
                for r in rows
            ][:limit]

            return {
                "status": "success",
                "data": {
                    "identifier": identifier,
                    "compound_name": title,
                    "pubchem_cid": cid,
                    "ctd_query_term": term_used,
                    "disease_count": len(diseases),
                    "diseases": diseases,
                },
                "metadata": {
                    "source": "CTD (Comparative Toxicogenomics Database)",
                    "pubchem_url": f"https://pubchem.ncbi.nlm.nih.gov/compound/{cid}",
                },
            }
        except CTDBackendUnavailable as e:
            return {"status": "error", "error": str(e)}
        except requests.exceptions.RequestException as e:
            return {"status": "error", "error": f"Request failed: {str(e)}"}