Python药监局数据爬取实战Selenium反反爬与GBK编码解决方案政府类网站的数据爬取向来是技术挑战的高发区。最近在协助某医药数据分析团队构建药品监管信息库时我们遇到了典型的药监局数据采集困境——明明第一次请求成功获取数据刷新后却返回空结果GBK编码导致的乱码问题频发即使用上Selenium也很快被识别阻断。本文将分享经过实战验证的解决方案这些方法同样适用于其他政府门户网站的数据采集。1. 动态参数反爬机制破解药监局网站的反爬策略中最令人头疼的莫过于动态参数验证。许多开发者遇到的第一次成功后续失败问题根源在于以下几个关键点# 典型的问题URL结构示例 problematic_url http://app1.nmpa.gov.cn/...State1curstart1State1tableName...State1关键发现重复出现的State1参数实际上是服务端会话验证的标记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;q0.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(全国药品抽查, encodinggbk) # 关键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-featuresAutomationControlled) 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(optionsoptions) driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }) window.navigator.chrome { runtime: {}, // etc. }; }) return driver实战中发现的三个关键细节每次操作后需要随机延迟2-5秒鼠标移动轨迹需要模拟人类操作使用ActionChains分页操作不宜过快建议每3页后休息10-15秒3. GBK编码处理全方案药监局网站的部分接口仍使用GBK编码这是许多乱码问题的根源。我们总结出以下处理方案编码问题解决矩阵场景解决方案代码示例URL参数编码使用urllib.parse.quote指定GBKquote(医疗器械, encodinggbk)响应内容解码指定response.encodingresponse.encoding gbk文件存储明确指定写入编码open(data.txt, w, encodinggb18030)数据库存储配置连接字符集pymysql.connect(charsetgbk)典型处理代码from urllib.parse import quote import requests # URL构建 product_type quote(医疗器械生产企业许可, encodinggbk) url fhttp://example.com/search?key{product_type} # 响应处理 response requests.get(url) response.encoding gbk # 关键解码设置 content response.text4. 请求频率与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管理方案对比方案类型优点缺点适用场景免费代理池零成本稳定性差低频测试商业轮换代理高可用费用较高生产环境云服务器轮换完全控制配置复杂大规模采集Tor网络匿名性强速度极慢特殊需求5. 数据存储与异常处理稳定的数据存储方案需要考虑政府网站的特点健壮性存储框架import pymysql import json from contextlib import contextmanager contextmanager def db_connection(): conn pymysql.connect( hostlocalhost, useruser, passwordpass, databasenmpa_data, charsetgb18030 # 关键编码设置 ) 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 dataVALUES(data), raw_jsonVALUES(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( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(nmpa_crawler.log, encodingutf-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(fPage {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个月期间仅需微调参数即可适应网站的微小变动。