生物医药蛋白分子数据采集：支撑大模型训练的技术实践分享

生物医药蛋白分子数据采集：支撑大模型训练的技术实践分享

一、学术数据库反爬机制分析

1.1 典型反爬策略

• 速率限制：PDB对同一IP设置500req/h的请求限制

• CAPTCHA验证：UniProt对高频请求触发Google Recaptcha

• 请求指纹识别：STRING要求必须使用浏览器User-Agent
• 动态渲染内容：RCSB PDB的结构数据需执行JavaScript加载

1.2 生物医药数据特性

• 数据格式复杂：包含XML/MMCIF/TSV等多种格式

• 结构验证严格：蛋白质三维结构数据需通过验证工具检查

• 合规要求高：需遵守《赫尔辛基宣言》等学术伦理规范

二、基础数据采集实现

2.1 环境配置

# Python环境
conda create -n bio-crawler python=3.10
conda activate bio-crawler
pip install biopython requests beautifulsoup4 selenium

# Java环境
mvn dependency:add -DgroupId=org.jsoup -DartifactId=jsoup -Dversion=1.17.2

2.2 基础采集代码

# PDB结构数据采集
def fetch_pdb_structure(pdb_id):
    url = f"https://www.rcsb.org/structure/{pdb_id}"
    headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        soup = BeautifulSoup(response.text, "html.parser")
        structure_data = soup.find("pre", id="mmcif")
        if structure_data:
            with open(f"{pdb_id}.cif", "w") as f:
                f.write(structure_data.text)
            return True
    return False

2.3 原始采集结果

pdb_ids = ["1A2B", "3C4D", "5E6F"]
for pdb_id in pdb_ids:
    print(f"Fetching {pdb_id}: {fetch_pdb_structure(pdb_id)}")

# 输出结果：
Fetching 1A2B: True
Fetching 3C4D: False (429 Too Many Requests)
Fetching 5E6F: False (403 Forbidden)

2.4 遇到的问题总结

三、反爬对抗技术实现

3.1 动态住宅代理配置

3.1.1 注册与通道创建

1. 账号注册

访问 [Bright Data官网](bright data )，用这个链接注册可以直接5折，不需要额外操作。点击注册及登录链接，即可直接享受折扣，非常推荐使用！

使用邮箱完成注册。登录成功之后，可看到以下界面

1. 创建代理通道

• 进入控制台 → 代理网络 → 动态住宅代理

• 点击「创建通道」，配置以下参数：

• 通道类型：选择「动态住宅代理」

• 并发数：根据采集需求设置（建议初始设为5-10）

• 地域锁定：选择目标数据库服务器所在地（如美国/欧洲节点）

• IP池大小：默认「自动」或指定最小IP数量

通道创建成功会提示如下

安装证书

测试一下

有多种代码示例，可根据不同的项目语言进行选择使用，

3.1.2 获取代理认证信息

在通道详情页获取以下参数：

PROXY_HOST = "brd.superproxy.io"  # 通用代理主机
PROXY_PORT = 22225                # 动态住宅代理端口
PROXY_USER = "brd-customer-hl_5aaa3f49-zone_residential_proxy1"  # 通道用户名
PROXY_PASS = "mq2tfjmm5t9d"       # 通道密码

3.1.3 代码集成代理

这里需要替换为你自己的信息即可

{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}

# 构建代理隧道
proxies = {
    "http": f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}",
    "https": f"https://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
}

# 验证代理有效性
def is_proxy_valid():
    try:
        response = requests.get(
            "https://api64.ipify.org?format=json",
            proxies=proxies,
            timeout=5
        )
        print(f"Current Proxy IP: {response.text}")
        return response.status_code == 200
    except Exception as e:
        print(f"Proxy validation failed: {str(e)}")
        return False

3.1.4 高级配置优化

1. 设备指纹伪装

   headers = {
       "User-Agent": random.choice([
           "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
           "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
       ]),
       "Accept-Language": "en-US,en;q=0.9"
   }

2. 地理位置筛选

   # 通过URL参数指定地域
   proxies = {
       "https": f"https://{PROXY_USER}-geo-us:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}"
   }

通过以上步骤，可快速构建稳定的动态住宅代理环境，有效突破学术数据库反爬限制。

3.2 Web Scraper API

如果觉得自己写代码效率不高，可以使用这个API ,开箱即用，支持海量的数据拉取。目前使用75折，很划算。

简单来说，这个工具就像给AI模型训练数据采集装了个「超级外挂」

注册成功之后可以看到

搜索需要的数据之后

按照步骤配置完成之后，可以看到你选择的数据示例：

3.2.1 代码实现

进入控制台 → Web Scraper → 获取 API 密钥python

API_KEY = "your_api_key"  # 从控制台复制

3.2.2 基本请求示例

import requests

API_URL = "https://scraper-api.brightdata.com"

def fetch_pubmed_article(article_id):
    params = {
        "url": f"https://pubmed.ncbi.nlm.nih.gov/{article_id}/",
        "output": "json",
        "geo": "us",
        "parse": "pubmed_article"
    }
    
    response = requests.get(
        API_URL,
        params=params,
        auth=(API_KEY, '')  # 认证方式1：URL参数
    )
    
    return response.json()

# 使用示例
data = fetch_pubmed_article("37000000")
print(data["title"])

3.2.3 高级参数配置

# 完整参数列表
params = {
    "url": "https://url.com",# 指定采集url
    "output": "json",  # 支持json, csv, html
    "geo": "us",      # 指定采集地域
    "parse": "default",# 选择解析模板
    "javascript": "enabled", # 启用JS渲染
    "timeout": 30,     # 超时时间（秒）
    "premium_proxy": "true" # 使用高级代理
}

3.2.4 预构建爬虫模板

我也有试了一下这个API，确实挺方便的减少很多代码量

# 支持的典型站点示例
SUPPORTED_SITES = [
    "pubmed.ncbi.nlm.nih.gov",
    "www.ncbi.nlm.nih.gov/pmc",
    "www.ebi.ac.uk/proteins",
    "www.biorxiv.org"
]

开箱即用的解析器：自动提取结构化数据python

# 解析PubMed文章元数据
params = {
    "url": "https://pubmed.ncbi.nlm.nih.gov/37000000/",
    "parse": "pubmed_article"
}

3.3 智能请求调度

另外在代码中也可以加上智能请求调度的配置，用来退避请求控制

# 带指数退避的请求控制
def schedule_request(attempt=1):
    max_attempts = 5
    delay = 2 ** attempt + random.uniform(0, 1)
    
    if attempt > max_attempts:
        raise Exception("Max attempts exceeded")
    
    time.sleep(delay)
    return attempt + 1

3.4 多线程采集优化

使用线程池并发采集，采集效率更高

# 使用线程池并发采集
from concurrent.futures import ThreadPoolExecutor

def process_pdb_id(pdb_id):
    attempt = 1
    while attempt <= 3:
        try:
            success = fetch_pdb_structure(pdb_id, proxies)
            if success:
                print(f"Success: {pdb_id}")
                return
            else:
                attempt = schedule_request(attempt)
        except Exception as e:
            print(f"Error processing {pdb_id}: {str(e)}")
            attempt = schedule_request(attempt)

# 启动10线程并发
with ThreadPoolExecutor(max_workers=10) as executor:
    executor.map(process_pdb_id, pdb_ids)

四、数据质量保障

4.1 数据格式验证

# PDB文件验证脚本
from Bio.PDB import PDBParser

def validate_pdb_file(file_path):
    try:
        parser = PDBParser(QUIET=True)
        structure = parser.get_structure("test", file_path)
        return len(list(structure.get_atoms())) > 0
    except Exception as e:
        return False

4.2 学术合规处理

# 添加请求延迟遵守数据库条款
import time

def polite_request():
    time.sleep(1)  # 至少1秒间隔

五、性能优化与效果对比

5.1 优化策略对比

5.2 关键技术点

1. 代理网络动态切换：基于地理位置的智能节点选择

2. 请求指纹随机化：

   headers = {
       "User-Agent": random.choice([
           "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
           "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
       ])
   }

   def clean_pdb_data(data):
       return [atom for atom in data if atom.get_full_id()[3][0] != 'W']

六、学术合规与风险控制

6.1 数据库API优先

# 使用PDB API获取数据
response = requests.get(
    "https://data.rcsb.org/rest/v1/core/entry/1A2B",
    headers={"Accept": "application/json"}
)

6.2 伦理审查机制

1. 确保数据不涉及人类受试者隐私

2. 添加数据使用声明：

   print("This data is used for academic research purposes only.")