医疗数据爬虫代理：如何在不被封锁的情况下收集信息

医疗数据爬虫是一项需要特别选择代理的任务。医疗门户网站、临床研究数据库和制药资源使用先进的数据收集保护系统。在本文中，我们将讨论如何正确设置代理以安全地爬取医疗信息，避免被封禁，并有效地收集所需数据。

为什么医疗网站会封禁爬虫

医疗门户网站和数据库对自动化信息收集特别敏感，原因有几个。首先，许多网站是商业性质的，通过付费订阅出售数据访问权限。自动爬虫可能会违反使用条款和许可协议。

其次，医疗数据通常包含受法律保护的机密信息（美国的HIPAA，欧洲的GDPR）。资源所有者必须控制对这些数据的访问，并防止其未经授权的传播。因此，他们使用先进的保护系统：

• 速率限制 — 限制每个IP地址在单位时间内的请求数量（通常为每分钟10-50个请求）
• 指纹识别 — 分析浏览器特征、HTTP头、资源加载顺序
• 验证码 — reCAPTCHA v3等系统在可疑活动时触发
• IP封禁 — 临时或永久封禁数据中心的IP地址
• Cloudflare及类似服务 — 在CDN层面防止机器人

第三个原因是服务器负载。医疗数据库通常包含数百万条记录，大规模爬虫可能会对基础设施造成显著负担。因此，管理员积极打击自动化数据收集，监测典型的机器人行为模式：请求之间的相同间隔、线性页面浏览、缺少JavaScript和cookies。

选择哪种类型的代理用于医疗数据

选择代理类型对成功爬取医疗数据至关重要。不同的来源需要不同的处理方式。我们来看看主要的代理类型及其适用性：

对于大多数医疗数据爬虫任务，建议使用住宅代理。它们在成本和效率之间提供了最佳平衡。数据中心代理仅适用于没有保护的开放来源。移动代理应在其他类型无效时使用。

针对特定来源的选择建议

• PubMed, PubMed Central — 数据中心代理足够，但限制速率为每秒3个请求
• ClinicalTrials.gov — 数据中心代理，有官方API
• Elsevier, Springer, Wiley — 必须使用住宅代理，使用高级指纹识别
• DrugBank, RxList — 住宅代理，积极防止爬虫
• FDA, EMA数据库 — 数据中心代理适用，但爬虫速度较慢

主要医疗数据来源及其保护

医疗数据分布在多个来源中，每个来源都有其特性和保护级别。了解这些特性将有助于正确设置爬虫策略。

开放的政府数据库

PubMed/PubMed Central — 最大的医学出版物数据库，包含超过3500万条记录。美国国家医学图书馆（NLM）提供官方的E-utilities API，这是访问数据的首选方式。直接爬取网页界面是可能的，但限制为每个IP每秒3个请求。超过限制会导致24小时的临时封禁。

ClinicalTrials.gov — 临床研究数据库，包含来自220个国家的超过400,000项研究的信息。也提供API以供程序访问。网页界面受到速率限制 — 每个IP每5分钟最多100个请求。使用基本的防机器人保护，但没有Cloudflare。

FDA药物数据库 — FDA批准的药物数据库。通过网页界面和openFDA API提供开放访问。限制：匿名用户每分钟240个请求，使用API密钥的用户每分钟1000个请求。封禁很少发生，但在激进爬虫时可能会出现。

商业科学出版社

Elsevier (ScienceDirect) — 最大的科学文献出版商之一。使用多层保护：Cloudflare、浏览器指纹识别、用户行为分析。检测到自动下载模式：顺序访问文章、缺少JavaScript、非典型User-Agent。发现爬虫后，会在账户级别封禁IP，并可能封禁整个机构。必须使用带轮换和完全浏览器模拟的住宅代理。

Springer Nature — 类似的保护，额外监测页面滚动速度和鼠标移动。使用机器学习检测机器人。建议每个IP每小时爬取不超过10-15篇文章，并在请求之间随机延迟。

Wiley Online Library — 保护较少，但仍要求使用代理。每个IP每小时允许大约50个请求而不被封禁。使用会话cookies跟踪活动。

制药数据库

DrugBank — 综合药物数据库。免费版本仅限于网页界面，商业版本提供API和数据导出。网页版本受到Cloudflare和速率限制保护 — 每分钟最多20个请求。通过缺少cookies和JavaScript检测自动化。

RxList, Drugs.com — 面向消费者的流行药物指南。使用Cloudflare并积极打击爬虫。几乎瞬间封禁数据中心的IP。需要住宅代理和较慢的爬虫速度（每分钟5-10页）。

设置IP轮换以进行长期爬虫

正确的IP地址轮换是成功爬取医疗数据的关键因素。主要有两种方法：基于请求的轮换和基于时间的轮换。

基于请求的轮换

在这种方法中，每个请求通过新的IP地址发送。这最大限度地降低了被封禁的风险，但可能会导致那些通过cookies跟踪会话的网站出现问题。适合爬取列表和目录，不需要保持会话状态。

大多数住宅代理提供通过特殊端点的自动轮换。例如，使用轮换代理端点时，每个新的TCP连接都会获得新的IP。这在Python的requests库中自动工作，因为默认情况下为每个请求创建新的连接。

基于时间的轮换（粘性会话）

粘性会话允许在特定时间内使用一个IP地址（通常为5-30分钟），然后自动更换。这对于需要授权或通过cookies跟踪会话状态的网站很有用。您可以使用一个IP爬取多个页面，模拟真实用户的行为，然后IP会自动更换。

对于医疗网站，建议使用持续时间为10-15分钟的粘性会话。在此期间可以爬取10-20个页面（根据延迟），之后IP会更换，您开始“新会话”。这看起来很自然，并降低了被检测的风险。

IP地址池的大小

对于长期爬虫，IP地址池的大小很重要。如果您在一周内使用相同的100个IP，网站可能会注意到模式并封禁所有这些地址。住宅代理通常提供数百万个IP的访问，这几乎消除了重复使用同一地址的可能性。

使用数据中心代理时，建议至少拥有500-1000个IP的池，以进行中等量的爬虫（每月10,000-50,000页）。对于大规模爬虫（数十万页），最好使用住宅代理及其庞大的IP池。

使用代理的Python代码示例

让我们看看如何使用流行的Python库设置代理以爬取医疗数据的实际示例。从基本示例开始，逐步复杂化。

使用requests库的基本设置

import requests
from time import sleep
import random

# 设置代理（替换为您的数据）
PROXY_HOST = "proxy.example.com"
PROXY_PORT = "8080"
PROXY_USER = "username"
PROXY_PASS = "password"

proxies = {
    'http': f'http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}',
    'https': f'http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}'
}

# 模拟真实浏览器的请求头
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1'
}

# 请求PubMed的示例
url = "https://pubmed.ncbi.nlm.nih.gov/?term=diabetes"

try:
    response = requests.get(url, proxies=proxies, headers=headers, timeout=30)
    print(f"状态码: {response.status_code}")
    print(f"内容长度: {len(response.content)}")
    
    # 在请求之间添加延迟（PubMed必需）
    sleep(random.uniform(1.0, 3.0))
    
except requests.exceptions.RequestException as e:
    print(f"错误: {e}")

带轮换和重试逻辑的高级设置

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from time import sleep
import random

class ProxyRotator:
    def __init__(self, proxy_list):
        """
        proxy_list: 代理字典列表
        [{'http': 'http://user:pass@host:port', 'https': '...'}, ...]
        """
        self.proxy_list = proxy_list
        self.current_index = 0
    
    def get_next_proxy(self):
        """获取下一个代理"""
        proxy = self.proxy_list[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxy_list)
        return proxy

def create_session_with_retries():
    """创建带自动重试的会话"""
    session = requests.Session()
    
    # 设置自动重试
    retry_strategy = Retry(
        total=3,  # 最多3次尝试
        backoff_factor=1,  # 尝试之间的延迟：1, 2, 4秒
        status_forcelist=[429, 500, 502, 503, 504],  # 重试的状态码
        allowed_methods=["GET", "POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

def scrape_with_rotation(urls, proxy_rotator):
    """使用代理轮换爬取URL列表"""
    session = create_session_with_retries()
    results = []
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.9',
    }
    
    for url in urls:
        # 为每个请求获取新的代理
        proxy = proxy_rotator.get_next_proxy()
        
        try:
            response = session.get(
                url, 
                proxies=proxy, 
                headers=headers, 
                timeout=30
            )
            
            if response.status_code == 200:
                results.append({
                    'url': url,
                    'status': 'success',
                    'content_length': len(response.content)
                })
                print(f"✓ 成功: {url}")
            else:
                results.append({
                    'url': url,
                    'status': 'failed',
                    'error': f"状态码: {response.status_code}"
                })
                print(f"✗ 失败: {url} (状态: {response.status_code})")
        
        except requests.exceptions.RequestException as e:
            results.append({
                'url': url,
                'status': 'error',
                'error': str(e)
            })
            print(f"✗ 错误: {url} ({e})")
        
        # 请求之间的随机延迟（重要！）
        sleep(random.uniform(2.0, 5.0))
    
    return results

# 示例使用
proxy_list = [
    {
        'http': 'http://user1:pass1@proxy1.example.com:8080',
        'https': 'http://user1:pass1@proxy1.example.com:8080'
    },
    {
        'http': 'http://user2:pass2@proxy2.example.com:8080',
        'https': 'http://user2:pass2@proxy2.example.com:8080'
    }
]

rotator = ProxyRotator(proxy_list)

urls_to_scrape = [
    "https://pubmed.ncbi.nlm.nih.gov/?term=diabetes",
    "https://pubmed.ncbi.nlm.nih.gov/?term=cancer",
    "https://pubmed.ncbi.nlm.nih.gov/?term=covid"
]

results = scrape_with_rotation(urls_to_scrape, rotator)

使用Selenium处理JavaScript网站

许多现代医疗网站使用JavaScript加载内容。在这种情况下，需要无头浏览器：

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

def create_proxy_driver(proxy_host, proxy_port, proxy_user, proxy_pass):
    """创建带代理的Chrome WebDriver"""
    
    chrome_options = Options()
    
    # 无头模式（无GUI）
    chrome_options.add_argument('--headless')
    chrome_options.add_argument('--no-sandbox')
    chrome_options.add_argument('--disable-dev-shm-usage')
    
    # 设置代理
    chrome_options.add_argument(f'--proxy-server=http://{proxy_host}:{proxy_port}')
    
    # 关闭自动化（对绕过检测很重要）
    chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
    chrome_options.add_experimental_option('useAutomationExtension', False)
    
    # User-Agent
    chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')
    
    driver = webdriver.Chrome(options=chrome_options)
    
    # 对于需要身份验证的代理，需要使用扩展或通过能力设置（更复杂的选项）
    
    return driver

def scrape_with_selenium(url, driver):
    """爬取页面并等待JavaScript加载"""
    
    driver.get(url)
    
    # 等待元素加载（例如，搜索结果）
    try:
        wait = WebDriverWait(driver, 10)
        results = wait.until(
            EC.presence_of_element_located((By.CLASS_NAME, "results-article"))
        )
        
        # 提取数据
        articles = driver.find_elements(By.CLASS_NAME, "results-article")
        
        data = []
        for article in articles:
            try:
                title = article.find_element(By.CLASS_NAME, "docsum-title").text
                authors = article.find_element(By.CLASS_NAME, "docsum-authors").text
                
                data.append({
                    'title': title,
                    'authors': authors
                })
            except:
                continue
        
        return data
        
    except Exception as e:
        print(f"等待元素时出错: {e}")
        return []

# 示例使用
proxy_host = "proxy.example.com"
proxy_port = "8080"
proxy_user = "username"
proxy_pass = "password"

driver = create_proxy_driver(proxy_host, proxy_port, proxy_user, proxy_pass)

try:
    url = "https://pubmed.ncbi.nlm.nih.gov/?term=diabetes"
    results = scrape_with_selenium(url, driver)
    
    for result in results:
        print(f"标题: {result['title']}")
        print(f"作者: {result['authors']}\n")
        
finally:
    driver.quit()

请求速率控制与绕过速率限制

速率限制是医疗网站防止爬虫的主要保护措施之一。正确设置请求速率对长期爬虫而不被封禁至关重要。

确定安全速率

第一步是确定特定网站的限制。这可以通过实验性地逐步提高请求速率，直到出现429（请求过多）错误或封禁。对于大多数医疗网站，安全值如下：

• PubMed — 每秒最多3个请求（官方建议）
• ClinicalTrials.gov — 每分钟20个请求安全，最多100个请求在5分钟内可接受
• 商业出版社 — 每个IP每小时10-15个请求
• 制药数据库 — 每分钟5-10个请求

在Python中实现速率限制器

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_calls, period):
        """
        max_calls: 最大调用次数
        period: 时间周期（秒）
        例如：RateLimiter(3, 1) = 每秒3个请求
        """
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
    
    def __call__(self, func):
        """限制函数调用速率的装饰器"""
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # 删除超出周期的旧调用
            while self.calls and self.calls[0] < now - self.period:
                self.calls.popleft()
            
            # 如果达到限制，则等待
            if len(self.calls) >= self.max_calls:
                sleep_time = self.period - (now - self.calls[0])
                if sleep_time > 0:
                    print(f"达到速率限制，等待 {sleep_time:.2f}s")
                    time.sleep(sleep_time)
                    # 等待后清空调用记录
                    self.calls.clear()
            
            # 记录调用时间
            self.calls.append(time.time())
            
            # 执行函数
            return func(*args, **kwargs)
        
        return wrapper

# 示例使用
@RateLimiter(max_calls=3, period=1)  # 每秒3个请求
def fetch_pubmed_page(url):
    response = requests.get(url, headers=headers, proxies=proxies)
    return response

# 现在函数会自动遵守速率限制
for i in range(10):
    result = fetch_pubmed_page(f"https://pubmed.ncbi.nlm.nih.gov/?term=test&page={i}")
    print(f"页面 {i} 已获取")

自适应速率限制

更高级的方法是根据服务器的响应动态调整速率。如果收到429或503错误，则自动降低速率：

import time
import random

class AdaptiveRateLimiter:
    def __init__(self, initial_delay=1.0, max_delay=60.0):
        self.current_delay = initial_delay
        self.initial_delay = initial_delay
        self.max_delay = max_delay
        self.success_count = 0
    
    def wait(self):
        """在下一个请求之前等待"""
        # 添加随机性以增加自然性
        actual_delay = self.current_delay * random.uniform(0.8, 1.2)
        time.sleep(actual_delay)
    
    def on_success(self):
        """在成功请求时调用"""
        self.success_count += 1
        
        # 在10次成功请求后稍微加快速度
        if self.success_count >= 10:
            self.current_delay = max(
                self.initial_delay,
                self.current_delay * 0.9
            )
            self.success_count = 0
    
    def on_rate_limit(self):
        """在收到429或类似错误时调用"""
        # 将延迟加倍，但不超过最大值
        self.current_delay = min(
            self.current_delay * 2,
            self.max_delay
        )
        self.success_count = 0
        print(f"达到速率限制！将延迟增加到 {self.current_delay:.2f}s")
    
    def on_error(self):
        """在其他错误时调用"""
        # 略微增加延迟
        self.current_delay = min(
            self.current_delay * 1.5,
            self.max_delay
        )
        self.success_count = 0

# 示例使用
limiter = AdaptiveRateLimiter(initial_delay=2.0, max_delay=30.0)

for url in urls_to_scrape:
    limiter.wait()
    
    try:
        response = requests.get(url, proxies=proxies, headers=headers)
        
        if response.status_code == 200:
            limiter.on_success()
            # 处理数据
            
        elif response.status_code == 429:
            limiter.on_rate_limit()
            # 稍后重试
            
        else:
            limiter.on_error()
            
    except requests.exceptions.RequestException:
        limiter.on_error()

医疗网站的正确请求头和User-Agent

医疗网站分析HTTP头以检测机器人。错误或缺失的头部是使用优质代理时被封禁的常见原因。

必需的请求头

每个请求中必须包含的最小请求头：

headers = {
    # User-Agent — 必须是最新的浏览器
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    
    # Accept — 浏览器接受的内容类型
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
    
    # Accept-Language — 用户语言
    'Accept-Language': 'en-US,en;q=0.9',
    
    # Accept-Encoding — 支持压缩
    'Accept-Encoding': 'gzip, deflate, br',
    
    # Connection — 保持连接
    'Connection': 'keep-alive',
    
    # Upgrade-Insecure-Requests — 自动切换到HTTPS
    'Upgrade-Insecure-Requests': '1',
    
    # DNT — Do Not Track（可选，但增加真实性）
    'DNT': '1',
    
    # Sec-Fetch-* 头（对现代浏览器很重要）
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1',
    
    # Cache-Control
    'Cache-Control': 'max-age=0'
}

User-Agent的轮换

使用相同的User-Agent可能会引起怀疑。建议在多个最新浏览器之间进行轮换：

import random

USER_AGENTS = [
    # Windows上的Chrome
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    
    # Mac上的Chrome
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    
    # Windows上的Firefox
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
    
    # Mac上的Firefox
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:121.0) Gecko/20100101 Firefox/121.0',
    
    # Mac上的Safari
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15',
    
    # Windows上的Edge
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0'
]

def get_random_headers():
    """获取带随机User-Agent的请求头"""
    return {
        'User-Agent': random.choice(USER_AGENTS),
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
        'Accept-Language': 'en-US,en;q=0.9',
        'Accept-Encoding': 'gzip, deflate, br',
        'Connection': 'keep-alive',
        'Upgrade-Insecure-Requests': '1',
        'DNT': '1'
    }

# 使用示例
for url in urls:
    headers = get_random_headers()
    response = requests.get(url, headers=headers, proxies=proxies)

表单的Referer和Origin

在处理搜索表单或发送POST请求时，务必添加Referer和Origin头：

# 对于搜索表单的POST请求
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept-Encoding': 'gzip, deflate, br',
    'Content-Type': 'application/x-www-form-urlencoded',
    'Origin': 'https://example.com',
    'Referer': 'https://example.com/search',
    'Connection': 'keep-alive'
}

# 带表单数据的POST请求
data = {
    'query': 'diabetes',
    'page': '1'
}

response = requests.post(
    'https://example.com/search',
    headers=headers,
    data=data,
    proxies=proxies
)

常见问题及其解决方案

在爬取医疗数据时会遇到特定问题。让我们看看最常见的问题及其解决方案。

问题：Cloudflare封禁所有请求

症状： 收到“正在检查您的浏览器”或403 Forbidden错误，提到Cloudflare。

解决方案：

• 使用住宅代理而不是数据中心代理 — Cloudflare默认封禁数据中心的IP
• 切换到Selenium或Puppeteer — 无头浏览器更容易通过Cloudflare的检查
• 使用Python的cloudscraper库 — 它可以自动绕过Cloudflare的基本保护
• 启用cookies和JavaScript — Cloudflare会检查它们的存在
• 添加TLS指纹识别 — 使用curl_cffi模拟真实浏览器的TLS层

问题：收到429 Too Many Requests错误

症状： 在成功请求几次后，服务器开始返回429。

解决方案：

• 增加请求之间的延迟 — 尝试从3-5秒开始
• 启用IP轮换 — 每个请求通过新的IP可以解除速率限制
• 检查429响应中的Retry-After头 — 它指示需要等待多少秒
• 在重试时使用指数延迟 — 1秒、2秒、4秒、8秒等

问题：代理速度慢或经常掉线

症状： 超时错误，页面加载非常缓慢，连接中断。

解决方案：

• 将请求的超时增加到30-60秒 — 住宅代理可能较慢
• 使用地理上接近的代理 — 如果爬取欧洲网站，请使用欧洲IP
• 检查代理提供商的质量 — 便宜的代理通常不稳定
• 添加重试逻辑 — 在连接错误时自动重试请求
• 使用连接池 — 通过requests.Session()重用TCP连接

问题：网站要求授权或订阅

症状： 访问文章的完整文本受到限制，需要登录。

解决方案：

• 使用机构访问 — 许多大学和医院有订阅
• 检查是否有开放获取版本 — 许多文章可以通过存储库免费获取
• 使用API而不是爬虫 — 一些出版社为研究人员提供API
• 仅爬取元数据（标题、作者、摘要） — 通常可以免费访问

问题：JavaScript内容未加载

症状： HTML中没有所需数据，仅显示加载旋转器或空容器。

解决方案：

• 切换到Selenium/Puppeteer — 它们可以执行JavaScript
• 查找API端点 — 在浏览器中打开开发者工具，网络选项卡，查找带数据的XHR请求
• 使用requests-html — 一个可以执行JavaScript的库