1. 项目概述为什么我们需要批量获取化合物信息在药物研发、材料科学、环境监测乃至食品添加剂分析等领域化学信息学Cheminformatics正扮演着越来越核心的角色。无论是进行虚拟筛选、构建定量构效关系QSAR模型还是简单地整理一个化合物库第一步往往都是获取化合物的基础数据分子式、分子量、SMILES结构式、LogP、氢键供受体数等等。手动去PubChem、ChemSpider这些大型公共数据库一个个查面对成百上千个化合物这无异于一场噩梦。我最近就遇到了一个典型场景实验室新合成了一批衍生物导师让我快速评估一下这批化合物的类药五原则Lipinski‘s Rule of Five符合情况。名单上有87个化合物如果手动操作光是复制粘贴就能耗掉一整天还极易出错。这时候一个自动化的批量下载工具就成了救命稻草。PubChem作为全球最大的免费化学信息数据库自然是首选数据源。而PubChemPy这个Python库就是连接我们和这座化学数据宝库的桥梁。它封装了PubChem的PUG REST API让我们能用几行代码就轻松获取海量化合物的标准化信息。这个项目的核心价值就在于将繁琐、重复的手工查询工作自动化解放研究人员的双手和大脑把时间留给更有创造性的数据分析和科学发现。无论是处理几十个还是上万个化合物其逻辑都是一致的一次编写终身受用。2. 核心工具解析PubChemPy与PubChem PUG REST API在动手写代码之前我们得先搞清楚手里的工具到底是什么以及它背后的工作原理。这能帮助我们在遇到问题时知道该往哪个方向排查。2.1 PubChemPy让API调用变得像喝水一样简单PubChemPy并不是PubChem官方维护的库而是一个由社区开发者贡献的、非常优秀的第三方Python封装库。它的设计哲学就是“简单”。它把PubChem那套功能强大但略显复杂的PUG REST API包装成了高度Pythonic的、面向对象的方法。举个例子如果你想获取阿司匹林Aspirin的化合物标识符CID用原生的HTTP请求你需要构造一个这样的URL并处理返回的XML或JSONhttps://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/name/aspirin/cids/JSON而用PubChemPy你只需要import pubchempy as pcp cids pcp.get_cids(Aspirin, name)看是不是直观多了它自动处理了URL编码、请求发送、错误响应和结果解析直接返回一个干净的Python列表[2244]。它的主要功能模块包括化合物查询通过名称、SMILES、InChIKey、CID等获取化合物对象。属性获取批量获取化合物的物理化学属性、二维/三维结构、生物活性等。下载下载化合物的SDF、JSON、XML等格式的数据文件。交互进行化学结构相似性搜索、子结构搜索等。注意PubChemPy是一个“薄”封装。这意味着它主要提供便捷的访问方式但对于非常高级或最新的PubChem API功能有时可能需要等待库更新或者直接使用requests库调用底层API。不过对于99%的批量下载需求它都绰绰有余。2.2 PubChem PUG REST API数据之源理解PubChemPy的底层——PUG REST API有助于我们理解一些限制和优化策略。PUG代表“PubChem Power User Gateway”。关键特性与限制请求频率限制PubChem对未经验证的请求有频率限制大约每秒5-10次请求。这是我们在设计批量下载程序时必须首要考虑的因素。无脑快速循环请求会导致IP被暂时封锁。批量处理API本身支持批量查询这是提升效率的关键。例如可以通过一个请求查询多个CID的属性而不是一个个查。PubChemPy的get_properties函数就利用了这一点。数据格式支持JSON、XML、SDF、PNG结构图等多种返回格式。JSON最便于用Python处理。丰富的数据域除了基本的分子信息还可以获取生物活性BioAssay、文献引用、供应商信息、计算描述符等。我们的批量下载脚本本质上就是围绕这些API特性在效率批量请求和稳定性遵守频率限制之间找到最佳平衡点。3. 实战构建健壮的化合物信息批量下载器光说不练假把式。接下来我将一步步拆解如何构建一个可用于生产环境的批量下载脚本。这个脚本不仅要能跑通还要考虑错误处理、日志记录和性能确保处理上千个化合物时也能稳定运行。3.1 环境准备与依赖安装首先确保你的Python环境建议3.7以上已经就绪。安装PubChemPy非常简单pip install pubchempy此外我们还会用到pandas来处理和保存表格数据用time模块来控制请求间隔。pip install pandas我习惯将项目结构组织如下compound_downloader/ ├── config.py # 存放配置如输入文件路径、输出目录、请求延迟等 ├── downloader.py # 核心下载逻辑 ├── utils.py # 工具函数如日志设置、数据清洗 ├── input/ # 存放包含化合物名称列表的文件如compounds.txt ├── output/ # 程序运行后结果文件会生成在这里 └── run.py # 主运行脚本3.2 核心下载逻辑设计与实现假设我们的输入是一个纯文本文件compounds.txt每行一个化合物名称或SMILES、InChIKey等。我们的目标是输出一个CSV文件包含每个化合物的名称、CID、分子式、分子量、SMILES、LogP等关键信息。步骤1读取化合物列表# utils.py 或 downloader.py 中 def load_compound_list(file_path): 从文本文件加载化合物列表并去重、去空 with open(file_path, r, encodingutf-8) as f: # 读取所有行去除首尾空白过滤掉空行 compounds [line.strip() for line in f if line.strip()] # 去重但保留原始顺序如果需要 seen set() unique_compounds [] for comp in compounds: if comp not in seen: seen.add(comp) unique_compounds.append(comp) print(f从 {file_path} 加载了 {len(compounds)} 个条目去重后剩余 {len(unique_compounds)} 个唯一化合物。) return unique_compounds步骤2分批次获取CID化合物标识符CID是PubChem中每个化合物的唯一身份证。我们首先需要根据名称获取CID。这里有一个重要技巧批量查询和延迟重试。# downloader.py import pubchempy as pcp import time import logging def get_cids_in_batch(compound_names, batch_size100, delay0.2): 批量获取化合物CID并加入延迟以避免触发API限制。 参数: compound_names: 化合物名称列表 batch_size: 每批查询的数量PubChem API单次请求上限通常较高但保守点设为100 delay: 批次之间的延迟秒 返回: dict: {化合物名称: CID} 的映射字典。未找到的化合物CID为None。 name_to_cid {} total len(compound_names) for i in range(0, total, batch_size): batch compound_names[i:ibatch_size] logging.info(f正在处理第 {i//batch_size 1}/{(total-1)//batch_size 1} 批本批 {len(batch)} 个化合物...) try: # 关键使用get_cids函数传入列表进行批量查询 # namespace 指定输入类型为name results pcp.get_cids(batch, namespacename, list_returnflat) # 注意get_cids返回一个扁平列表顺序与输入batch一致。 # 如果某个名称未找到对应位置为0。 for name, cid in zip(batch, results): name_to_cid[name] cid if cid ! 0 else None except pcp.BadRequestError as e: logging.error(f批次 {batch} 请求失败 (BadRequest): {e}) # 如果整个批次失败将本批所有化合物标记为失败 for name in batch: name_to_cid[name] None except Exception as e: logging.error(f批次 {batch} 请求发生未知错误: {e}) # 同样标记本批失败 for name in batch: name_to_cid[name] None # 批次间延迟遵守API频率限制 if i batch_size total: # 如果不是最后一批 time.sleep(delay) # 统计结果 found sum(1 for cid in name_to_cid.values() if cid is not None) not_found sum(1 for cid in name_to_cid.values() if cid is None) logging.info(fCID查询完成。成功找到 {found} 个未找到 {not_found} 个。) return name_to_cid实操心得pcp.get_cids的list_returnflat参数非常关键。它返回一个与输入列表长度一致的简单列表使得名称与CID的对应关系清晰明了。如果不指定它可能返回一个字典处理起来稍麻烦。另外将未找到的化合物CID设为None而不是0或-1在后续的Pandas数据处理中更容易过滤和识别。步骤3根据CID批量获取化合物属性拿到CID后我们就可以获取详细属性了。PubChemPy的get_properties函数是另一个支持批量操作的利器。def get_compound_properties(cid_list, propertiesNone, delay0.2): 根据CID列表批量获取化合物属性。 参数: cid_list: CID整数列表 properties: 需要获取的属性名列表。为None时使用默认常用属性集。 delay: 每次请求后的延迟秒PubChem对属性查询的限制可能更严格。 返回: list: 字典列表每个字典对应一个化合物的属性。失败的条目为None。 if properties is None: # 定义一组常用且可靠的化学信息学描述符 properties [MolecularFormula, MolecularWeight, CanonicalSMILES, XLogP, HydrogenBondDonorCount, HydrogenBondAcceptorCount, RotatableBondCount, TPSA, Complexity] all_results [] # 这里我们不再分批次因为get_properties函数内部可能已经处理了批量。 # 但为了绝对稳妥特别是CID列表很长时我们可以手动分小批。 batch_size 50 # 属性查询的批次可以小一些 total_cids len(cid_list) for i in range(0, total_cids, batch_size): batch_cids cid_list[i:ibatch_size] logging.info(f正在获取属性批次 {i//batch_size 1}/{(total_cids-1)//batch_size 1}CIDs: {batch_cids[:5]}...) try: # 核心调用批量获取属性 results pcp.get_properties(properties, batch_cids, as_dataframeFalse) # as_dataframeFalse 返回字典列表更方便我们与CID对应 all_results.extend(results) except pcp.NotFoundError: # 如果整个批次未找到可能性小添加None占位 logging.warning(f批次CID {batch_cids} 未找到属性。) all_results.extend([None] * len(batch_cids)) except pcp.BadRequestError as e: logging.error(f属性请求失败 (BadRequest) for batch {batch_cids}: {e}) all_results.extend([None] * len(batch_cids)) except Exception as e: logging.error(f属性请求未知错误 for batch {batch_cids}: {e}) all_results.extend([None] * len(batch_cids)) if i batch_size total_cids: time.sleep(delay) # 严格遵守延迟 return all_results注意事项属性名如‘XLogP’是大小写敏感的必须严格按照PubChem的文档来。as_dataframeFalse参数确保我们拿到的是Python字典列表方便后续与我们的名称-CID映射进行合并。如果设置为True会返回一个Pandas DataFrame但当某些化合物缺失某些属性时索引对齐可能会有点棘手。步骤4整合数据并保存现在我们有name_to_cid字典和property_list列表。需要将它们合并并处理可能存在的缺失或失败情况。import pandas as pd def compile_results(compound_names, name_to_cid, property_list): 将名称、CID和属性列表整合成一个Pandas DataFrame。 data [] for name in compound_names: cid name_to_cid.get(name) # 找到该名称在property_list中的索引 # 由于我们处理时顺序一致可以通过遍历或构建映射来关联。更稳健的方法是构建CID到属性的映射。 prop_dict None if cid is not None: # 在property_list中查找CID匹配的项 # 注意property_list中的每一项是一个字典其中包含CID键 for prop in property_list: if prop and prop.get(CID) cid: prop_dict prop break row {Compound_Name: name, CID: cid} if prop_dict: # 将属性字典中的键值对添加到行中 # 排除已经有的CID键避免重复 for key, value in prop_dict.items(): if key ! CID: row[key] value else: # 如果没有找到属性将所有属性列设为NaN # 这里我们需要知道有哪些属性列可以从第一个非None的属性字典获取keys if property_list and property_list[0] is not None: for key in property_list[0].keys(): if key ! CID: row[key] None data.append(row) df pd.DataFrame(data) # 调整列顺序让名称和CID在前 cols [Compound_Name, CID] [col for col in df.columns if col not in [Compound_Name, CID]] df df[cols] return df def save_to_csv(df, output_path): 保存DataFrame到CSV文件并处理可能的中文编码问题 df.to_csv(output_path, indexFalse, encodingutf-8-sig) # utf-8-sig确保Excel打开不乱码 logging.info(f结果已保存至: {output_path})3.3 主程序流程与异常处理框架将上述函数串联起来并加入完整的日志和异常处理就构成了我们的主程序。# run.py import logging import sys from config import INPUT_FILE, OUTPUT_FILE, REQUEST_DELAY from utils import load_compound_list, setup_logging from downloader import get_cids_in_batch, get_compound_properties, compile_results, save_to_csv def main(): # 1. 设置日志 setup_logging() logging.info(*50) logging.info(化合物信息批量下载程序启动) logging.info(*50) # 2. 加载化合物列表 try: compound_names load_compound_list(INPUT_FILE) if not compound_names: logging.error(输入文件为空或格式错误程序退出。) sys.exit(1) except FileNotFoundError: logging.error(f未找到输入文件: {INPUT_FILE}) sys.exit(1) except Exception as e: logging.error(f读取输入文件时发生错误: {e}) sys.exit(1) # 3. 批量获取CID logging.info(阶段1根据化合物名称查询CID...) name_to_cid get_cids_in_batch(compound_names, delayREQUEST_DELAY) # 提取所有成功找到的CID valid_cids [cid for cid in name_to_cid.values() if cid is not None] logging.info(f共有 {len(valid_cids)} 个有效CID用于查询属性。) if not valid_cids: logging.warning(没有找到任何有效的CID程序将只输出名称和空CID。) # 创建一个只有名称和CID的DataFrame df pd.DataFrame([{Compound_Name: name, CID: cid} for name, cid in name_to_cid.items()]) save_to_csv(df, OUTPUT_FILE) logging.info(程序结束无属性数据。) return # 4. 批量获取化合物属性 logging.info(阶段2根据CID批量获取化合物属性...) property_list get_compound_properties(valid_cids, delayREQUEST_DELAY) # 5. 整合数据 logging.info(阶段3整合数据...) df compile_results(compound_names, name_to_cid, property_list) # 6. 保存结果 save_to_csv(df, OUTPUT_FILE) # 7. 打印简要统计 success_count df[CID].notna().sum() prop_success_count df.dropna(subset[MolecularWeight]).shape[0] # 以某个关键属性为例 logging.info(f任务完成共处理 {len(compound_names)} 个化合物。) logging.info(f - 成功解析CID: {success_count} 个) logging.info(f - 成功获取属性: {prop_success_count} 个) logging.info(*50) if __name__ __main__: main()对应的config.py和utils.py示例# config.py INPUT_FILE ./input/compounds.txt OUTPUT_FILE ./output/compound_info.csv REQUEST_DELAY 0.3 # 秒保守起见设置比API限制稍高的延迟 LOG_FILE ./output/download.log # utils.py import logging def setup_logging(log_file./output/download.log): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) ] )4. 高级技巧与性能优化一个能跑的程序和一个好用的工具之间往往差的就是这些细节的打磨。4.1 处理模糊名称与同义词化合物名称常常不唯一。“Aspirin”也被称为“Acetylsalicylic acid”。PubChemPy的get_cids在名称匹配时默认使用“最接近”的匹配。但有时它会返回多个CID例如查询一个常见缩写时。get_cids的返回类型取决于参数list_returnflat: 返回列表每个输入名称对应一个CID如果有多个匹配可能是第一个。list_returndict: 返回字典键是输入名称值是该名称匹配到的所有CID的列表。策略对于关键任务建议先使用list_returndict检查是否有名称匹配到了多个CID然后人工或通过规则如选择第一个或选择分子量在某个范围的来确定一个。可以在get_cids_in_batch函数中增加这个逻辑。def get_cids_verbose(compound_names): 获取CID并记录匹配详情 result pcp.get_cids(compound_names, namespacename, list_returndict) ambiguous {} for name, cid_list in result.items(): if len(cid_list) 1: ambiguous[name] cid_list logging.warning(f化合物 {name} 匹配到多个CID: {cid_list}将使用第一个: {cid_list[0]}) # 即使多个我们也取第一个作为代表 name_to_cid[name] cid_list[0] if cid_list else None return name_to_cid, ambiguous4.2 异步请求与速率控制优化当处理成千上万个化合物时即使每批100个批次间的固定延迟如0.3秒也会导致总时间很长。更高级的策略是使用异步IO如asyncio和aiohttp来并发发送请求同时使用令牌桶Token Bucket等算法精确控制总体请求速率在不超过PubChem限制的前提下最大化效率。不过这引入了显著的复杂性。对于大多数实验室规模的批量任务几百到几千个简单的批次延迟已经足够可靠。我的建议是除非你经常需要处理超过5000个化合物的大列表否则优先选择简单稳定的同步方案。复杂性是万恶之源一个稳定运行但稍慢的脚本远胜于一个快但偶尔崩溃或漏数据的脚本。4.3 数据缓存与断点续传对于超大规模任务或者网络不稳定的环境实现缓存和断点续传是专业性的体现。缓存将已成功查询的{名称: CID}和{CID: 属性}以JSON格式保存到本地文件。下次运行时先加载缓存只查询未缓存的部分。断点续传将任务列表化合物名称本身也保存进度。程序启动时读取进度文件从中断处继续。这需要将主循环改造为更状态化的形式。这两个功能可以极大地提升使用体验特别是在调试阶段不用每次失败都从头开始。5. 常见问题排查与实战心得在实际运行中你肯定会遇到各种各样的问题。下面是我踩过的一些坑和解决方案。5.1 网络问题与超时处理PubChem服务器在国外国内访问有时不稳定。PubChemPy底层使用requests库我们可以通过修改默认会话Session来设置超时和重试。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_retry_session(retries3, backoff_factor0.5, status_forcelist(500, 502, 504)): 创建一个带重试机制的requests Session session requests.Session() retry Retry( totalretries, readretries, connectretries, backoff_factorbackoff_factor, status_forceliststatus_forcelist, ) adapter HTTPAdapter(max_retriesretry) session.mount(http://, adapter) session.mount(https://, adapter) return session # 在调用任何pcp函数之前设置自定义session pcp.session create_retry_session()5.2 化合物未找到或属性缺失这是最常见的问题。可能的原因和应对策略问题现象可能原因排查与解决思路CID查询返回None或01. 名称拼写错误或非标准。2. 该化合物不在PubChem中。3. 名称是商标名或俗名PubChem未收录。1. 检查输入文件确认名称准确。可尝试在PubChem网站手动搜索确认。2. 尝试使用化合物的SMILES或InChIKey查询如果已知。3. 记录下未找到的化合物后续手动处理。属性查询返回None或部分属性为None1. 该CID对应的记录确实缺失某些属性。2. 属性名拼写错误。3. 网络请求超时或中断。1. 在PubChem网站用该CID查看详情确认属性是否存在。2. 核对properties参数列表确保与PubChem属性名完全一致。3. 检查日志中的错误信息可能是网络问题导致整个批次失败。获取到的属性值异常如分子量为0数据在PubChem中本身如此可能是计算错误或特殊化合物。在后续数据分析步骤中加入数据清洗逻辑过滤掉明显异常的值。处理建议在compile_results函数中对于属性缺失的化合物不要直接丢弃整行而是保留行并用NaN填充缺失属性。这样在最终的数据分析阶段你可以清楚地知道哪些数据是完整的哪些需要补充或剔除。5.3 脚本运行速度太慢如果觉得脚本慢按以下顺序检查和优化检查延迟时间REQUEST_DELAY是主要瓶颈。可以尝试从0.3秒逐步降低到0.15秒观察是否会被限制。务必谨慎先用小规模列表测试。增大批次大小get_cids和get_properties的batch_size可以适当增加。对于CID查询尝试200或250。对于属性查询尝试100。注意单次请求的URL长度有限制约8000字符过大的批次会导致HTTP 414错误。并行化高级如前所述使用concurrent.futures或asyncio实现并发。关键是要控制总并发数例如使用一个固定大小的线程池如5个线程并在每个线程内部保持延迟。这比简单的批次延迟复杂得多但能显著提升速度。使用本地缓存对于重复查询的化合物缓存是终极提速方案。5.4 输出文件在Excel中打开乱码这是编码问题。确保使用df.to_csv(..., encodingutf-8-sig)。utf-8-sig会在文件开头添加一个BOM字节顺序标记帮助Excel正确识别UTF-8编码。5.5 内存占用过高处理极大量数据如十万级以上时一次性将所有结果保存在内存的列表中可能会导致内存不足。解决方案是使用流式处理或分块处理流式写入每处理完一批化合物比如1000个就立即将这一批的结果追加写入到CSV文件中然后释放这部分内存。使用数据库对于海量数据考虑使用SQLite或磁盘上的HDF5存储而不是全部放在Pandas DataFrame里。最后分享一个我个人的小习惯在脚本的output目录下除了最终的compound_info.csv我总会同时保存一个failed_compounds.txt里面记录所有未能成功获取CID或属性的化合物名称。这样当主要任务完成后我可以集中精力手动处理这一小部分“疑难杂症”而不是淹没在成千上万的成功记录里找不到它们。这个简单的习惯多次拯救了我让我能快速定位和解决最后那5%的问题。
Python自动化批量获取PubChem化合物信息：PubChemPy实战指南
发布时间：2026/7/30 7:55:43
1. 项目概述为什么我们需要批量获取化合物信息在药物研发、材料科学、环境监测乃至食品添加剂分析等领域化学信息学Cheminformatics正扮演着越来越核心的角色。无论是进行虚拟筛选、构建定量构效关系QSAR模型还是简单地整理一个化合物库第一步往往都是获取化合物的基础数据分子式、分子量、SMILES结构式、LogP、氢键供受体数等等。手动去PubChem、ChemSpider这些大型公共数据库一个个查面对成百上千个化合物这无异于一场噩梦。我最近就遇到了一个典型场景实验室新合成了一批衍生物导师让我快速评估一下这批化合物的类药五原则Lipinski‘s Rule of Five符合情况。名单上有87个化合物如果手动操作光是复制粘贴就能耗掉一整天还极易出错。这时候一个自动化的批量下载工具就成了救命稻草。PubChem作为全球最大的免费化学信息数据库自然是首选数据源。而PubChemPy这个Python库就是连接我们和这座化学数据宝库的桥梁。它封装了PubChem的PUG REST API让我们能用几行代码就轻松获取海量化合物的标准化信息。这个项目的核心价值就在于将繁琐、重复的手工查询工作自动化解放研究人员的双手和大脑把时间留给更有创造性的数据分析和科学发现。无论是处理几十个还是上万个化合物其逻辑都是一致的一次编写终身受用。2. 核心工具解析PubChemPy与PubChem PUG REST API在动手写代码之前我们得先搞清楚手里的工具到底是什么以及它背后的工作原理。这能帮助我们在遇到问题时知道该往哪个方向排查。2.1 PubChemPy让API调用变得像喝水一样简单PubChemPy并不是PubChem官方维护的库而是一个由社区开发者贡献的、非常优秀的第三方Python封装库。它的设计哲学就是“简单”。它把PubChem那套功能强大但略显复杂的PUG REST API包装成了高度Pythonic的、面向对象的方法。举个例子如果你想获取阿司匹林Aspirin的化合物标识符CID用原生的HTTP请求你需要构造一个这样的URL并处理返回的XML或JSONhttps://pubchem.ncbi.nlm.nih.gov/rest/pug/compound/name/aspirin/cids/JSON而用PubChemPy你只需要import pubchempy as pcp cids pcp.get_cids(Aspirin, name)看是不是直观多了它自动处理了URL编码、请求发送、错误响应和结果解析直接返回一个干净的Python列表[2244]。它的主要功能模块包括化合物查询通过名称、SMILES、InChIKey、CID等获取化合物对象。属性获取批量获取化合物的物理化学属性、二维/三维结构、生物活性等。下载下载化合物的SDF、JSON、XML等格式的数据文件。交互进行化学结构相似性搜索、子结构搜索等。注意PubChemPy是一个“薄”封装。这意味着它主要提供便捷的访问方式但对于非常高级或最新的PubChem API功能有时可能需要等待库更新或者直接使用requests库调用底层API。不过对于99%的批量下载需求它都绰绰有余。2.2 PubChem PUG REST API数据之源理解PubChemPy的底层——PUG REST API有助于我们理解一些限制和优化策略。PUG代表“PubChem Power User Gateway”。关键特性与限制请求频率限制PubChem对未经验证的请求有频率限制大约每秒5-10次请求。这是我们在设计批量下载程序时必须首要考虑的因素。无脑快速循环请求会导致IP被暂时封锁。批量处理API本身支持批量查询这是提升效率的关键。例如可以通过一个请求查询多个CID的属性而不是一个个查。PubChemPy的get_properties函数就利用了这一点。数据格式支持JSON、XML、SDF、PNG结构图等多种返回格式。JSON最便于用Python处理。丰富的数据域除了基本的分子信息还可以获取生物活性BioAssay、文献引用、供应商信息、计算描述符等。我们的批量下载脚本本质上就是围绕这些API特性在效率批量请求和稳定性遵守频率限制之间找到最佳平衡点。3. 实战构建健壮的化合物信息批量下载器光说不练假把式。接下来我将一步步拆解如何构建一个可用于生产环境的批量下载脚本。这个脚本不仅要能跑通还要考虑错误处理、日志记录和性能确保处理上千个化合物时也能稳定运行。3.1 环境准备与依赖安装首先确保你的Python环境建议3.7以上已经就绪。安装PubChemPy非常简单pip install pubchempy此外我们还会用到pandas来处理和保存表格数据用time模块来控制请求间隔。pip install pandas我习惯将项目结构组织如下compound_downloader/ ├── config.py # 存放配置如输入文件路径、输出目录、请求延迟等 ├── downloader.py # 核心下载逻辑 ├── utils.py # 工具函数如日志设置、数据清洗 ├── input/ # 存放包含化合物名称列表的文件如compounds.txt ├── output/ # 程序运行后结果文件会生成在这里 └── run.py # 主运行脚本3.2 核心下载逻辑设计与实现假设我们的输入是一个纯文本文件compounds.txt每行一个化合物名称或SMILES、InChIKey等。我们的目标是输出一个CSV文件包含每个化合物的名称、CID、分子式、分子量、SMILES、LogP等关键信息。步骤1读取化合物列表# utils.py 或 downloader.py 中 def load_compound_list(file_path): 从文本文件加载化合物列表并去重、去空 with open(file_path, r, encodingutf-8) as f: # 读取所有行去除首尾空白过滤掉空行 compounds [line.strip() for line in f if line.strip()] # 去重但保留原始顺序如果需要 seen set() unique_compounds [] for comp in compounds: if comp not in seen: seen.add(comp) unique_compounds.append(comp) print(f从 {file_path} 加载了 {len(compounds)} 个条目去重后剩余 {len(unique_compounds)} 个唯一化合物。) return unique_compounds步骤2分批次获取CID化合物标识符CID是PubChem中每个化合物的唯一身份证。我们首先需要根据名称获取CID。这里有一个重要技巧批量查询和延迟重试。# downloader.py import pubchempy as pcp import time import logging def get_cids_in_batch(compound_names, batch_size100, delay0.2): 批量获取化合物CID并加入延迟以避免触发API限制。 参数: compound_names: 化合物名称列表 batch_size: 每批查询的数量PubChem API单次请求上限通常较高但保守点设为100 delay: 批次之间的延迟秒 返回: dict: {化合物名称: CID} 的映射字典。未找到的化合物CID为None。 name_to_cid {} total len(compound_names) for i in range(0, total, batch_size): batch compound_names[i:ibatch_size] logging.info(f正在处理第 {i//batch_size 1}/{(total-1)//batch_size 1} 批本批 {len(batch)} 个化合物...) try: # 关键使用get_cids函数传入列表进行批量查询 # namespace 指定输入类型为name results pcp.get_cids(batch, namespacename, list_returnflat) # 注意get_cids返回一个扁平列表顺序与输入batch一致。 # 如果某个名称未找到对应位置为0。 for name, cid in zip(batch, results): name_to_cid[name] cid if cid ! 0 else None except pcp.BadRequestError as e: logging.error(f批次 {batch} 请求失败 (BadRequest): {e}) # 如果整个批次失败将本批所有化合物标记为失败 for name in batch: name_to_cid[name] None except Exception as e: logging.error(f批次 {batch} 请求发生未知错误: {e}) # 同样标记本批失败 for name in batch: name_to_cid[name] None # 批次间延迟遵守API频率限制 if i batch_size total: # 如果不是最后一批 time.sleep(delay) # 统计结果 found sum(1 for cid in name_to_cid.values() if cid is not None) not_found sum(1 for cid in name_to_cid.values() if cid is None) logging.info(fCID查询完成。成功找到 {found} 个未找到 {not_found} 个。) return name_to_cid实操心得pcp.get_cids的list_returnflat参数非常关键。它返回一个与输入列表长度一致的简单列表使得名称与CID的对应关系清晰明了。如果不指定它可能返回一个字典处理起来稍麻烦。另外将未找到的化合物CID设为None而不是0或-1在后续的Pandas数据处理中更容易过滤和识别。步骤3根据CID批量获取化合物属性拿到CID后我们就可以获取详细属性了。PubChemPy的get_properties函数是另一个支持批量操作的利器。def get_compound_properties(cid_list, propertiesNone, delay0.2): 根据CID列表批量获取化合物属性。 参数: cid_list: CID整数列表 properties: 需要获取的属性名列表。为None时使用默认常用属性集。 delay: 每次请求后的延迟秒PubChem对属性查询的限制可能更严格。 返回: list: 字典列表每个字典对应一个化合物的属性。失败的条目为None。 if properties is None: # 定义一组常用且可靠的化学信息学描述符 properties [MolecularFormula, MolecularWeight, CanonicalSMILES, XLogP, HydrogenBondDonorCount, HydrogenBondAcceptorCount, RotatableBondCount, TPSA, Complexity] all_results [] # 这里我们不再分批次因为get_properties函数内部可能已经处理了批量。 # 但为了绝对稳妥特别是CID列表很长时我们可以手动分小批。 batch_size 50 # 属性查询的批次可以小一些 total_cids len(cid_list) for i in range(0, total_cids, batch_size): batch_cids cid_list[i:ibatch_size] logging.info(f正在获取属性批次 {i//batch_size 1}/{(total_cids-1)//batch_size 1}CIDs: {batch_cids[:5]}...) try: # 核心调用批量获取属性 results pcp.get_properties(properties, batch_cids, as_dataframeFalse) # as_dataframeFalse 返回字典列表更方便我们与CID对应 all_results.extend(results) except pcp.NotFoundError: # 如果整个批次未找到可能性小添加None占位 logging.warning(f批次CID {batch_cids} 未找到属性。) all_results.extend([None] * len(batch_cids)) except pcp.BadRequestError as e: logging.error(f属性请求失败 (BadRequest) for batch {batch_cids}: {e}) all_results.extend([None] * len(batch_cids)) except Exception as e: logging.error(f属性请求未知错误 for batch {batch_cids}: {e}) all_results.extend([None] * len(batch_cids)) if i batch_size total_cids: time.sleep(delay) # 严格遵守延迟 return all_results注意事项属性名如‘XLogP’是大小写敏感的必须严格按照PubChem的文档来。as_dataframeFalse参数确保我们拿到的是Python字典列表方便后续与我们的名称-CID映射进行合并。如果设置为True会返回一个Pandas DataFrame但当某些化合物缺失某些属性时索引对齐可能会有点棘手。步骤4整合数据并保存现在我们有name_to_cid字典和property_list列表。需要将它们合并并处理可能存在的缺失或失败情况。import pandas as pd def compile_results(compound_names, name_to_cid, property_list): 将名称、CID和属性列表整合成一个Pandas DataFrame。 data [] for name in compound_names: cid name_to_cid.get(name) # 找到该名称在property_list中的索引 # 由于我们处理时顺序一致可以通过遍历或构建映射来关联。更稳健的方法是构建CID到属性的映射。 prop_dict None if cid is not None: # 在property_list中查找CID匹配的项 # 注意property_list中的每一项是一个字典其中包含CID键 for prop in property_list: if prop and prop.get(CID) cid: prop_dict prop break row {Compound_Name: name, CID: cid} if prop_dict: # 将属性字典中的键值对添加到行中 # 排除已经有的CID键避免重复 for key, value in prop_dict.items(): if key ! CID: row[key] value else: # 如果没有找到属性将所有属性列设为NaN # 这里我们需要知道有哪些属性列可以从第一个非None的属性字典获取keys if property_list and property_list[0] is not None: for key in property_list[0].keys(): if key ! CID: row[key] None data.append(row) df pd.DataFrame(data) # 调整列顺序让名称和CID在前 cols [Compound_Name, CID] [col for col in df.columns if col not in [Compound_Name, CID]] df df[cols] return df def save_to_csv(df, output_path): 保存DataFrame到CSV文件并处理可能的中文编码问题 df.to_csv(output_path, indexFalse, encodingutf-8-sig) # utf-8-sig确保Excel打开不乱码 logging.info(f结果已保存至: {output_path})3.3 主程序流程与异常处理框架将上述函数串联起来并加入完整的日志和异常处理就构成了我们的主程序。# run.py import logging import sys from config import INPUT_FILE, OUTPUT_FILE, REQUEST_DELAY from utils import load_compound_list, setup_logging from downloader import get_cids_in_batch, get_compound_properties, compile_results, save_to_csv def main(): # 1. 设置日志 setup_logging() logging.info(*50) logging.info(化合物信息批量下载程序启动) logging.info(*50) # 2. 加载化合物列表 try: compound_names load_compound_list(INPUT_FILE) if not compound_names: logging.error(输入文件为空或格式错误程序退出。) sys.exit(1) except FileNotFoundError: logging.error(f未找到输入文件: {INPUT_FILE}) sys.exit(1) except Exception as e: logging.error(f读取输入文件时发生错误: {e}) sys.exit(1) # 3. 批量获取CID logging.info(阶段1根据化合物名称查询CID...) name_to_cid get_cids_in_batch(compound_names, delayREQUEST_DELAY) # 提取所有成功找到的CID valid_cids [cid for cid in name_to_cid.values() if cid is not None] logging.info(f共有 {len(valid_cids)} 个有效CID用于查询属性。) if not valid_cids: logging.warning(没有找到任何有效的CID程序将只输出名称和空CID。) # 创建一个只有名称和CID的DataFrame df pd.DataFrame([{Compound_Name: name, CID: cid} for name, cid in name_to_cid.items()]) save_to_csv(df, OUTPUT_FILE) logging.info(程序结束无属性数据。) return # 4. 批量获取化合物属性 logging.info(阶段2根据CID批量获取化合物属性...) property_list get_compound_properties(valid_cids, delayREQUEST_DELAY) # 5. 整合数据 logging.info(阶段3整合数据...) df compile_results(compound_names, name_to_cid, property_list) # 6. 保存结果 save_to_csv(df, OUTPUT_FILE) # 7. 打印简要统计 success_count df[CID].notna().sum() prop_success_count df.dropna(subset[MolecularWeight]).shape[0] # 以某个关键属性为例 logging.info(f任务完成共处理 {len(compound_names)} 个化合物。) logging.info(f - 成功解析CID: {success_count} 个) logging.info(f - 成功获取属性: {prop_success_count} 个) logging.info(*50) if __name__ __main__: main()对应的config.py和utils.py示例# config.py INPUT_FILE ./input/compounds.txt OUTPUT_FILE ./output/compound_info.csv REQUEST_DELAY 0.3 # 秒保守起见设置比API限制稍高的延迟 LOG_FILE ./output/download.log # utils.py import logging def setup_logging(log_file./output/download.log): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) ] )4. 高级技巧与性能优化一个能跑的程序和一个好用的工具之间往往差的就是这些细节的打磨。4.1 处理模糊名称与同义词化合物名称常常不唯一。“Aspirin”也被称为“Acetylsalicylic acid”。PubChemPy的get_cids在名称匹配时默认使用“最接近”的匹配。但有时它会返回多个CID例如查询一个常见缩写时。get_cids的返回类型取决于参数list_returnflat: 返回列表每个输入名称对应一个CID如果有多个匹配可能是第一个。list_returndict: 返回字典键是输入名称值是该名称匹配到的所有CID的列表。策略对于关键任务建议先使用list_returndict检查是否有名称匹配到了多个CID然后人工或通过规则如选择第一个或选择分子量在某个范围的来确定一个。可以在get_cids_in_batch函数中增加这个逻辑。def get_cids_verbose(compound_names): 获取CID并记录匹配详情 result pcp.get_cids(compound_names, namespacename, list_returndict) ambiguous {} for name, cid_list in result.items(): if len(cid_list) 1: ambiguous[name] cid_list logging.warning(f化合物 {name} 匹配到多个CID: {cid_list}将使用第一个: {cid_list[0]}) # 即使多个我们也取第一个作为代表 name_to_cid[name] cid_list[0] if cid_list else None return name_to_cid, ambiguous4.2 异步请求与速率控制优化当处理成千上万个化合物时即使每批100个批次间的固定延迟如0.3秒也会导致总时间很长。更高级的策略是使用异步IO如asyncio和aiohttp来并发发送请求同时使用令牌桶Token Bucket等算法精确控制总体请求速率在不超过PubChem限制的前提下最大化效率。不过这引入了显著的复杂性。对于大多数实验室规模的批量任务几百到几千个简单的批次延迟已经足够可靠。我的建议是除非你经常需要处理超过5000个化合物的大列表否则优先选择简单稳定的同步方案。复杂性是万恶之源一个稳定运行但稍慢的脚本远胜于一个快但偶尔崩溃或漏数据的脚本。4.3 数据缓存与断点续传对于超大规模任务或者网络不稳定的环境实现缓存和断点续传是专业性的体现。缓存将已成功查询的{名称: CID}和{CID: 属性}以JSON格式保存到本地文件。下次运行时先加载缓存只查询未缓存的部分。断点续传将任务列表化合物名称本身也保存进度。程序启动时读取进度文件从中断处继续。这需要将主循环改造为更状态化的形式。这两个功能可以极大地提升使用体验特别是在调试阶段不用每次失败都从头开始。5. 常见问题排查与实战心得在实际运行中你肯定会遇到各种各样的问题。下面是我踩过的一些坑和解决方案。5.1 网络问题与超时处理PubChem服务器在国外国内访问有时不稳定。PubChemPy底层使用requests库我们可以通过修改默认会话Session来设置超时和重试。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_retry_session(retries3, backoff_factor0.5, status_forcelist(500, 502, 504)): 创建一个带重试机制的requests Session session requests.Session() retry Retry( totalretries, readretries, connectretries, backoff_factorbackoff_factor, status_forceliststatus_forcelist, ) adapter HTTPAdapter(max_retriesretry) session.mount(http://, adapter) session.mount(https://, adapter) return session # 在调用任何pcp函数之前设置自定义session pcp.session create_retry_session()5.2 化合物未找到或属性缺失这是最常见的问题。可能的原因和应对策略问题现象可能原因排查与解决思路CID查询返回None或01. 名称拼写错误或非标准。2. 该化合物不在PubChem中。3. 名称是商标名或俗名PubChem未收录。1. 检查输入文件确认名称准确。可尝试在PubChem网站手动搜索确认。2. 尝试使用化合物的SMILES或InChIKey查询如果已知。3. 记录下未找到的化合物后续手动处理。属性查询返回None或部分属性为None1. 该CID对应的记录确实缺失某些属性。2. 属性名拼写错误。3. 网络请求超时或中断。1. 在PubChem网站用该CID查看详情确认属性是否存在。2. 核对properties参数列表确保与PubChem属性名完全一致。3. 检查日志中的错误信息可能是网络问题导致整个批次失败。获取到的属性值异常如分子量为0数据在PubChem中本身如此可能是计算错误或特殊化合物。在后续数据分析步骤中加入数据清洗逻辑过滤掉明显异常的值。处理建议在compile_results函数中对于属性缺失的化合物不要直接丢弃整行而是保留行并用NaN填充缺失属性。这样在最终的数据分析阶段你可以清楚地知道哪些数据是完整的哪些需要补充或剔除。5.3 脚本运行速度太慢如果觉得脚本慢按以下顺序检查和优化检查延迟时间REQUEST_DELAY是主要瓶颈。可以尝试从0.3秒逐步降低到0.15秒观察是否会被限制。务必谨慎先用小规模列表测试。增大批次大小get_cids和get_properties的batch_size可以适当增加。对于CID查询尝试200或250。对于属性查询尝试100。注意单次请求的URL长度有限制约8000字符过大的批次会导致HTTP 414错误。并行化高级如前所述使用concurrent.futures或asyncio实现并发。关键是要控制总并发数例如使用一个固定大小的线程池如5个线程并在每个线程内部保持延迟。这比简单的批次延迟复杂得多但能显著提升速度。使用本地缓存对于重复查询的化合物缓存是终极提速方案。5.4 输出文件在Excel中打开乱码这是编码问题。确保使用df.to_csv(..., encodingutf-8-sig)。utf-8-sig会在文件开头添加一个BOM字节顺序标记帮助Excel正确识别UTF-8编码。5.5 内存占用过高处理极大量数据如十万级以上时一次性将所有结果保存在内存的列表中可能会导致内存不足。解决方案是使用流式处理或分块处理流式写入每处理完一批化合物比如1000个就立即将这一批的结果追加写入到CSV文件中然后释放这部分内存。使用数据库对于海量数据考虑使用SQLite或磁盘上的HDF5存储而不是全部放在Pandas DataFrame里。最后分享一个我个人的小习惯在脚本的output目录下除了最终的compound_info.csv我总会同时保存一个failed_compounds.txt里面记录所有未能成功获取CID或属性的化合物名称。这样当主要任务完成后我可以集中精力手动处理这一小部分“疑难杂症”而不是淹没在成千上万的成功记录里找不到它们。这个简单的习惯多次拯救了我让我能快速定位和解决最后那5%的问题。
![微信抢红包终极指南：如何用免费插件轻松抢到每个红包！[特殊字符]](http://pic.xiahunao.cn/yaotu/微信抢红包终极指南：如何用免费插件轻松抢到每个红包！[特殊字符])
