5. 保存清洗脚本
全部平台
保存为 clean_chemical_data.py,检查输出字段,只在明确需要时启用 --pubchem。
"""Preserve original CSV rows while adding RDKit identifiers and optional CIDs.
Sources: https://www.rdkit.org/docs/GettingStartedInPython.html
https://docs.pubchempy.org/en/latest/guide/searching.html
Source-reviewed example; not executed. No salt removal or charge normalization.
"""
import argparse
import csv
import json
import time
from pathlib import Path
from rdkit import Chem, rdBase
def main():
parser = argparse.ArgumentParser()
parser.add_argument("input", type=Path)
parser.add_argument("output", type=Path)
parser.add_argument("--pubchem", action="store_true")
args = parser.parse_args()
added = ["source_row", "canonical_isomeric_smiles", "validation_status",
"duplicate_of_row", "rdkit_version", "pubchem_cids", "lookup_status", "lookup_error"]
with args.input.open(encoding="utf-8-sig", newline="") as source:
reader = csv.DictReader(source)
fields = reader.fieldnames or []
if "smiles" not in fields or len(fields) != len(set(fields)) or set(fields).intersection(added):
parser.error("Require a unique smiles column and no collisions with output columns.")
seen = {}
lookup_cache = {}
# Exclusive creation prevents overwriting either source or existing output.
with args.output.open("x", encoding="utf-8", newline="") as target:
writer = csv.DictWriter(target, fieldnames=fields + added)
writer.writeheader()
for row_number, row in enumerate(reader, start=2):
if None in row:
parser.error("Malformed CSV row; partial output was retained for inspection.")
result = dict(row, source_row=row_number, canonical_isomeric_smiles="",
validation_status="", duplicate_of_row="", rdkit_version=rdBase.rdkitVersion,
pubchem_cids="[]", lookup_status="not_requested", lookup_error="")
smiles = (row.get("smiles") or "").strip()
molecule = Chem.MolFromSmiles(smiles) if smiles else None
if molecule is None:
result["validation_status"] = "empty" if not smiles else "invalid"
else:
canonical = Chem.MolToSmiles(molecule, canonical=True, isomericSmiles=True)
result["canonical_isomeric_smiles"] = canonical
result["validation_status"] = "valid"
if canonical in seen:
result["duplicate_of_row"] = seen[canonical]
else:
seen[canonical] = row_number
if args.pubchem:
if canonical not in lookup_cache:
import pubchempy as pcp
try:
compounds = pcp.get_compounds(canonical, "smiles")
cids = sorted({item.cid for item in compounds if item.cid is not None})
lookup_cache[canonical] = (json.dumps(cids), "matched" if cids else "not_found", "")
except (pcp.PubChemPyError, OSError, ValueError) as error:
lookup_cache[canonical] = ("[]", "error", str(error))
time.sleep(0.3)
result["pubchem_cids"], result["lookup_status"], result["lookup_error"] = lookup_cache[canonical]
writer.writerow(result)
if __name__ == "__main__":
main()
官方来源预期结果
脚本与 .venv 位于同一目录,输入行尚未修改。
6. 保存验收样例
全部平台
保存为 input.csv,包含重复、非法 SMILES、空值与盐结构。
id,smiles
ethanol-a,CCO
ethanol-b,OCC
invalid,not-a-smiles
empty,
salt,CC(=O)[O-].[Na+]
官方来源7. 运行本地清洗
全部平台
Windows 将解释器替换为 ..venv\Scripts\python.exe。本地确定性验收先不添加 --pubchem。
.venv/bin/python clean_chemical_data.py input.csv cleaned.csv
官方来源预期结果
保留五行:三行合法、一行非法、一行空值;ethanol-b 的重复指向源行 2,盐仍保留点号分隔的钠阳离子与乙酸根;lookup_status 均为 not_requested,输入文件未改动。
9. 可选执行 PubChem 补全
全部平台
使用不同输出文件名,记录检索时间并查看 lookup_status/lookup_error;保留多个 CID,不对机密数据调用外部服务。
.venv/bin/python clean_chemical_data.py input.csv enriched.csv --pubchem
官方来源预期结果
原行数不变,合法行得到实际 CID 列表或明确查询错误。