PubMed Entrez API 批量下载实战

说实话,手动去PubMed网站一条条复制摘要,几百篇文献搞下来手都得抽筋。我之前接过一个药物重定位项目,要分析两千多篇文献,光是复制粘贴就花了两天,还漏了不少。后来痛下决心,把Entrez API吃透了——嗯,从那以后,这类活儿基本就是几分钟的事。

这一章,我带你走一遍完整流程:怎么用Python调Entrez API,怎么批量拉摘要,怎么结构化存下来。你跟着敲一遍代码,以后遇到文献批量下载,心里就有底了。

本章目标:掌握PubMed Entrez API的核心调用方式,能写出批量下载摘要并结构化的Python脚本。

3.1 Entrez API 是什么?

Entrez是NCBI提供的一个统一数据检索系统。PubMed只是它旗下的一部分。通过Entrez API,你可以在程序里直接查文献、拿摘要、抓MeSH词条,甚至下载全文链接。

说白了,就是给你一个HTTP接口,你用GET/POST发请求,它返回XML或JSON格式的数据。我们不用浏览器点来点去,写几行代码就能搞定。

核心的API端点有三个:

  • ESearch:根据查询词返回PubMed ID(PMIDs)列表
  • EFetch:根据PMIDs拉取完整记录(摘要、作者、期刊等)
  • ESummary:根据PMIDs拉取摘要信息,比EFetch轻量一些

我个人习惯用 ESearch + EFetch 这个组合。ESearch拿到ID列表,EFetch一次最多拉200条记录的完整数据,效率很高。

3.2 准备工作:注册API Key

NCBI对未注册的IP有限制——每秒3个请求。如果你不注册Key,批量下载时很容易被429(Too Many Requests)拦下来。我建议一开始就把Key注册好。

注册很简单:

  1. 登录NCBI网站(https://www.ncbi.nlm.nih.gov/)
  2. 进入账号设置,找到API Key管理
  3. 申请一个Key,免费

有了Key之后,每秒可以发10个请求。对于大多数项目,这个量完全够用了。

我的习惯:把Key存在环境变量 NCBI_API_KEY 里,脚本里用 os.getenv() 读取。这样代码发到GitHub上也不会暴露Key。

3.3 核心调用:ESearch + EFetch

下面我写一个完整的Python示例。这个脚本会搜索阿托伐他汀(Atorvastatin)与CYP3A4相关的文献,下载最近3年的摘要。

先装好依赖库:

pip install biopython requests

Biopython封装了Entrez API的调用逻辑,比自己拼URL要方便很多。你如果愿意用requests直接调也没问题,但我建议新手先从Biopython入手。

完整代码:

import os
import time
from Bio import Entrez

# 设置邮箱和API Key
Entrez.email = "your_email@example.com"
Entrez.api_key = os.getenv("NCBI_API_KEY", "")

def search_pubmed(query, max_results=100):
    """根据查询词获取PMIDs列表"""
    handle = Entrez.esearch(
        db="pubmed",
        term=query,
        retmax=max_results,
        sort="relevance"
    )
    record = Entrez.read(handle)
    handle.close()
    return record["IdList"]

def fetch_abstracts(pmids):
    """根据PMIDs列表批量拉取摘要"""
    if not pmids:
        return []
    handle = Entrez.efetch(
        db="pubmed",
        id=pmids,
        retmode="xml",
        rettype="abstract"
    )
    records = Entrez.read(handle)
    handle.close()
    return records

# 搜索阿托伐他汀与CYP3A4的相关文献
query = "atorvastatin AND CYP3A4 AND 2022:2024[dp]"
pmid_list = search_pubmed(query, max_results=50)
print(f"找到 {len(pmid_list)} 篇文献")

# 每200篇一批拉取,避免服务器压力
batch_size = 20
all_articles = []
for i in range(0, len(pmid_list), batch_size):
    batch = pmid_list[i:i+batch_size]
    articles = fetch_abstracts(batch)
    all_articles.extend(articles)
    print(f"已拉取 {i+len(batch)} / {len(pmid_list)} 篇")
    # 加个延时,别把NCBI打挂了
    time.sleep(0.5)

print(f"总共拉取 {len(all_articles)} 篇摘要")

运行完这段代码,all_articles 里就是结构化数据了。每个元素是一个字典,里面包含PubMedId、Title、Abstract、Authors、Journal、PubDate等字段。

注意:EFetch返回的记录结构比较深。如果你直接打印 articles,可能会看到一堆嵌套的字典和列表。我建议写一个解析函数,把需要的字段抽出来存成扁平结构——比如CSV或者DataFrame。

3.4 解析EFetch返回的数据

看一个具体的解析例子:

def parse_article(article):
    """从EFetch返回的条目中提取关键信息"""
    medline = article.get("MedlineCitation", {})
    article_data = medline.get("Article", {})
    
    # PubMed ID
    pmid = str(medline.get("PMID", ""))
    
    # 标题
    title = article_data.get("ArticleTitle", "")
    
    # 摘要(可能是多段,合并成字符串)
    abstract_parts = article_data.get("Abstract", {}).get("AbstractText", [])
    abstract = " ".join(
        [part.get("_text", "") if isinstance(part, dict) else str(part)
         for part in abstract_parts]
    )
    
    # 作者列表(只取姓和名首字母)
    authors = []
    author_list = article_data.get("AuthorList", [])
    for author in author_list:
        last = author.get("LastName", "")
        initials = author.get("Initials", "")
        authors.append(f"{last} {initials}")
    
    # 期刊名
    journal = article_data.get("Journal", {}).get("Title", "")
    
    # 出版日期
    pub_date = article_data.get("Journal", {}).get(
        "JournalIssue", {}
    ).get("PubDate", {})
    year = pub_date.get("Year", "")
    month = pub_date.get("Month", "")
    day = pub_date.get("Day", "")
    
    return {
        "pmid": pmid,
        "title": title,
        "abstract": abstract,
        "authors": "; ".join(authors),
        "journal": journal,
        "pub_date": f"{year}-{month}-{day}"
    }

调用起来也很简单:

parsed = [parse_article(a) for a in all_articles]
# 存成CSV
import csv
with open("atorvastatin_cyp3a4.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=parsed[0].keys())
    writer.writeheader()
    writer.writerows(parsed)
print("数据已保存 to atorvastatin_cyp3a4.csv")

3.5 ESummary 的妙用

有时候你不想拉完整的摘要,只想要标题、作者、发表年份做初步筛选。这种场景下,用 ESummary 比EFetch快得多。

ESummary返回的是精简版数据:

def fetch_summaries(pmids):
    """使用ESummary获取精简信息"""
    handle = Entrez.esummary(
        db="pubmed",
        id=pmids,
        retmode="json"
    )
    records = Entrez.read(handle)
    handle.close()
    return records

# 拿前5篇的摘要信息试一下
summaries = fetch_summaries(pmid_list[:5])
for uid, summary in summaries.items():
    print(f"{uid}: {summary['Title']}")

ESummary返回的数据里,Title、Source、PubDate、AuthorList这些字段都有,但没有完整的Abstract。我个人的使用策略是:

  • 先用ESummary做快速筛选,过滤掉不相关的文献
  • 再对筛选后的PMIDs调用EFetch,拉完整摘要

这样能省掉大量不必要的请求——你想想看,如果一开始就拉两千篇完整摘要,中间如果有一半不相关,就白费了。

3.6 避坑指南

我在实战中踩过不少坑,挑几个最常见的说说:

我曾经遇到的最大坑:忘记了延时控制。有一次写循环时忘了加 time.sleep(),结果1000篇文献的请求在5秒内全发出去了。NCBI直接把我IP封了24小时。后来我养成了一个习惯——不管用不用API Key,都在循环里加0.3秒以上的延时。

另外几个要注意的地方:

  • ESearch的retmax上限是10000:如果你要检索的文献超过一万篇,必须用WebEnv+query_key做分批翻页,不能指望一次拉完
  • EFetch单次最多200条:不要传一个5000的列表进去,会被截断
  • XML解析有时会丢字段:有些记录缺少AuthorList或Abstract,你的解析函数要做空值处理,否则会报KeyError
  • 中文摘要问题:PubMed里中文文献不多,但如果遇到,EFetch返回的字段编码是UTF-8,一般没问题——不过个别老文献可能用GB2312,你得加个异常捕获做编码探测

3.7 批量下载的完整流程

把上面的知识点串起来,一个完整的流程是这样的:

  1. 构造查询词(带日期范围、字段限定符)
  2. ESearch获取PMIDs列表
  3. 分批通过ESummary快速筛选(可选)
  4. 对筛选后的PMIDs分批调用EFetch拉完整记录
  5. 解析XML,提取结构化字段
  6. 存成CSV、JSON或直接导入DataFrame做后续分析

这个流程我已经在十几个项目里跑过了,从几十篇到两万篇都验证过。只要注意API限流和错误处理,基本不会出问题。

3.8 核心流程图

下面这张图把整个数据流展示出来,方便你对照理解:

Entrez API 批量下载摘要流程 用户查询词 ESearch → PMIDs列表 ESummary 快速筛选 通过? 调整查询 EFetch 拉取完整记录 关键参数 db="pubmed" retmode="xml/json" rettype="abstract" batch_size ≤ 200 输出格式 CSV / JSON DataFrame 结构化字段

3.9 小结

这一章的核心就是三件事:ESearch查ID → ESummary/EFetch拿数据 → 解析存结构化。代码我已经拆成函数了,你可以在自己的项目里直接复用。记得加上延时和异常处理——这是从我的惨痛教训里提炼出来的。

下个环节,我们会拿这些摘要数据做药物-靶点关系的自动化抽取。那时候你就能体会到,先把数据批量拉下来、结构化了,后面的分析会顺手很多。