📚 课程目录第4章数据采集与爬取工程

4. 数据采集与爬取工程

ADMET数据集构建的第一道工程关口是“把散落在不同来源的化合物—活性—ADMET属性数据稳定地取回来”。这一步的难点不在单次请求，而在于多源异构、接口差异、限流约束、断点续传与增量识别。本节围绕四类采集通道展开：结构化API、批量文件下载、网页抓取、文献补充材料提取，并给出限流与增量更新的工程实现方式。

一、四类采集通道的定位与选择

不同来源的数据形态决定了采集方式。结构化API适合ChEMBL、PubChem、DrugBank等提供REST接口的数据库，返回JSON/XML，字段稳定、可分页；批量文件下载适合ChEMBL的SQLite/MySQL dump、PubChem的FTP SDF、TDC的CSV包，一次获取全量、避免逐条请求；网页抓取适合无API但页面结构稳定的数据源，如部分药企公开的ADMET表格、监管机构页面；文献补充材料提取则针对PDF、XLSX、DOCX附件中的表格，往往包含未进入主库的ADMET实验值。

选择原则是：能用API就不用爬虫，能批量下载就不逐条调用，能解析结构化附件就不做OCR。采集通道的优先级直接影响后续治理成本。

二、API调用的工程实现

以ChEMBL为例，其REST接口支持按靶点、按化合物、按活性分页获取。工程实现的关键是封装统一的请求函数，处理分页游标、超时、重试与响应校验。下面是一个可复用的Python请求封装示例：

import requests, time, json, hashlib, os

BASE = "https://www.ebi.ac.uk/chembl/api/data"
SESSION = requests.Session()
SESSION.headers.update({"Accept": "application/json"})

def fetch_page(endpoint, params, max_retry=5):
    url = f"{BASE}/{endpoint}"
    for attempt in range(max_retry):
        try:
            r = SESSION.get(url, params=params, timeout=30)
            if r.status_code == 200:
                return r.json()
            if r.status_code in (429, 503):
                wait = min(2 ** attempt, 60)
                time.sleep(wait)
                continue
            r.raise_for_status()
        except requests.RequestException:
            time.sleep(min(2 ** attempt, 60))
    raise RuntimeError(f"failed: {endpoint} {params}")

def iter_activities(target_id, page_size=1000):
    offset = 0
    while True:
        data = fetch_page("activity", {
            "target_chembl_id": target_id,
            "limit": page_size,
            "offset": offset
        })
        records = data.get("activities", [])
        if not records:
            break
        for rec in records:
            yield rec
        offset += page_size
        time.sleep(0.2)  # 主动限速，避免触发429

这段代码体现了三个工程要点：一是用Session复用连接；二是对429/503做指数退避，等待时间随重试次数翻倍并封顶60秒；三是每页之间主动sleep 0.2秒，把请求速率控制在服务端容忍范围内。分页游标用offset递增，直到返回空列表为止。

三、批量下载与断点续传

批量文件往往体积较大，如ChEMBL的SQLite dump约数GB。直接下载容易因网络中断而前功尽弃，因此需要支持断点续传。实现方式是记录已下载字节数，用HTTP Range头续传：

def download_with_resume(url, dest, chunk=1 << 20):
    pos = os.path.getsize(dest) if os.path.exists(dest) else 0
    headers = {"Range": f"bytes={pos}-"} if pos else {}
    with requests.get(url, headers=headers, stream=True, timeout=60) as r:
        r.raise_for_status()
        mode = "ab" if pos else "wb"
        with open(dest, mode) as f:
            for block in r.iter_content(chunk):
                f.write(block)

下载完成后应校验文件完整性，例如比对官方提供的MD5或记录文件大小与预期值。校验通过后再进入解析流程，避免半截文件污染下游。

四、网页抓取与文献补充材料提取

网页抓取优先使用结构化解析而非正则。对HTML表格，用pandas.read_html或BeautifulSoup定位table节点，提取表头与行数据；对分页列表，识别“下一页”链接或页码参数，循环抓取并去重。抓取前应确认目标站点的robots.txt与使用条款，控制请求频率，避免对服务端造成压力。

文献补充材料提取的难点在格式多样。PDF中的表格可用pdfplumber提取，XLSX用openpyxl，DOCX用python-docx。提取后必须做人工抽检，因为PDF表格常出现跨页断裂、合并单元格错位。建议对每个附件记录来源文献DOI、附件文件名、提取时间与抽检结论，形成可追溯的提取日志。

五、限流处理与增量更新机制

限流处理的核心是“主动限速+被动退避”。主动限速通过令牌桶控制每秒请求数，例如设定QPS=3，则每333毫秒放行一个请求；被动退避在收到429或503时触发，等待时间按指数增长。两者结合可显著降低被封禁风险。

增量更新依赖水位线。对API来源，记录上次成功采集的最大last_modified时间戳或版本号，下次只请求该时间之后的数据；对批量文件，记录文件版本号或发布日期，仅当版本变化时重新下载；对网页与文献，记录URL或DOI的哈希，新条目才进入采集队列。更新时对每条记录计算主键哈希，与已有数据比对，标记为新增、变更或删除，变更记录保留旧版本以便回溯。

增量更新不是“只取新数据”，而是“识别变化并保留变化痕迹”。建议在staging层维护一张采集日志表，字段包括：source、endpoint、watermark、fetched_at、record_count、status。这张表是排障与复现的关键依据。

六、采集质量检查要点

请求成功率：单次采集任务的成功响应占比，低于95%需排查限流或接口变更。
记录数一致性：分页累计条数与接口返回的total_count是否吻合。
字段完整性：关键字段（如SMILES、靶点、活性值、单位）缺失率是否在可接受范围。
去重效果：主键哈希重复率，重复条目应在staging层合并或标记。
可追溯性：每条记录能否回溯到原始响应快照与采集时间。

采集工程的目标不是“抓到最多”，而是“抓得稳、可复现、可增量”。把原始快照、采集日志、水位线和主键哈希这四件事做扎实，后续的清洗、标准化与数据集发布才有可靠基础。