"""
Search and get metadata for articles in Pubmed.
"""
import logging
import requests
from time import sleep
from typing import List
from functools import lru_cache
import xml.etree.ElementTree as ET
from indra.util import UnicodeXMLTreeBuilder as UTB
from indra.util import pretty_save_xml
logger = logging.getLogger(__name__)
pubmed_search = 'https://eutils.ncbi.nlm.nih.gov/entrez/eutils/esearch.fcgi'
pubmed_fetch = 'https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi'
# Send request can't be cached by lru_cache because it takes a dict
# (a mutable/unhashable type) as an argument. We cache the callers instead.
def send_request(url, data):
try:
res = requests.get(url, params=data)
except requests.exceptions.Timeout as e:
logger.error('PubMed request timed out')
logger.error('url: %s, data: %s' % (url, data))
logger.error(e)
return None
except requests.exceptions.RequestException as e:
logger.error('PubMed request exception')
logger.error('url: %s, data: %s' % (url, data))
logger.error(e)
return None
if res.status_code == 429:
sleep(0.5)
res = requests.get(url, params=data)
if not res.status_code == 200:
logger.error('Got return code %d from pubmed client.'
% res.status_code)
return None
tree = ET.XML(res.content, parser=UTB())
return tree
[docs]@lru_cache(maxsize=100)
def get_ids(search_term, **kwargs):
"""Search Pubmed for paper IDs given a search term.
Search options can be passed as keyword arguments, some of which are
custom keywords identified by this function, while others are passed on
as parameters for the request to the PubMed web service
For details on parameters that can be used in PubMed searches, see
https://www.ncbi.nlm.nih.gov/books/NBK25499/#chapter4.ESearch Some useful
parameters to pass are db='pmc' to search PMC instead of pubmed reldate=2
to search for papers within the last 2 days mindate='2016/03/01',
maxdate='2016/03/31' to search for papers in March 2016.
PubMed, by default, limits returned PMIDs to a small number, and this
number can be controlled by the "retmax" parameter. This function
uses a retmax value of 10,000 by default (the maximum supported by PubMed)
that can be changed via the corresponding keyword argument. Note also
the retstart argument along with retmax to page across batches of IDs.
Parameters
----------
search_term : str
A term for which the PubMed search should be performed.
use_text_word : Optional[bool]
If True, the "[tw]" string is appended to the search term to constrain
the search to "text words", that is words that appear as whole
in relevant parts of the PubMed entry (excl. for instance the journal
name or publication date) like the title and abstract. Using this
option can eliminate spurious search results such as all articles
published in June for a search for the "JUN" gene, or journal names
that contain Acad for a search for the "ACAD" gene.
See also: https://www.nlm.nih.gov/bsd/disted/pubmedtutorial/020_760.html
Default : True
kwargs : kwargs
Additional keyword arguments to pass to the PubMed search as
parameters.
"""
use_text_word = kwargs.pop('use_text_word', True)
if use_text_word:
search_term += '[tw]'
params = {'term': search_term,
'retmax': 10000,
'retstart': 0,
'db': 'pubmed',
'sort': 'pub+date'}
params.update(kwargs)
tree = send_request(pubmed_search, params)
if tree is None:
return []
if tree.find('ERROR') is not None:
logger.error(tree.find('ERROR').text)
return []
if tree.find('ErrorList') is not None:
for err in tree.find('ErrorList'):
logger.error('Error - %s: %s' % (err.tag, err.text))
return []
count = int(tree.find('Count').text)
id_terms = tree.findall('IdList/Id')
if id_terms is None:
return []
ids = [idt.text for idt in id_terms]
if count != len(ids):
logger.warning('Not all ids were retrieved for search %s;\n'
'limited at %d.' % (search_term, params['retmax']))
return ids
[docs]def get_id_count(search_term):
"""Get the number of citations in Pubmed for a search query.
Parameters
----------
search_term : str
A term for which the PubMed search should be performed.
Returns
-------
int or None
The number of citations for the query, or None if the query fails.
"""
params = {'term': search_term,
'rettype': 'count',
'db': 'pubmed'}
tree = send_request(pubmed_search, params)
if tree is None:
return None
else:
count = list(tree)[0].text
return int(count)
[docs]@lru_cache(maxsize=100)
def get_ids_for_gene(hgnc_name, **kwargs):
"""Get the curated set of articles for a gene in the Entrez database.
Search parameters for the Gene database query can be passed in as
keyword arguments.
Parameters
----------
hgnc_name : str
The HGNC name of the gene. This is used to obtain the HGNC ID
(using the hgnc_client module) and in turn used to obtain the Entrez
ID associated with the gene. Entrez is then queried for that ID.
"""
from indra.databases import hgnc_client
# Get the HGNC ID for the HGNC name
hgnc_id = hgnc_client.get_hgnc_id(hgnc_name)
if hgnc_id is None:
raise ValueError('Invalid HGNC name.')
# Get the Entrez ID
entrez_id = hgnc_client.get_entrez_id(hgnc_id)
if entrez_id is None:
raise ValueError('Entrez ID not found in HGNC table.')
# Query the Entrez Gene database
params = {'db': 'gene',
'retmode': 'xml',
'id': entrez_id}
params.update(kwargs)
tree = send_request(pubmed_fetch, params)
if tree is None:
return []
if tree.find('ERROR') is not None:
logger.error(tree.find('ERROR').text)
return []
# Get all PMIDs from the XML tree
id_terms = tree.findall('.//PubMedId')
if id_terms is None:
return []
# Use a set to remove duplicate IDs
ids = list(set([idt.text for idt in id_terms]))
return ids
[docs]def get_ids_for_mesh(mesh_id, major_topic=False, **kwargs):
"""Return PMIDs that are annotated with a given MeSH ID.
Parameters
----------
mesh_id : str
The MeSH ID of a term to search for, e.g., D009101.
major_topic : bool
If True, only papers for which the given MeSH ID is annotated as
a major topic are returned. Otherwise all annotations are considered.
Default: False
**kwargs
Any further PudMed search arguments that are passed to
get_ids.
"""
from indra.databases import mesh_client
mesh_name = mesh_client.get_mesh_name(mesh_id)
if not mesh_name:
logger.error('Could not get MeSH name for ID %s' % mesh_id)
return []
suffix = 'majr' if major_topic else 'mh'
search_term = '%s [%s]' % (mesh_name, suffix)
ids = get_ids(search_term, use_text_word=False, **kwargs)
if mesh_id.startswith('C') and not major_topic:
# Get pmids for supplementary concepts as well
search_term = '%s [nm]' % mesh_name
ids2 = get_ids(search_term, use_text_word=False, **kwargs)
ids = list(set(ids) | set(ids2))
return ids
[docs]def get_article_xml(pubmed_id):
"""Get the Article subtree a single article from the Pubmed database.
Parameters
----------
pubmed_id : str
A PubMed ID.
Returns
-------
xml.etree.ElementTree.Element
The XML ElementTree Element that represents the Article portion of the
PubMed entry.
"""
full_xml_tree = get_full_xml(pubmed_id)
if full_xml_tree is None:
return None
article = full_xml_tree.find('PubmedArticle/MedlineCitation/Article')
return article # May be none
[docs]@lru_cache(maxsize=100)
def get_full_xml(pubmed_id, fname=None):
"""Get the full XML tree of a single article from the Pubmed database.
Parameters
----------
pubmed_id : str
A PubMed ID.
fname : Optional[str]
If given, the XML is saved to the given file name.
Returns
-------
xml.etree.ElementTree.Element
The root element of the XML tree representing the PubMed entry.
The root is a PubmedArticleSet with a single PubmedArticle element
that contains the article metadata.
"""
if pubmed_id.upper().startswith('PMID'):
pubmed_id = pubmed_id[4:]
params = {'db': 'pubmed',
'retmode': 'xml',
'id': pubmed_id}
tree = send_request(pubmed_fetch, params)
if fname:
pretty_save_xml(tree, fname)
return tree
[docs]def get_title(pubmed_id):
"""Get the title of an article in the Pubmed database."""
article = get_article_xml(pubmed_id)
if article is None:
return None
return _get_title_from_article_element(article)
def _get_title_from_article_element(article):
title_tag = article.find('ArticleTitle')
title = None
if title_tag is not None:
title = title_tag.text
if hasattr(title_tag, 'itertext'):
title = ''.join(list(title_tag.itertext()))
return title
def _abstract_from_article_element(article, prepend_title=False):
abstract = article.findall('Abstract/AbstractText')
if abstract is None:
return None
abstract_text = ' '.join(['' if not hasattr(abst, 'itertext')
else ' '.join(list(abst.itertext()))
for abst in abstract])
if prepend_title:
title = _get_title_from_article_element(article)
if title is not None:
if not title.endswith('.'):
title += '.'
abstract_text = title + ' ' + abstract_text
return abstract_text
[docs]def get_abstract(pubmed_id, prepend_title=True):
"""Get the abstract of an article in the Pubmed database."""
article = get_article_xml(pubmed_id)
if article is None:
return None
return _abstract_from_article_element(article, prepend_title)
# A function to get the text for the element, or None if not found
def _find_elem_text(root, xpath_string):
elem = root.find(xpath_string)
return None if elem is None else elem.text
def _get_issue_info(journal: ET.Element):
# Issue info
issue = journal.find('JournalIssue')
issue_volume = _find_elem_text(issue, 'Volume')
issue_issue = _find_elem_text(issue, 'Issue')
issue_pub_date = issue.find('PubDate')
if issue_pub_date is not None:
# Get issue year
issue_year = _find_elem_text(issue_pub_date, 'Year')
issue_year = int(issue_year) if issue_year else None
else:
issue_year = None
return {
"volume": issue_volume,
"issue": issue_issue,
"year": issue_year
}
[docs]def get_issn_info(
medline_citation: ET.Element,
get_issns_from_nlm: str = "never"
):
"""Given a medline citation, get the issn info from the article
Parameters
----------
medline_citation : xml.etree.ElementTree.Element
The MedlineCitation element of the PubMed XML tree.
get_issns_from_nlm : Literal['never', 'missing', 'always']
Whether to recover ISSN values from the NLM catalog. Options are
'never', 'missing', and 'always'. If 'missing', then the ISSN
values will be recovered from the NLM catalog if they are not found
in the XML. If 'always', then the ISSN values will be recovered from
the NLM catalog regardless of whether they are found in the XML.
Default is 'never' (i.e., never recover from NLM catalog regardless
of whether they are found in the XML).
Returns
-------
dict
A dictionary journal, issue, and ISSN info. The structure is as
follows:
{
"journal_title": str,
"journal_abbrev": str,
"journal_nlm_id": str,
"issn_dict": {
"issn": str,
"issn_l": str,
"type": "print"|"electronic"|"other",
},
"issue_dict": {
"volume": str,
"issue": str,
"year": int
}
}
"""
if get_issns_from_nlm not in ['never', 'missing', 'always']:
raise ValueError("get_issns_from_nlm must be one of 'never', "
"'missing', or 'always'")
# Journal info
journal = medline_citation.find('Article/Journal')
journal_title = _find_elem_text(journal, 'Title')
journal_abbrev = _find_elem_text(journal, 'ISOAbbreviation')
# Issue info
issue_info = _get_issue_info(journal)
# Get the ISSN from the article record
issn_dict = {}
issn_element = journal.find("ISSN")
if issn_element is not None:
issn_type = issn_element.attrib.get("IssnType", "other").lower()
issn = issn_element.text
issn_dict["issn"] = issn
issn_dict["type"] = issn_type
# Get the linking ISSN from the article record
issn_linking = _find_elem_text(medline_citation,
"MedlineJournalInfo/ISSNLinking")
if issn_linking:
issn_dict["issn_l"] = issn_linking
nlm_id = _find_elem_text(medline_citation,
'MedlineJournalInfo/NlmUniqueID')
# Get ISSN values from the NLM catalog
if nlm_id and (
get_issns_from_nlm == 'always' or
get_issns_from_nlm == 'missing' and not any(issn_dict.values())
):
nlm_issn_list = get_issns_for_journal(nlm_id)
if nlm_issn_list:
issn_dict['alternate_issns'] = nlm_issn_list
return {
"journal_title": journal_title,
"journal_abbrev": journal_abbrev,
"journal_nlm_id": nlm_id,
"issn_dict": issn_dict,
"issue_dict": issue_info,
}
def _get_journal_info(medline_citation, get_issns_from_nlm: bool):
# Journal info
journal = medline_citation.find('Article/Journal')
journal_title = _find_elem_text(journal, 'Title')
journal_abbrev = _find_elem_text(journal, 'ISOAbbreviation')
# Issue info
issue_info = _get_issue_info(journal)
# Add the ISSN from the article record
issn_set = set()
issn = _find_elem_text(journal, 'ISSN')
if issn:
issn_set.add(issn)
# Add the Linking ISSN from the article record
issn_linking = _find_elem_text(medline_citation,
'MedlineJournalInfo/ISSNLinking')
if issn_linking:
issn_set.add(issn_linking)
# Now get the list of ISSNs from the NLM Catalog
nlm_id = _find_elem_text(medline_citation,
'MedlineJournalInfo/NlmUniqueID')
if nlm_id and get_issns_from_nlm:
nlm_issn_list = get_issns_for_journal(nlm_id)
if nlm_issn_list:
issn_set.update(v for _, v in nlm_issn_list)
# Remove any duplicate issns
issn_list = list(issn_set)
return {
'journal_title': journal_title,
'journal_abbrev': journal_abbrev,
'issn_list': issn_list,
'issn_l': issn_linking,
'journal_nlm_id': nlm_id,
'issue': issue_info['issue'],
'volume': issue_info['volume'],
'year': issue_info['year'],
}
def _get_pubmed_publication_date(pubmed_data):
date_dict = dict.fromkeys(['year', 'month', 'day'])
# Order potential statuses in order of preferences
status_list = ['pubmed', 'accepted', 'revised', 'received', 'entrez']
# Look for various statuses, in order of preference as PubStatus in
# PubmedPubDate
for status in status_list:
pubmed_pub_date = \
pubmed_data.find('./History/PubMedPubDate[@PubStatus="%s"]'
% status)
if pubmed_pub_date is not None:
break
else:
logger.warning("Could not find pub date in: \n%s"
% ET.tostring(pubmed_data).decode('utf-8'))
return date_dict
def _find_date(element):
value = _find_elem_text(pubmed_pub_date, element)
return int(value) if value else None
# Get date elements from extracted pubmed_pub_date element
for date_elem in ['Year', 'Month', 'Day']:
date_dict[date_elem.lower()] = _find_date(date_elem)
return date_dict
def _parse_author(author_info, include_details=False):
if not include_details:
last_name = author_info.find("LastName")
if last_name is None:
return None
return last_name.text
parsed_info = {
"last_name": None,
"first_name": None,
"initials": None,
"suffix": None,
"identifier": None,
"affiliations": None,
}
affiliations = []
for element in author_info.findall("*"):
if element.tag == "AffiliationInfo":
affiliation_name = element.find("Affiliation").text
identifiers = [e.text for e in element.findall("Identifier")]
affiliations.append({"name": affiliation_name, "identifiers": identifiers})
elif element.tag == "LastName":
parsed_info["last_name"] = element.text
elif element.tag == "ForeName":
parsed_info["first_name"] = element.text
elif element.tag == "Initials":
parsed_info["initials"] = element.text
elif element.tag == "Suffix":
parsed_info["suffix"] = element.text
elif element.tag == "Identifier":
parsed_info["identifier"] = element.text
parsed_info["affiliations"] = affiliations
return parsed_info
def _get_references(reference_list, only_pmid=True):
"""Return a list of references for an article."""
if reference_list is None:
return None
references = []
for reference in reference_list.findall('Reference'):
pmid = _find_elem_text(reference, '*/ArticleId[@IdType="pubmed"]')
if only_pmid:
references.append(pmid)
else:
ref_dict = {
'pmid': pmid,
'doi': _find_elem_text(reference, '*/ArticleId[@IdType="doi"]'),
'pmcid': _find_elem_text(reference, '*/ArticleId[@IdType="pmcid"]'),
'citation': _find_elem_text(reference, 'Citation'),
}
references.append(ref_dict)
return references
def _get_article_info(medline_citation, pubmed_data, detailed_authors=False):
article = medline_citation.find('Article')
pmid = _find_elem_text(medline_citation, './PMID')
pii = _find_elem_text(article,
'./ELocationID[@EIdType="pii"][@ValidYN="Y"]')
# Look for the DOI in the ELocationID field...
doi = _find_elem_text(article,
'./ELocationID[@EIdType="doi"][@ValidYN="Y"]')
# ...and if that doesn't work, look in the ArticleIdList
if doi is None:
doi = _find_elem_text(pubmed_data, './/ArticleId[@IdType="doi"]')
# Try to get the PMCID
pmcid = _find_elem_text(pubmed_data, './/ArticleId[@IdType="pmc"]')
# Title
title = _get_title_from_article_element(article)
# Author list
author_elems = article.findall('AuthorList/Author')
author_names = None if author_elems is None \
else [_parse_author(au, detailed_authors) for au in author_elems]
# Get the page number entry
page = _find_elem_text(article, 'Pagination/MedlinePgn')
return {'pmid': pmid, 'pii': pii, 'doi': doi, 'pmcid': pmcid,
'title': title, 'authors': author_names, 'page': page}
[docs]def get_mesh_annotations(pmid):
"""Return a list of MeSH annotations for a given PubMed ID.
Parameters
----------
pmid : str
A PubMed ID.
Returns
-------
list of dict
A list of dicts that represent MeSH annotations with the following keys:
"mesh" representing the MeSH ID, "text" the standrd name associated with
the MeSH ID, "major_topic" a boolean flag set depending on whether
the given MeSH ID is assigned as a major topic to the article, and
"qualifier" which is a MeSH qualifier ID associated with the annotation,
if available, otherwise None.
"""
full_xml_tree = get_full_xml(pmid)
if not full_xml_tree:
return None
medline_citation = full_xml_tree.find('PubmedArticle/MedlineCitation')
if not medline_citation:
return None
annotations = _get_annotations(medline_citation)
return annotations.get('mesh_annotations')
def _get_annotations(medline_citation):
def _major_topic(e):
if e is not None and e.get('MajorTopicYN').upper() == 'Y':
return True
return False
info = []
for elem in medline_citation.findall('.//MeshHeading'):
dname = elem.find('DescriptorName')
qualifier_elems = elem.findall('QualifierName')
mid = dname.attrib['UI']
major = _major_topic(dname) or any(_major_topic(qual) for qual
in qualifier_elems)
qualifiers = [{'text': qual.text, 'mesh': qual.attrib['UI']}
for qual in qualifier_elems]
qual = qualifiers[0] if qualifiers else None
info.append({'type': 'main', 'mesh': mid, 'text': dname.text,
'major_topic': major,
# This is only here for backwards compatibility with
# INDRA DB which expects a single qualifier or None and
# turns the single qualifier into an int internally, so
# we can't easily put a joined string of multiple
# qualifiers here.
'qualifier': qual,
# This is the proper full list of qualifiers
'qualifiers': qualifiers})
for elem in medline_citation.findall('.//SupplMeshList/SupplMeshName'):
info.append({'type': 'supplementary', 'mesh': elem.attrib['UI'], 'text': elem.text,
'qualifier': None, 'qualifiers': [],
'major_topic': False})
return {'mesh_annotations': info}
[docs]@lru_cache(maxsize=1000)
def get_issns_for_journal(nlm_id):
"""Get a dict of the ISSN numbers for a journal given its NLM ID.
Information on NLM XML DTDs is available at
https://www.nlm.nih.gov/databases/dtd/
"""
params = {'db': 'nlmcatalog',
'retmode': 'xml',
'id': nlm_id}
tree = send_request(pubmed_fetch, params)
if tree is None:
return None
issn_list = [(e.attrib.get("IssnType", "other").lower(), e.text)
for e in tree.findall('.//ISSN')]
issn_linking = tree.find('.//ISSNLinking')
if issn_linking:
issn_list.append(("linking", issn_linking.text))
# No ISSNs found!
if not any(v for k, v in issn_list):
return None
return issn_list
[docs]def expand_pagination(pages):
"""Convert a page number to long form, e.g., from 456-7 to 456-457."""
# If there is no hyphen, it's a single page, and we're good to go
parts = pages.split('-')
if len(parts) == 1: # No hyphen, so no split
return pages
elif len(parts) == 2:
start = parts[0]
end = parts[1]
# If the end is the same number of digits as the start, then we
# don't change anything!
if len(start) == len(end):
return pages
# Otherwise, replace the last digits of start with the digits of end
num_end_digits = len(end)
new_end = start[:-num_end_digits] + end
return '%s-%s' % (start, new_end)
else: # More than one hyphen, something weird happened
logger.warning("Multiple hyphens in page number: %s" % pages)
return pages
[docs]def get_substance_annotations(pubmed_id: str) -> List[str]:
"""Return substance MeSH ID for a given PubMedID.
Note that substance annotations often overlap with MeSH annotations,
however, there are cases where a substance annotation is not available
under MeSH annotations.
Parameters
----------
pubmed_id :
PubMedID ID whose substance MeSH ID will be returned.
Returns
-------
:
Substance MeSH IDs corresponding to the given PubMed paper or
if None present or a failed query, an empty list will be returned.
"""
root = get_full_xml(pubmed_id)
nodes = root.findall('.//MedlineCitation/ChemicalList')
if len(nodes) == 0:
logger.error('Could not retrieve substance MeSH IDs for %s' % pubmed_id)
return []
uid = [b.attrib.get('UI') for node in nodes
for c in list(node) for b in c.iter('*')
if 'UI' in b.attrib]
return uid