DrugBank数据解析实战从XML到结构化CSV的Python高效处理指南如果你正在从事药物发现、生物信息学或计算化学相关的研究那么DrugBank数据库绝对是你绕不开的核心资源。这个集成了药物化学、药理学、靶点信息于一体的综合性知识库为科研人员提供了超过13000种药物的详细数据。但拿到那个庞大的XML文件后很多研究者都会面临一个现实问题如何高效地从中提取自己真正需要的信息我刚开始接触DrugBank数据时也曾在XML的复杂嵌套结构中迷失方向。官方提供的完整数据库文件通常超过100MB直接打开查看几乎不可能更不用说从中提取特定字段了。经过多次实践和优化我总结出了一套完整的Python处理流程不仅能自动提取药物-靶点关系还能生成带完整表头的CSV文件更重要的是我发现了比传统方法更高效的解析策略。1. 环境准备与数据获取1.1 DrugBank数据申请与下载首先需要明确的是DrugBank数据库对学术研究是免费开放的但需要注册并提交使用申请。这个过程其实比想象中简单访问DrugBank官网注册账号在下载页面选择Full Database的XML格式填写简单的使用目的说明通常只需说明用于非商业学术研究等待1-2个工作日即可获得下载链接注意确保在申请时明确说明研究目的并承诺不将数据用于商业用途或分享给第三方。学术用途的申请通过率很高不必担心被拒绝。下载完成后你会得到一个名为full_database.xml的文件文件大小通常在100-200MB之间。我建议在首次处理前先创建一个测试用的简化版本# 创建测试文件前1000行 with open(full_database.xml, r, encodingutf-8) as f: head [next(f) for _ in range(1000)] with open(test_sample.xml, w, encodingutf-8) as f: f.writelines(head)这样可以在小文件上测试代码避免在大文件上反复调试的漫长等待。1.2 Python环境配置对于DrugBank XML解析我推荐使用Anaconda创建独立环境避免包冲突# 创建新环境 conda create -n drugbank_analysis python3.9 conda activate drugbank_analysis # 安装核心依赖 pip install lxml pandas numpy这里我特别推荐使用lxml而不是xmltodict原因稍后会详细解释。如果你已经安装了其他XML解析库不用担心我们可以同时安装多个然后对比性能# 安装备选方案 pip install xmltodict beautifulsoup41.3 PyCharm配置建议在PyCharm中处理大文件时有几个实用技巧增加内存限制在Help→Edit Custom VM Options中添加-Xmx4096m这将Java堆内存增加到4GB避免处理大文件时崩溃。文件编码设置确保所有文件都使用UTF-8编码。在File→Settings→Editor→File Encodings中将Global Encoding、Project Encoding和Default encoding for properties files都设为UTF-8。使用科学模式对于数据探索可以启用PyCharm的科学模式Scientific Mode方便查看DataFrame和绘图。2. XML解析库深度对比lxml vs xmltodict2.1 性能基准测试在处理DrugBank这样的大型XML文件时解析库的选择直接影响处理效率。我做了详细的对比测试结果如下特性lxmlxmltodict内置xml.etree解析速度100MB文件12-15秒45-60秒25-35秒内存占用中等高低XPath支持完整支持不支持有限支持易用性中等高低处理嵌套结构优秀良好一般错误恢复能力强弱中等从实际使用体验来看lxml在速度和功能上都有明显优势。虽然学习曲线稍陡但一旦掌握处理效率会大幅提升。2.2 lxml实战解析让我们看看如何使用lxml高效解析DrugBank XMLfrom lxml import etree import pandas as pd from tqdm import tqdm import time def parse_drugbank_lxml(xml_path): 使用lxml解析DrugBank XML文件 start_time time.time() # 使用迭代解析避免一次性加载整个文件 context etree.iterparse(xml_path, events(end,), tag{http://www.drugbank.ca}drug) drugs_data [] for event, elem in tqdm(context, desc解析药物数据): try: drug_info extract_drug_info(elem) if drug_info: drugs_data.append(drug_info) # 清理已处理的元素释放内存 elem.clear() while elem.getprevious() is not None: del elem.getparent()[0] except Exception as e: print(f解析错误: {e}) continue print(f解析完成耗时: {time.time() - start_time:.2f}秒) return drugs_data def extract_drug_info(drug_elem): 提取单个药物的关键信息 # 获取DrugBank ID可能有多个取第一个 drugbank_ids drug_elem.findall({http://www.drugbank.ca}drugbank-id) primary_id None for id_elem in drugbank_ids: if id_elem.get(primary) true: primary_id id_elem.text break if not primary_id and drugbank_ids: primary_id drugbank_ids[0].text # 药物名称 name_elem drug_elem.find({http://www.drugbank.ca}name) drug_name name_elem.text if name_elem is not None else # 药物类型 drug_type drug_elem.get(type, ) # 适应症信息 indications [] indication_elem drug_elem.find({http://www.drugbank.ca}indication) if indication_elem is not None: indications.append(indication_elem.text) # 靶点信息 targets_data [] targets_elem drug_elem.find({http://www.drugbank.ca}targets) if targets_elem is not None: target_list targets_elem.findall({http://www.drugbank.ca}target) for target in target_list: target_info extract_target_info(target) if target_info: targets_data.append(target_info) return { drugbank_id: primary_id, name: drug_name, type: drug_type, indications: ;.join(indications) if indications else , targets: targets_data }这种迭代解析的方式特别适合处理大型XML文件因为它不会一次性将整个文件加载到内存中。2.3 xmltodict的适用场景虽然xmltodict在性能上不如lxml但它有一个独特的优势将XML直接转换为Python字典代码更直观。对于小型文件或快速原型开发它仍然是个不错的选择import xmltodict import json def parse_with_xmltodict(xml_path): 使用xmltodict解析适合小文件或测试 with open(xml_path, r, encodingutf-8) as f: xml_content f.read() # 转换为字典 data_dict xmltodict.parse(xml_content) # 可以保存为JSON便于查看结构 with open(drugbank_structure.json, w, encodingutf-8) as f: json.dump(data_dict, f, indent2, ensure_asciiFalse) return data_dict提示对于超过50MB的XML文件不建议使用xmltodict因为内存消耗会非常大。我曾在处理完整DrugBank文件时遇到过内存不足的问题最终不得不切换到lxml。3. 构建带表头的完整CSV导出系统3.1 智能表头生成策略很多现有的解析脚本生成的CSV没有表头这给后续的数据分析带来了不便。我设计了一个智能的表头生成系统不仅能自动添加表头还能根据提取的字段动态调整import csv from typing import List, Dict, Any class DrugBankCSVExporter: def __init__(self, output_pathdrugbank_export.csv): self.output_path output_path self.fieldnames [ DrugBank_ID, Drug_Name, Drug_Type, Indications, Target_UniProt_ID, Target_Name, Target_Organism, Target_Actions, Known_Action, PDB_ID, Gene_Name, GO_Classification ] def create_header(self, additional_fieldsNone): 创建CSV文件头 if additional_fields: self.fieldnames.extend(additional_fields) # 去重并排序 self.fieldnames sorted(list(set(self.fieldnames))) return self.fieldnames def detect_missing_headers(self, data_sample): 检测数据中可能缺失的字段 existing_fields set() missing_suggestions [] for drug in data_sample: for key in drug.keys(): existing_fields.add(key) # 检查靶点信息的嵌套字段 if targets in drug and drug[targets]: for target in drug[targets]: for key in target.keys(): existing_fields.add(fTarget_{key}) # 找出建议添加的字段 suggested_fields [ATC_Code, CAS_Number, Molecular_Weight, SMILES, InChI_Key, Drug_Group, Pathways] for field in suggested_fields: if field not in existing_fields: missing_suggestions.append(field) return missing_suggestions3.2 完整的数据提取与导出流程下面是一个完整的处理流程包含错误处理和进度显示def extract_detailed_drug_info(drug_elem): 提取药物的详细信息 info {} # 基础信息 info[DrugBank_ID] extract_primary_id(drug_elem) info[Drug_Name] extract_text(drug_elem, name) info[Drug_Type] drug_elem.get(type, ) # 描述信息 info[Description] extract_text(drug_elem, description) info[Indication] extract_text(drug_elem, indication) info[Pharmacology] extract_text(drug_elem, pharmacology) # 化学信息 info[CAS_Number] extract_text(drug_elem, cas-number) info[UNII] extract_text(drug_elem, unii) # 分子属性 properties drug_elem.find({http://www.drugbank.ca}calculated-properties) if properties is not None: for prop in properties.findall({http://www.drugbank.ca}property): kind extract_text(prop, kind) value extract_text(prop, value) if kind and value: info[fProperty_{kind}] value # 分类信息 info[Groups] extract_groups(drug_elem) info[Categories] extract_categories(drug_elem) info[ATC_Codes] extract_atc_codes(drug_elem) return info def extract_target_details(target_elem): 提取靶点的详细信息 target_info {} # 基础信息 target_info[Target_ID] extract_text(target_elem, id) target_info[Target_Name] extract_text(target_elem, name) target_info[Organism] extract_text(target_elem, organism) # 多肽信息 polypeptide target_elem.find({http://www.drugbank.ca}polypeptide) if polypeptide is not None: target_info[UniProt_ID] polypeptide.get(id, ) target_info[Gene_Name] extract_text(polypeptide, gene-name) target_info[Gene_Sequence] extract_text(polypeptide, gene-sequence) # GO分类 go_classifications [] go_class polypeptide.find({http://www.drugbank.ca}go-classifiers) if go_class is not None: for classifier in go_class.findall({http://www.drugbank.ca}go-classifier): category extract_text(classifier, category) description extract_text(classifier, description) if category and description: go_classifications.append(f{category}: {description}) target_info[GO_Classification] ; .join(go_classifications) # 作用信息 target_info[Known_Action] extract_text(target_elem, known-action) # 参考文献 references [] refs target_elem.find({http://www.drugbank.ca}references) if refs is not None: for article in refs.findall({http://www.drugbank.ca}articles/{http://www.drugbank.ca}article): pubmed_id extract_text(article, pubmed-id) if pubmed_id: references.append(pubmed_id) target_info[PubMed_References] ; .join(references) return target_info3.3 批量处理与增量导出对于大型数据集我建议使用分批处理和增量导出def batch_process_drugbank(xml_path, batch_size100, output_templatedrugbank_batch_{}.csv): 分批处理DrugBank数据 context etree.iterparse(xml_path, events(end,), tag{http://www.drugbank.ca}drug) batch_data [] batch_num 1 for event, elem in tqdm(context, desc分批处理): try: drug_info extract_detailed_drug_info(elem) batch_data.append(drug_info) # 达到批次大小时导出 if len(batch_data) batch_size: export_batch(batch_data, output_template.format(batch_num)) batch_data [] batch_num 1 # 清理内存 elem.clear() while elem.getprevious() is not None: del elem.getparent()[0] except Exception as e: print(f处理错误: {e}) continue # 导出最后一批 if batch_data: export_batch(batch_data, output_template.format(batch_num)) print(f分批处理完成共{batch_num}个批次) def export_batch(batch_data, output_file): 导出单个批次数据 if not batch_data: return # 动态确定所有字段 all_fields set() for drug in batch_data: all_fields.update(drug.keys()) # 排序字段 field_order [DrugBank_ID, Drug_Name, Drug_Type, Description, Indication, CAS_Number, Groups, Categories] other_fields sorted([f for f in all_fields if f not in field_order]) ordered_fields field_order other_fields # 写入CSV with open(output_file, w, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnamesordered_fields) writer.writeheader() writer.writerows(batch_data) print(f批次已导出到: {output_file})4. 按疾病筛选的精准提取技巧4.1 基于适应症的智能筛选DrugBank中的indication字段包含了药物的主要适应症信息我们可以利用这个字段快速筛选特定疾病的药物def extract_drugs_by_indication(xml_path, target_diseases, output_filefiltered_drugs.csv): 根据适应症筛选药物 target_diseases_lower [d.lower() for d in target_diseases] context etree.iterparse(xml_path, events(end,), tag{http://www.drugbank.ca}drug) filtered_drugs [] for event, elem in tqdm(context, desc按适应症筛选): try: # 提取适应症信息 indication_elem elem.find({http://www.drugbank.ca}indication) if indication_elem is not None and indication_elem.text: indication_text indication_elem.text.lower() # 检查是否包含目标疾病关键词 matches_disease any(disease in indication_text for disease in target_diseases_lower) if matches_disease: drug_info extract_detailed_drug_info(elem) # 添加匹配的疾病信息 drug_info[Matched_Disease] ; .join([ d for d in target_diseases if d.lower() in indication_text ]) drug_info[Indication_Text] indication_elem.text[:500] # 截取前500字符 filtered_drugs.append(drug_info) elem.clear() while elem.getprevious() is not None: del elem.getparent()[0] except Exception as e: print(f筛选错误: {e}) continue # 导出筛选结果 if filtered_drugs: export_to_csv(filtered_drugs, output_file) print(f找到 {len(filtered_drugs)} 个相关药物已导出到 {output_file}) else: print(未找到匹配的药物) return filtered_drugs # 使用示例提取糖尿病相关药物 diabetes_keywords [diabetes, diabetic, type 2 diabetes, type 1 diabetes, hyperglycemia, insulin resistance] diabetes_drugs extract_drugs_by_indication(full_database.xml, diabetes_keywords, diabetes_drugs.csv)4.2 多条件组合筛选在实际研究中我们往往需要更复杂的筛选条件。下面是一个支持多条件组合筛选的增强版本class DrugFilter: def __init__(self): self.filters [] def add_indication_filter(self, keywords, match_typeany): 添加适应症筛选条件 def filter_func(drug_info): indication drug_info.get(Indication, ).lower() if match_type any: return any(keyword.lower() in indication for keyword in keywords) elif match_type all: return all(keyword.lower() in indication for keyword in keywords) return False self.filters.append((indication, filter_func)) return self def add_drug_type_filter(self, drug_types): 添加药物类型筛选 def filter_func(drug_info): return drug_info.get(Drug_Type, ) in drug_types self.filters.append((type, filter_func)) return self def add_group_filter(self, groups): 添加药物分组筛选如approved, experimental等 def filter_func(drug_info): drug_groups drug_info.get(Groups, ).split(; ) return any(group in drug_groups for group in groups) self.filters.append((group, filter_func)) return self def add_target_filter(self, target_keywords): 添加靶点筛选条件 def filter_func(drug_info): targets_text .join([ t.get(Target_Name, ) t.get(Gene_Name, ) for t in drug_info.get(targets, []) ]).lower() return any(keyword.lower() in targets_text for keyword in target_keywords) self.filters.append((target, filter_func)) return self def apply_filters(self, drug_info): 应用所有筛选条件 for filter_name, filter_func in self.filters: if not filter_func(drug_info): return False return True # 使用示例筛选已批准的糖尿病药物且作用于特定靶点 drug_filter DrugFilter() drug_filter.add_indication_filter([diabetes, hyperglycemia]) drug_filter.add_group_filter([approved, investigational]) drug_filter.add_target_filter([insulin receptor, GLUT4, PPAR]) # 在解析过程中应用筛选 filtered_drugs [] for drug in parse_drugbank_lxml(full_database.xml): if drug_filter.apply_filters(drug): filtered_drugs.append(drug)4.3 靶点信息深度提取对于筛选出的药物我们通常需要详细的靶点信息。下面的函数可以提取完整的药物-靶点关系网络def extract_drug_target_network(xml_path, output_filesNone): 提取完整的药物-靶点关系网络 if output_files is None: output_files { drugs: drugs_detailed.csv, targets: targets_detailed.csv, interactions: drug_target_interactions.csv } drugs_data [] targets_data [] interactions [] target_id_map {} # 用于去重靶点 context etree.iterparse(xml_path, events(end,), tag{http://www.drugbank.ca}drug) for event, elem in tqdm(context, desc构建药物-靶点网络): try: drug_info extract_basic_drug_info(elem) drug_id drug_info[DrugBank_ID] # 提取靶点信息 targets_elem elem.find({http://www.drugbank.ca}targets) if targets_elem is not None: target_list targets_elem.findall({http://www.drugbank.ca}target) for target in target_list: target_info extract_complete_target_info(target) target_uniprot target_info.get(UniProt_ID) if target_uniprot: # 记录靶点去重 if target_uniprot not in target_id_map: target_id_map[target_uniprot] target_info targets_data.append(target_info) # 记录相互作用 interaction { DrugBank_ID: drug_id, Drug_Name: drug_info[Drug_Name], UniProt_ID: target_uniprot, Target_Name: target_info.get(Target_Name, ), Known_Action: target_info.get(Known_Action, ), Action_Type: extract_text(target, action), References: target_info.get(PubMed_References, ) } interactions.append(interaction) drugs_data.append(drug_info) elem.clear() while elem.getprevious() is not None: del elem.getparent()[0] except Exception as e: print(f网络构建错误: {e}) continue # 导出三个关联文件 export_dataframe(drugs_data, output_files[drugs]) export_dataframe(targets_data, output_files[targets]) export_dataframe(interactions, output_files[interactions]) print(f网络构建完成: {len(drugs_data)}个药物, {len(targets_data)}个靶点, {len(interactions)}个相互作用) return drugs_data, targets_data, interactions def export_dataframe(data, filename): 将数据导出为CSV if not data: print(f警告: {filename} 没有数据可导出) return df pd.DataFrame(data) # 处理嵌套结构 for col in df.columns: if df[col].apply(lambda x: isinstance(x, (list, dict))).any(): df[col] df[col].apply(lambda x: str(x) if isinstance(x, (list, dict)) else x) df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f数据已导出到: {filename} ({len(df)}行))5. 高级技巧与性能优化5.1 内存优化策略处理大型XML文件时内存管理至关重要。以下是我在实践中总结的优化技巧class MemoryEfficientXMLParser: def __init__(self, xml_path, chunk_size50): self.xml_path xml_path self.chunk_size chunk_size def parse_in_chunks(self): 分块解析XML减少内存占用 context etree.iterparse(self.xml_path, events(end,), tag{http://www.drugbank.ca}drug) chunk [] chunk_count 0 for event, elem in context: try: drug_data self.process_drug_element(elem) if drug_data: chunk.append(drug_data) # 定期清理内存 elem.clear() for ancestor in elem.xpath(ancestor-or-self::*): while ancestor.getprevious() is not None: del ancestor.getparent()[0] # 达到块大小时处理并清空 if len(chunk) self.chunk_size: yield chunk chunk [] chunk_count 1 # 每处理10个块进行一次垃圾回收 if chunk_count % 10 0: import gc gc.collect() except Exception as e: print(f处理错误: {e}) continue # 处理最后一块 if chunk: yield chunk def process_drug_element(self, elem): 处理单个药物元素只提取必要字段 # 使用最小化的数据结构 drug_info { id: self.extract_text_safe(elem, drugbank-id[primarytrue]), name: self.extract_text_safe(elem, name), type: elem.get(type, ), groups: self.extract_groups_simple(elem) } # 只提取关键靶点信息 targets [] targets_elem elem.find(targets) if targets_elem is not None: for target in targets_elem.findall(target)[:5]: # 限制每个药物的靶点数量 target_info { id: self.extract_text_safe(target, id), name: self.extract_text_safe(target, name), uniprot: self.extract_text_safe(target, polypeptide/id) } if target_info[uniprot]: targets.append(target_info) if targets: drug_info[targets] targets return drug_info def extract_text_safe(self, elem, xpath): 安全的文本提取 try: result elem.xpath(xpath) if result: return result[0].text if hasattr(result[0], text) else str(result[0]) except: pass return 5.2 并行处理加速对于多核CPU我们可以使用并行处理来加速解析过程from concurrent.futures import ProcessPoolExecutor, as_completed import multiprocessing as mp def parallel_parse_drugbank(xml_path, num_processesNone): 并行解析DrugBank XML if num_processes is None: num_processes mp.cpu_count() - 1 # 留一个核心给系统 # 第一步分割文件如果文件太大 chunk_files split_xml_by_drug_count(xml_path, drugs_per_chunk1000) # 第二步并行处理每个分块 all_results [] with ProcessPoolExecutor(max_workersnum_processes) as executor: future_to_chunk { executor.submit(parse_xml_chunk, chunk_file): chunk_file for chunk_file in chunk_files } for future in tqdm(as_completed(future_to_chunk), totallen(chunk_files), desc并行处理): chunk_file future_to_chunk[future] try: result future.result() all_results.extend(result) print(f完成处理: {chunk_file}) except Exception as e: print(f处理 {chunk_file} 时出错: {e}) # 第三步合并结果 combined_df pd.concat(all_results, ignore_indexTrue) return combined_df def split_xml_by_drug_count(xml_path, drugs_per_chunk1000): 将大XML文件按药物数量分割成多个小文件 # 这里需要实现XML分割逻辑 # 由于XML结构复杂实际实现需要考虑完整的标签结构 # 简化版使用命令行工具分割 import subprocess # 使用xml_split工具需要安装 output_dir xml_chunks os.makedirs(output_dir, exist_okTrue) # 注意这只是一个示意实际需要根据XML结构调整 command fxml_split -c {drugs_per_chunk} -l 2 {xml_path} {output_dir}/chunk subprocess.run(command, shellTrue, checkTrue) # 返回生成的文件列表 chunk_files [os.path.join(output_dir, f) for f in os.listdir(output_dir) if f.startswith(chunk) and f.endswith(.xml)] return sorted(chunk_files)5.3 错误处理与数据验证在实际处理中数据质量问题很常见。下面是一个健壮的数据验证和清洗流程class DataValidator: def __init__(self): self.validation_rules { DrugBank_ID: self.validate_drugbank_id, Drug_Name: self.validate_non_empty, UniProt_ID: self.validate_uniprot_id, CAS_Number: self.validate_cas_number, Molecular_Weight: self.validate_numeric } self.error_log [] def validate_drugbank_id(self, value): 验证DrugBank ID格式 if not value: return False, DrugBank ID为空 # DrugBank ID格式DB后跟5位数字 import re pattern r^DB\d{5}$ if not re.match(pattern, value): return False, f无效的DrugBank ID格式: {value} return True, def validate_uniprot_id(self, value): 验证UniProt ID格式 if not value: return True, # 允许为空 # UniProt ID格式检查 import re # 基本格式字母数字组合通常以P、Q、O等开头 if not re.match(r^[A-N,R-Z][0-9][A-Z][A-Z,0-9][A-Z,0-9][0-9]$|^[A-N,R-Z][0-9][A-Z][A-Z,0-9][A-Z,0-9][0-9][A-Z]$, value): return False, f可疑的UniProt ID格式: {value} return True, def validate_cas_number(self, value): 验证CAS号格式 if not value: return True, # CAS号格式XXX-XX-X import re pattern r^\d{1,7}-\d{2}-\d$ if not re.match(pattern, value): return False, f无效的CAS号格式: {value} # 验证校验位 parts value.split(-) if len(parts) ! 3: return False, fCAS号格式错误: {value} # 这里可以添加更详细的校验位验证 return True, def validate_batch(self, data_batch): 批量验证数据 valid_records [] for i, record in enumerate(data_batch): is_valid True errors [] for field, validator in self.validation_rules.items(): if field in record: value record[field] valid, error_msg validator(value) if not valid: is_valid False errors.append(f{field}: {error_msg}) if is_valid: valid_records.append(record) else: self.error_log.append({ record_index: i, record_id: record.get(DrugBank_ID, Unknown), errors: errors, data: record }) return valid_records def generate_validation_report(self, output_filevalidation_report.txt): 生成验证报告 with open(output_file, w, encodingutf-8) as f: f.write(数据验证报告\n) f.write( * 50 \n\n) f.write(f总错误数: {len(self.error_log)}\n\n) for error in self.error_log: f.write(f记录 {error[record_index]} (ID: {error[record_id]}):\n) for err in error[errors]: f.write(f - {err}\n) f.write(\n) print(f验证报告已生成: {output_file}) return self.error_log5.4 数据质量检查与修复在导出数据前进行全面的质量检查def perform_data_quality_check(dataframe): 执行数据质量检查 quality_report { total_records: len(dataframe), missing_values: {}, unique_counts: {}, data_types: {}, potential_issues: [] } # 检查缺失值 for column in dataframe.columns: missing_count dataframe[column].isna().sum() if missing_count 0: quality_report[missing_values][column] { count: missing_count, percentage: (missing_count / len(dataframe)) * 100 } # 检查唯一值数量 for column in [DrugBank_ID, Drug_Name, UniProt_ID]: if column in dataframe.columns: unique_count dataframe[column].nunique() quality_report[unique_counts][column] unique_count # 检查重复的DrugBank ID if column DrugBank_ID: duplicates dataframe[dataframe.duplicated(subset[column], keepFalse)] if len(duplicates) 0: quality_report[potential_issues].append( f发现 {len(duplicates)} 个重复的DrugBank ID ) # 检查数据类型 for column in dataframe.columns: dtype str(dataframe[column].dtype) sample_value dataframe[column].iloc[0] if len(dataframe) 0 else None quality_report[data_types][column] { dtype: dtype, sample: str(sample_value)[:100] # 截断长样本 } # 检查特定字段的格式 if DrugBank_ID in dataframe.columns: invalid_ids dataframe[~dataframe[DrugBank_ID].str.match(r^DB\d{5}$, naFalse)] if len(invalid_ids) 0: quality_report[potential_issues].append( f发现 {len(invalid_ids)} 个无效的DrugBank ID格式 ) return quality_report def generate_quality_report_html(report, output_filequality_report.html): 生成HTML格式的质量报告 html_content !DOCTYPE html html head titleDrugBank数据质量报告/title style body { font-family: Arial, sans-serif; margin: 40px; } h1 { color: #333; } h2 { color: #666; margin-top: 30px; } table { border-collapse: collapse; width: 100%; margin: 20px 0; } th, td { border: 1px solid #ddd; padding: 12px; text-align: left; } th { background-color: #f2f2f2; } .warning { background-color: #fff3cd; border-color: #ffeaa7; } .error { background-color: #f8d7da; border-color: #f5c6cb; } .success { background-color: #d4edda; border-color: #c3e6cb; } /style /head body h1DrugBank数据质量分析报告/h1 p生成时间: {timestamp}/p h2数据概览/h2 table trth指标/thth值/th/tr trtd总记录数/tdtd{total_records}/td/tr trtd总列数/tdtd{total_columns}/td/tr /table h2缺失值分析/h2 table trth字段/thth缺失数量/thth缺失比例/th/tr {missing_rows} /table h2唯一值统计/h2 table trth字段/thth唯一值数量/th/tr {unique_rows} /table h2潜在问题/h2 ul {issue_list} /ul /body /html from datetime import datetime # 构建表格行 missing_rows for field, info in report[missing_values].items(): missing_rows ftrtd{field}/tdtd{info[count]}/tdtd{info[percentage]:.2f}%/td/tr\n unique_rows for field, count in report[unique_counts].items(): unique_rows ftrtd{field}/tdtd{count}/td/tr\n issue_list for issue in report[potential_issues]: issue_list fli{issue}/li\n # 填充模板 html_content html_content.format( timestampdatetime.now().strftime(%Y-%m-%d %H:%M:%S), total_recordsreport[total_records], total_columnslen(report[data_types]), missing_rowsmissing_rows, unique_rowsunique_rows, issue_listissue_list ) with open(output_file, w, encodingutf-8) as f: f.write(html_content) print(f质量报告已生成: {output_file}) return output_file这套完整的DrugBank数据处理方案在实际项目中经过了多次迭代优化。从最初的简单解析到现在的生产级数据处理流程最大的体会是处理好边缘情况和数据质量比实现核心功能更重要。每次解析大型数据集时总会遇到一些意料之外的数据格式问题完善的错误处理和验证机制能节省大量调试时间。对于刚开始接触DrugBank数据的研究者我的建议是先从小的测试文件开始确保解析逻辑正确后再处理完整数据集同时一定要保存中间结果避免因为一个小错误而重新解析整个文件。XML解析虽然技术门槛不高但细节决定成败特别是在处理真实世界数据时。