Add PubMed literature mining API

README.md

@@ -1,5 +1,37 @@
 # OpenTCM
 
+## PubMed Literature Mining API
+
+OpenTCM includes a lightweight PubMed integration for retrieving literature and
+summarising the returned publication metadata. It uses the official NCBI
+E-utilities API and does not require a new Python dependency.
+
+Start the application, then request:
+
+```text
+GET /api/pubmed?query=acupuncture%20osteoarthritis&max_results=20
+```
+
+The response includes a list of records with PubMed links, DOI, authors,
+journal, publication date, and publication type. The `analytics` object
+aggregates the returned records by publication year, journal, and publication
+type for quick bibliometric exploration. `max_results` accepts values from 1
+to 100 and defaults to 20.
+
+For NCBI's recommended request identification and higher rate limits, set the
+following optional environment variables before starting the app:
+
+```text
+PUBMED_EMAIL=researcher@example.org
+NCBI_API_KEY=your_ncbi_api_key
+```
+
+Set `PUBMED_EMAIL` in deployed instances so NCBI can identify the application
+owner. Without an API key, NCBI limits E-utilities clients to three requests per
+second; an API key raises the default limit to ten requests per second. See the
+[NCBI E-utilities usage guidelines](https://www.ncbi.nlm.nih.gov/books/NBK25497/)
+before operating the endpoint at scale.
+
 OpenTCM is a web application designed for intelligent question answering in Traditional Chinese Medicine (TCM). Built upon a knowledge graph and a Large Language Model (LLM, using Kimi as an example), OpenTCM combines structured TCM knowledge with the understanding and generation capabilities of LLMs. This enables the system to provide users with well-sourced, comprehensive, and easy-to-understand TCM information. The project supports streaming responses to enhance user interaction.
 
 Our paper, *"OpenTCM: A GraphRAG-Empowered LLM-based System for Traditional Chinese Medicine Knowledge Retrieval and Diagnosis"*, was accepted by BIGCOM25 and received the Best Paper Award 🏆.

app.py

@@ -5,6 +5,7 @@
 from dotenv import load_dotenv
 import json
 import time
+from pubmed_mining import PubMedClient, PubMedQueryError, PubMedRequestError
 
 
 load_dotenv()
@@ -28,6 +29,7 @@
 
 
 tcm_app_instance = None
+pubmed_client = PubMedClient()
 
 if TCM_RAG_APP_LOADED:
     csv_path = os.getenv("TCM_CSV_PATH", r"tcm_KG.csv") 
@@ -159,6 +161,20 @@ def error_stream():
             'Connection': 'keep-alive'
         }
     )
+
+
+@app.route('/api/pubmed', methods=['GET'])
+def search_pubmed():
+    query = request.args.get('query', '')
+    max_results = request.args.get('max_results', default=20, type=int)
+
+    try:
+        return jsonify(pubmed_client.search(query, max_results=max_results))
+    except PubMedQueryError as error:
+        return jsonify({'error': str(error)}), 400
+    except PubMedRequestError as error:
+        logger.error('PubMed search failed: %s', error)
+        return jsonify({'error': 'PubMed is temporarily unavailable.'}), 502
 
 @app.route('/static/images/<filename>')
 def serve_image(filename):

pubmed_mining.py

@@ -0,0 +1,200 @@
+"""PubMed search and lightweight bibliometric analysis for OpenTCM."""
+
+from collections import Counter
+import os
+import re
+from typing import Any, Dict, List, Optional
+
+import requests
+
+
+EUTILS_BASE_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
+DEFAULT_MAX_RESULTS = 20
+MAX_RESULTS = 100
+MAX_QUERY_LENGTH = 500
+TOOL_NAME = "OpenTCM"
+
+
+class PubMedError(Exception):
+    """Base exception for PubMed mining failures."""
+
+
+class PubMedQueryError(PubMedError):
+    """Raised when a PubMed query is invalid."""
+
+
+class PubMedRequestError(PubMedError):
+    """Raised when PubMed cannot be reached or returns invalid data."""
+
+
+class PubMedClient:
+    """Fetch PubMed records and derive summary statistics from their metadata."""
+
+    def __init__(
+        self,
+        session: Optional[requests.Session] = None,
+        timeout: int = 20,
+        email: Optional[str] = None,
+        api_key: Optional[str] = None,
+    ) -> None:
+        self.session = session or requests.Session()
+        self.timeout = timeout
+        self.email = email or os.getenv("PUBMED_EMAIL")
+        self.api_key = api_key or os.getenv("NCBI_API_KEY")
+
+    def search(self, query: str, max_results: int = DEFAULT_MAX_RESULTS) -> Dict[str, Any]:
+        """Return PubMed records and publication-level aggregation for a query."""
+        normalized_query = self._validate_query(query)
+        validated_max_results = self._validate_max_results(max_results)
+        search_payload = self._get_json(
+            "esearch.fcgi",
+            {
+                "db": "pubmed",
+                "term": normalized_query,
+                "retmax": validated_max_results,
+                "retmode": "json",
+                "sort": "relevance",
+            },
+        )
+        search_result = search_payload.get("esearchresult", {})
+        identifiers = search_result.get("idlist", [])
+        total_results = self._as_int(search_result.get("count"))
+
+        if not identifiers:
+            return {
+                "query": normalized_query,
+                "total_results": total_results,
+                "records": [],
+                "analytics": self._build_analytics([]),
+            }
+
+        summary_payload = self._get_json(
+            "esummary.fcgi",
+            {
+                "db": "pubmed",
+                "id": ",".join(identifiers),
+                "retmode": "json",
+            },
+        )
+        summary_result = summary_payload.get("result", {})
+        records = [
+            self._to_record(summary_result[identifier])
+            for identifier in identifiers
+            if identifier in summary_result
+        ]
+
+        return {
+            "query": normalized_query,
+            "total_results": total_results,
+            "records": records,
+            "analytics": self._build_analytics(records),
+        }
+
+    def _get_json(self, endpoint: str, params: Dict[str, Any]) -> Dict[str, Any]:
+        request_params = dict(params)
+        request_params["tool"] = TOOL_NAME
+        if self.email:
+            request_params["email"] = self.email
+        if self.api_key:
+            request_params["api_key"] = self.api_key
+
+        try:
+            response = self.session.get(
+                f"{EUTILS_BASE_URL}/{endpoint}",
+                params=request_params,
+                timeout=self.timeout,
+                headers={"User-Agent": "OpenTCM-PubMed-Mining/1.0"},
+            )
+            response.raise_for_status()
+            payload = response.json()
+        except (requests.RequestException, ValueError) as error:
+            raise PubMedRequestError("Unable to retrieve PubMed data.") from error
+
+        if not isinstance(payload, dict):
+            raise PubMedRequestError("PubMed returned an invalid response.")
+        if payload.get("error"):
+            raise PubMedRequestError(f"PubMed returned an error: {payload['error']}")
+        return payload
+
+    @staticmethod
+    def _validate_query(query: str) -> str:
+        if not isinstance(query, str) or not query.strip():
+            raise PubMedQueryError("A non-empty query is required.")
+        normalized_query = query.strip()
+        if len(normalized_query) > MAX_QUERY_LENGTH:
+            raise PubMedQueryError(
+                f"query must not exceed {MAX_QUERY_LENGTH} characters."
+            )
+        return normalized_query
+
+    @staticmethod
+    def _validate_max_results(max_results: int) -> int:
+        if isinstance(max_results, bool) or not isinstance(max_results, int):
+            raise PubMedQueryError("max_results must be an integer.")
+        if not 1 <= max_results <= MAX_RESULTS:
+            raise PubMedQueryError(f"max_results must be between 1 and {MAX_RESULTS}.")
+        return max_results
+
+    @staticmethod
+    def _as_int(value: Any) -> int:
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return 0
+
+    @staticmethod
+    def _to_record(summary: Dict[str, Any]) -> Dict[str, Any]:
+        authors = [
+            author.get("name", "")
+            for author in summary.get("authors", [])
+            if author.get("name")
+        ]
+        article_ids = summary.get("articleids", [])
+        doi = next(
+            (
+                article_id.get("value")
+                for article_id in article_ids
+                if article_id.get("idtype") == "doi"
+            ),
+            None,
+        )
+        published_at = summary.get("pubdate") or summary.get("epubdate") or ""
+        pmid = str(summary.get("uid", ""))
+        return {
+            "pmid": pmid,
+            "title": summary.get("title", ""),
+            "journal": summary.get("fulljournalname") or summary.get("source", ""),
+            "published_at": published_at,
+            "authors": authors,
+            "publication_types": summary.get("pubtype", []),
+            "doi": doi,
+            "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/" if pmid else None,
+        }
+
+    @staticmethod
+    def _build_analytics(records: List[Dict[str, Any]]) -> Dict[str, Any]:
+        year_counts: Counter[str] = Counter()
+        journal_counts: Counter[str] = Counter()
+        publication_type_counts: Counter[str] = Counter()
+
+        for record in records:
+            year_match = re.search(r"\b(?:19|20)\d{2}\b", record["published_at"])
+            if year_match:
+                year_counts[year_match.group(0)] += 1
+            if record["journal"]:
+                journal_counts[record["journal"]] += 1
+            publication_type_counts.update(record["publication_types"])
+
+        return {
+            "returned_results": len(records),
+            "publication_years": dict(sorted(year_counts.items(), reverse=True)),
+            "top_journals": PubMedClient._top_counts(journal_counts),
+            "publication_types": PubMedClient._top_counts(publication_type_counts),
+        }
+
+    @staticmethod
+    def _top_counts(counter: Counter[str]) -> List[Dict[str, Any]]:
+        return [
+            {"name": name, "count": count}
+            for name, count in counter.most_common()
+        ]

tests/test_app_pubmed.py

@@ -0,0 +1,47 @@
+import unittest
+from unittest.mock import patch
+
+from app import app
+from pubmed_mining import PubMedRequestError
+
+
+class SuccessfulPubMedClient:
+    def search(self, query, max_results):
+        return {
+            "query": query,
+            "total_results": 1,
+            "records": [],
+            "analytics": {"returned_results": 0},
+        }
+
+
+class UnavailablePubMedClient:
+    def search(self, query, max_results):
+        raise PubMedRequestError("service unavailable")
+
+
+class PubMedRouteTests(unittest.TestCase):
+    def setUp(self):
+        self.client = app.test_client()
+
+    def test_route_returns_search_results(self):
+        with patch("app.pubmed_client", SuccessfulPubMedClient()):
+            response = self.client.get("/api/pubmed?query=ginseng&max_results=5")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.get_json()["query"], "ginseng")
+
+    def test_route_rejects_invalid_parameters(self):
+        response = self.client.get("/api/pubmed?query=&max_results=invalid")
+
+        self.assertEqual(response.status_code, 400)
+
+    def test_route_maps_pubmed_failures_to_bad_gateway(self):
+        with patch("app.pubmed_client", UnavailablePubMedClient()):
+            response = self.client.get("/api/pubmed?query=ginseng")
+
+        self.assertEqual(response.status_code, 502)
+        self.assertEqual(
+            response.get_json(),
+            {"error": "PubMed is temporarily unavailable."},
+        )