企业官网建设流程全解析

Python药监局数据爬取实战：Selenium反反爬与GBK编码解决方案

政府类网站的数据爬取向来是技术挑战的高发区。最近在协助某医药数据分析团队构建药品监管信息库时，我们遇到了典型的"药监局数据采集困境"——明明第一次请求成功获取数据，刷新后却返回空结果；GBK编码导致的乱码问题频发；即使用上Selenium也很快被识别阻断。本文将分享经过实战验证的解决方案，这些方法同样适用于其他政府门户网站的数据采集。

1. 动态参数反爬机制破解

药监局网站的反爬策略中，最令人头疼的莫过于动态参数验证。许多开发者遇到的"第一次成功，后续失败"问题，根源在于以下几个关键点：

# 典型的问题URL结构示例 problematic_url = 'http://app1.nmpa.gov.cn/...&State=1&curstart=1&State=1&tableName=...&State=1'

关键发现：

• 重复出现的State=1参数实际上是服务端会话验证的标记
• curstart参数虽然控制分页，但其有效性依赖于上下文会话状态
• URL中的tableView等参数需要GBK编码处理

解决方案采用请求会话保持+参数动态生成策略：

import requests from urllib.parse import quote session = requests.Session() session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Connection': 'keep-alive' }) def build_url(page): base = "http://app1.nmpa.gov.cn/datasearchcnda/face3/search.jsp" params = { 'tableId': '121', 'bcId': '152894...', # 需替换为实际值 'curstart': str(page), 'tableView': quote('全国药品抽查', encoding='gbk') # 关键GBK编码处理 } return f"{base}?{'&'.join(f'{k}={v}' for k,v in params.items())}"

2. Selenium反检测高级策略

当基础请求被拦截时，Selenium往往是第二选择。但药监局网站对WebDriver的检测相当严格，需要多层次的规避措施：

完整反检测配置方案：

from selenium import webdriver from selenium.webdriver.chrome.options import Options def create_stealth_driver(): options = Options() options.add_argument("--disable-blink-features=AutomationControlled") options.add_argument("--disable-infobars") options.add_argument("--disable-notifications") options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) # 关键JavaScript注入 driver = webdriver.Chrome(options=options) driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", { "source": """ Object.defineProperty(navigator, 'webdriver', { get: () => undefined }) window.navigator.chrome = { runtime: {}, // etc. }; """ }) return driver

实战中发现的三个关键细节：

1. 每次操作后需要随机延迟（2-5秒）
2. 鼠标移动轨迹需要模拟人类操作（使用ActionChains）
3. 分页操作不宜过快，建议每3页后休息10-15秒

3. GBK编码处理全方案

药监局网站的部分接口仍使用GBK编码，这是许多乱码问题的根源。我们总结出以下处理方案：

编码问题解决矩阵：

典型处理代码：

from urllib.parse import quote import requests # URL构建 product_type = quote('医疗器械生产企业（许可）', encoding='gbk') url = f'http://example.com/search?key={product_type}' # 响应处理 response = requests.get(url) response.encoding = 'gbk' # 关键解码设置 content = response.text

4. 请求频率与IP轮换策略

即使解决了技术层面的反爬问题，操作频率控制仍是成功的关键。我们开发了一套智能节流系统：

动态延迟算法：

import random import time def smart_delay(last_request_time): elapsed = time.time() - last_request_time base_interval = 3.0 # 基础间隔 # 随机波动(±30%) fluctuation = random.uniform(0.7, 1.3) # 根据历史请求动态调整 dynamic_factor = 1 + (0.5 if elapsed < 2 else -0.2) delay = base_interval * fluctuation * dynamic_factor time.sleep(max(delay, 1.0)) # 不低于1秒

IP管理方案对比：

5. 数据存储与异常处理

稳定的数据存储方案需要考虑政府网站的特点：

健壮性存储框架：

import pymysql import json from contextlib import contextmanager @contextmanager def db_connection(): conn = pymysql.connect( host='localhost', user='user', password='pass', database='nmpa_data', charset='gb18030' # 关键编码设置 ) try: yield conn finally: conn.close() def save_item(item): with db_connection() as conn: cursor = conn.cursor() try: cursor.execute(""" INSERT INTO products (id, name, data, raw_json) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE data=VALUES(data), raw_json=VALUES(raw_json) """, (item['id'], item['name'], json.dumps(item), json.dumps(item))) conn.commit() except Exception as e: conn.rollback() log_error(e) raise

异常处理清单：

• 网络超时重试（最多3次）
• 响应数据校验（检查关键字段）
• 数据库死锁处理
• 编码异常回退机制
• 存储空间监控

6. 完整工作流实现

将上述方案整合为自动化采集系统：

import logging from datetime import datetime logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('nmpa_crawler.log', encoding='utf-8'), logging.StreamHandler() ] ) class NMPACrawler: def __init__(self): self.session = requests.Session() self.last_request = time.time() def crawl_page(self, page): try: url = self.build_url(page) self.smart_delay() response = self.session.get(url) response.encoding = 'gbk' if not self.validate_response(response): raise ValueError("Invalid response content") return self.parse_html(response.text) except Exception as e: logging.error(f"Page {page} failed: {str(e)}") return None def run(self, start_page, end_page): for page in range(start_page, end_page + 1): data = self.crawl_page(page) if data: self.store_data(data) if page % 3 == 0: # 每3页休息 time.sleep(random.uniform(10, 15))

在实际项目中，这套方案成功实现了对药监局多个数据表的稳定采集，日均获取数据量超过2万条，可用率达到98%以上。最难能可贵的是，系统已经稳定运行超过6个月，期间仅需微调参数即可适应网站的微小变动。