From b4ec6be46902c1368cd7df2bb50584d03a12d39e Mon Sep 17 00:00:00 2001 From: dainiubi438-debug Date: Sun, 9 Aug 2026 19:36:16 +0800 Subject: [PATCH] Add PubMed literature mining API --- README.md | 32 ++++++ app.py | 16 +++ pubmed_mining.py | 200 ++++++++++++++++++++++++++++++++++++ tests/test_app_pubmed.py | 47 +++++++++ tests/test_pubmed_mining.py | 94 +++++++++++++++++ 5 files changed, 389 insertions(+) create mode 100644 pubmed_mining.py create mode 100644 tests/test_app_pubmed.py create mode 100644 tests/test_pubmed_mining.py diff --git a/README.md b/README.md index 16853ae..e534475 100644 --- a/README.md +++ b/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 🏆. diff --git a/app.py b/app.py index 9e99717..f226f50 100644 --- a/app.py +++ b/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/') def serve_image(filename): diff --git a/pubmed_mining.py b/pubmed_mining.py new file mode 100644 index 0000000..53b4290 --- /dev/null +++ b/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() + ] diff --git a/tests/test_app_pubmed.py b/tests/test_app_pubmed.py new file mode 100644 index 0000000..f346d1b --- /dev/null +++ b/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."}, + ) diff --git a/tests/test_pubmed_mining.py b/tests/test_pubmed_mining.py new file mode 100644 index 0000000..350428e --- /dev/null +++ b/tests/test_pubmed_mining.py @@ -0,0 +1,94 @@ +import unittest +from unittest.mock import Mock + +import requests + +from pubmed_mining import PubMedClient, PubMedQueryError, PubMedRequestError + + +class FakeResponse: + def __init__(self, payload): + self.payload = payload + + def raise_for_status(self): + return None + + def json(self): + return self.payload + + +class PubMedClientTests(unittest.TestCase): + def test_search_returns_records_and_bibliometric_summary(self): + session = Mock() + session.get.side_effect = [ + FakeResponse({"esearchresult": {"count": "42", "idlist": ["1", "2"]}}), + FakeResponse( + { + "result": { + "uids": ["1", "2"], + "1": { + "uid": "1", + "title": "Herbal medicine trial", + "fulljournalname": "Journal of TCM", + "pubdate": "2024 Jan 10", + "authors": [{"name": "Li Wei"}], + "pubtype": ["Clinical Trial"], + "articleids": [{"idtype": "doi", "value": "10.1/example"}], + }, + "2": { + "uid": "2", + "title": "Acupuncture review", + "fulljournalname": "Journal of TCM", + "pubdate": "2023 Dec", + "authors": [{"name": "Zhang Min"}], + "pubtype": ["Review"], + "articleids": [], + }, + } + } + ), + ] + client = PubMedClient( + session=session, + email="maintainer@example.org", + api_key="test-api-key", + ) + + result = client.search("acupuncture", max_results=2) + + self.assertEqual(result["query"], "acupuncture") + self.assertEqual(result["total_results"], 42) + self.assertEqual(result["records"][0]["doi"], "10.1/example") + self.assertEqual(result["analytics"]["publication_years"], {"2024": 1, "2023": 1}) + self.assertEqual(result["analytics"]["top_journals"], [{"name": "Journal of TCM", "count": 2}]) + self.assertEqual(session.get.call_count, 2) + search_params = session.get.call_args_list[0].kwargs["params"] + self.assertEqual(search_params["tool"], "OpenTCM") + self.assertEqual(search_params["email"], "maintainer@example.org") + self.assertEqual(search_params["api_key"], "test-api-key") + + def test_search_rejects_an_empty_query(self): + with self.assertRaises(PubMedQueryError): + PubMedClient(session=Mock()).search(" ") + + def test_search_wraps_network_errors(self): + session = Mock() + session.get.side_effect = requests.Timeout() + + with self.assertRaises(PubMedRequestError): + PubMedClient(session=session).search("ginseng") + + def test_search_rejects_ncbi_error_payloads(self): + session = Mock() + session.get.return_value = FakeResponse({"error": "API rate limit exceeded"}) + + with self.assertRaises(PubMedRequestError): + PubMedClient(session=session).search("ginseng") + + def test_search_rejects_excessively_long_queries(self): + with self.assertRaises(PubMedQueryError): + PubMedClient(session=Mock()).search("x" * 501) + + +if __name__ == "__main__": + unittest.main()