Update genesis/api_clients/pubmed_api.py

genesis/api_clients/pubmed_api.py

@@ -1,92 +1,88 @@
 # genesis/api_clients/pubmed_api.py
+
 import os
 import requests
-import xml.etree.ElementTree as ET
-from typing import List, Dict
+import html
+from xml.etree import ElementTree as ET
 
-NCBI_BASE = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
-PUBMED_API_KEY = os.getenv("PUBMED_API_KEY")  # Optional, set in Hugging Face / .env
+PUBMED_BASE = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
 
-# -------------------------
-# Core Helpers
-# -------------------------
-def pubmed_search(query: str, max_results: int = 10) -> List[str]:
+def search_pubmed_literature(query: str, max_results: int = 10):
     """
-    Search PubMed for a query and return a list of PubMed IDs.
+    Search PubMed literature using NCBI E-utilities API.
+
+    Args:
+        query (str): Search query (e.g. "synthetic biology cancer therapy")
+        max_results (int): Maximum number of results to return.
+
+    Returns:
+        list[dict]: Each dict has 'title', 'authors', 'link'
     """
+    if not query.strip():
+        return []
+
+    api_key = os.getenv("PUBMED_API_KEY")  # optional
     params = {
         "db": "pubmed",
         "term": query,
         "retmax": max_results,
-        "api_key": PUBMED_API_KEY
+        "retmode": "xml"
     }
-    r = requests.get(f"{NCBI_BASE}/esearch.fcgi", params=params)
-    r.raise_for_status()
-    root = ET.fromstring(r.text)
-    return [id_elem.text for id_elem in root.findall(".//Id")]
+    if api_key:
+        params["api_key"] = api_key
 
-def fetch_details(pubmed_ids: List[str]) -> List[Dict]:
-    """
-    Fetch detailed metadata for PubMed articles.
-    """
-    if not pubmed_ids:
-        return []
+    try:
+        # Step 1: Search for IDs
+        search_url = f"{PUBMED_BASE}/esearch.fcgi"
+        search_res = requests.get(search_url, params=params, timeout=10)
+        search_res.raise_for_status()
+        root = ET.fromstring(search_res.text)
+        ids = [id_elem.text for id_elem in root.findall(".//Id")]
 
-    params = {
-        "db": "pubmed",
-        "id": ",".join(pubmed_ids),
-        "retmode": "xml",
-        "api_key": PUBMED_API_KEY
-    }
-    r = requests.get(f"{NCBI_BASE}/efetch.fcgi", params=params)
-    r.raise_for_status()
-
-    articles = []
-    root = ET.fromstring(r.text)
-    for article in root.findall(".//PubmedArticle"):
-        title_elem = article.find(".//ArticleTitle")
-        abstract_elem = article.find(".//Abstract/AbstractText")
-        pmid_elem = article.find(".//PMID")
-        authors = [
-            f"{a.find('LastName').text} {a.find('ForeName').text}"
-            for a in article.findall(".//Author")
-            if a.find("LastName") is not None and a.find("ForeName") is not None
-        ]
-
-        articles.append({
-            "pmid": pmid_elem.text if pmid_elem is not None else "",
-            "title": title_elem.text if title_elem is not None else "",
-            "abstract": abstract_elem.text if abstract_elem is not None else "",
-            "authors": authors,
-            "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid_elem.text}/" if pmid_elem is not None else ""
-        })
-    return articles
-
-# -------------------------
-# High-Level Search + Fetch
-# -------------------------
-def search_pubmed(query: str, max_results: int = 10) -> List[Dict]:
-    """
-    Search and return structured PubMed results.
-    """
-    ids = pubmed_search(query, max_results)
-    return fetch_details(ids)
+        if not ids:
+            return []
 
-# -------------------------
-# Cross-Domain Integration
-# -------------------------
-def literature_to_entities(query: str) -> Dict:
-    """
-    Link PubMed literature to related drugs (ChEMBL), genes (NCBI), and ontology terms (BioPortal).
-    """
-    from genesis.api_clients import chembl_api, ncbi_api, bioportal_api  # Lazy import to avoid cycles
-
-    literature_data = {
-        "query": query,
-        "articles": search_pubmed(query),
-        "related_drugs": chembl_api.search_molecule(query),
-        "related_genes": ncbi_api.search_gene(query),
-        "ontology_mappings": bioportal_api.search_and_map(query)
-    }
+        # Step 2: Fetch article details
+        fetch_url = f"{PUBMED_BASE}/efetch.fcgi"
+        fetch_params = {
+            "db": "pubmed",
+            "id": ",".join(ids),
+            "retmode": "xml"
+        }
+        if api_key:
+            fetch_params["api_key"] = api_key
+
+        fetch_res = requests.get(fetch_url, params=fetch_params, timeout=10)
+        fetch_res.raise_for_status()
+        fetch_root = ET.fromstring(fetch_res.text)
+
+        results = []
+        for article in fetch_root.findall(".//PubmedArticle"):
+            title_elem = article.find(".//ArticleTitle")
+            title = html.unescape(title_elem.text) if title_elem is not None else "No title"
 
-    return literature_data
+            authors = []
+            for author in article.findall(".//Author"):
+                last = author.find("LastName")
+                fore = author.find("ForeName")
+                if last is not None and fore is not None:
+                    authors.append(f"{fore.text} {last.text}")
+
+            pmid_elem = article.find(".//PMID")
+            pmid = pmid_elem.text if pmid_elem is not None else ""
+            link = f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid else ""
+
+            results.append({
+                "title": title,
+                "authors": ", ".join(authors) if authors else "N/A",
+                "link": link
+            })
+
+        return results
+
+    except requests.exceptions.RequestException as e:
+        print(f"[PubMed API Error] {e}")
+        return []
+    except ET.ParseError as e:
+        print(f"[PubMed Parse Error] {e}")
+        return []