1. 动态网页爬虫的核心挑战
当你打开国家药监局化妆品生产许可信息网站时,页面显示的企业列表看似普通,但右键查看网页源代码会发现——这些数据根本不在HTML里!这就是典型的动态加载数据场景,也是现代网页反爬虫的常见手段。传统爬虫直接解析HTML的方法在这里完全失效,因为关键数据是通过JavaScript异步加载的。
我刚开始接触这类爬虫时,曾花了整整三天时间才搞明白问题所在。后来发现,动态网页的数据传输通常遵循这样的规律:网页骨架通过初始HTML加载,真实数据则通过AJAX请求以JSON格式传输。这就好比去餐厅吃饭,服务员先给你菜单(HTML骨架),等你点完菜(触发AJAX请求)后,厨房才会真正准备菜品(JSON数据)。
2. 逆向工程:捕捉数据接口
要攻克动态数据,首先需要化身"网络侦探"。以Chrome浏览器为例:
- 按F12打开开发者工具
- 切换到Network面板
- 勾选"Preserve log"防止请求记录丢失
- 在页面上执行你需要爬取数据的操作(如点击分页)
这时你会看到一堆网络请求,重点关注XHR和Fetch类型的请求。我在分析药监局网站时,发现这样一个关键请求:
http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?method=getXkzsList通过检查这个请求的Headers和Response,确认了几个重要特征:
- 请求方法是POST
- 提交的数据包含页码、每页条数等参数
- 返回的是标准的JSON格式数据
这就是我们要找的"数据管道"!这个过程就像侦探破案,需要仔细观察每个线索。记得有次我漏看了Request Headers里的Content-Type,导致始终无法正确获取数据,教训深刻。
3. 构造POST请求的艺术
找到接口只是第一步,正确构造POST请求才是技术活。requests库虽然简单,但细节决定成败。以下是经过实战检验的请求模板:
import requests url = 'http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?method=getXkzsList' headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8' } data = { 'on': 'true', 'page': '1', # 当前页码 'pageSize': '15', # 每页条数 'productName': '', 'conditionType': '1', 'applyname': '', 'applysn': '' } response = requests.post(url, headers=headers, data=data)这里有几个容易踩坑的地方:
- Headers伪装:缺少User-Agent可能会被拒绝访问
- 参数编码:药监局接口需要x-www-form-urlencoded格式
- 参数完整性:少传任何一个参数都可能导致返回空数据
我曾遇到过一个奇葩情况:同样的代码在公司网络能跑,回家就连不上。后来发现是公司网络代理自动添加了额外Header,这提醒我们环境一致性多么重要。
4. JSON数据深度解析技巧
拿到JSON响应只是开始,如何高效提取数据才是关键。Python的json模块虽然简单,但配合一些技巧能让代码更健壮:
import json # 假设response是requests返回的对象 try: result = response.json() if result['list']: # 检查关键字段是否存在 for item in result['list']: company_id = item.get('ID', '') # 使用get避免KeyError name = item.get('EPS_NAME', '') print(f"企业ID: {company_id}, 名称: {name}") except json.JSONDecodeError as e: print(f"JSON解析失败: {e}")处理JSON数据时,我总结了几条黄金法则:
- 永远假设字段可能不存在,使用dict.get()方法
- 添加异常处理应对网络波动导致的残缺数据
- 先打印完整响应,确认数据结构再编写解析逻辑
有个项目我忘了做异常处理,结果半夜爬虫崩溃,错过了重要数据采集窗口期,这个教训让我养成了严谨的编码习惯。
5. 分页抓取与数据存储实战
单页数据远远不够,我们需要系统化抓取全部数据。结合药监局网站的特点,分页抓取的完整流程如下:
import requests import json import time base_url = 'http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do' all_data = [] for page in range(1, 11): # 假设抓取前10页 data = { 'on': 'true', 'page': str(page), 'pageSize': '15', 'conditionType': '1', } try: response = requests.post( url=base_url + '?method=getXkzsList', headers=headers, data=data, timeout=10 ) response.raise_for_status() # 检查HTTP错误 page_data = response.json() if not page_data.get('list'): break # 没有数据时退出循环 all_data.extend(page_data['list']) print(f"已抓取第{page}页,累计{len(all_data)}条数据") time.sleep(1) # 礼貌性延迟 except Exception as e: print(f"第{page}页抓取失败: {e}") continue # 持久化存储 with open('pharma_companies.json', 'w', encoding='utf-8') as f: json.dump(all_data, f, ensure_ascii=False, indent=2)这个流程中我特别加入了几个实用技巧:
- 超时设置:避免某个请求卡死整个程序
- 异常恢复:单页失败不影响整体任务
- 速率控制:适当sleep避免被封IP
- 数据校验:检查关键字段确保数据质量
存储时使用ensure_ascii=False可以完美保存中文,indent参数让生成的JSON文件更易读。记得有次我忘了设置编码,打开文件全是\uXXXX的Unicode转义符,排查了半天才发现问题。
6. 企业详情数据的抓取策略
获取企业列表只是第一步,每个企业的详细信息才是价值所在。通过分析我们发现,详情页也是动态加载的,其URL模式为:
http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?method=getXkzsById&id=企业ID基于之前获取的企业ID列表,我们可以批量抓取详情数据:
detail_url = 'http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?method=getXkzsById' company_details = [] for company in all_data[:50]: # 示例只处理前50家 params = {'id': company['ID']} try: resp = requests.post(detail_url, headers=headers, data=params) detail = resp.json() if detail.get('businessLicenseNumber'): company_details.append(detail) print(f"获取到{company['EPS_NAME']}的详细信息") else: print(f"企业{company['ID']}数据不完整") except Exception as e: print(f"获取{company['ID']}详情失败: {e}") time.sleep(0.5) # 降低请求频率在处理详情数据时,有几个特别需要注意的点:
- 数据验证:检查关键字段如许可证号是否为空
- 错误隔离:单条记录失败不应中断整个流程
- 增量抓取:建议先保存已抓取的ID,避免重复请求
我曾经犯过一个错误:没有验证数据完整性就直接存储,结果发现30%的记录缺少关键字段。现在我会先抽样检查数据质量,再决定是否进行批量处理。
7. 反爬虫策略与应对方案
药监局网站虽然没设严格的反爬,但遵循爬虫礼仪很重要。根据我的经验,以下策略能有效降低被封风险:
- 请求头伪装:完整的headers包括:
headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': 'http://scxk.nmpa.gov.cn:81/xk/', 'Origin': 'http://scxk.nmpa.gov.cn:81', 'Accept-Language': 'zh-CN,zh;q=0.9', }- 请求频率控制:随机延迟更自然
import random time.sleep(random.uniform(0.5, 2))- 代理IP池:应对IP封锁
proxies = { 'http': 'http://user:pass@proxy_ip:port', 'https': 'https://user:pass@proxy_ip:port' } response = requests.post(url, proxies=proxies)- 请求失败重试:
from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1)) def safe_request(url): return requests.post(url, timeout=10)我曾因为请求太频繁被封IP,后来采用"慢启动"策略:开始时间隔长,逐渐缩短间隔,直到接近但不超过网站的响应极限。这种方法在长期运行的爬虫中特别有效。
8. 数据清洗与质量检查
原始数据往往存在各种问题,良好的清洗流程能提升数据价值。针对药监局数据,我通常会做以下处理:
- 字段标准化:
def clean_license_number(raw): return raw.strip().replace(' ', '').upper()- 空值处理:
details = [{ 'name': x.get('EPS_NAME', '').strip() or None, 'license': clean_license_number(x.get('businessLicenseNumber', '')) } for x in company_details]- 数据去重:
import pandas as pd df = pd.DataFrame(company_details).drop_duplicates('ID')- 有效性验证:
valid_companies = [x for x in company_details if x.get('businessLicenseNumber') and len(x['businessLicenseNumber']) >= 10]数据清洗最容易被忽视的是一致性检查。有次我发现同一家企业的许可证号在不同页面居然不同,后来发现是网站数据问题。现在我会对关键字段做交叉验证,确保数据准确可靠。
9. 完整项目架构建议
对于生产级爬虫项目,我推荐采用如下架构:
pharma_crawler/ ├── crawler/ # 核心爬取逻辑 │ ├── api.py # 接口请求封装 │ ├── parser.py # 数据解析 │ └── storage.py # 数据存储 ├── config/ # 配置管理 │ ├── settings.py # 全局配置 │ └── proxies.txt # 代理列表 ├── utils/ # 工具函数 │ ├── logger.py # 日志配置 │ └── helper.py # 通用函数 ├── main.py # 入口文件 └── requirements.txt # 依赖清单关键组件实现示例:
# logger.py import logging def setup_logger(name): logger = logging.getLogger(name) logger.setLevel(logging.INFO) handler = logging.FileHandler('crawler.log') formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) return logger# api.py from tenacity import retry from utils.logger import setup_logger logger = setup_logger('api') @retry(stop=stop_after_attempt(3)) def fetch_data(url, params, headers): try: response = requests.post(url, data=params, headers=headers) response.raise_for_status() return response.json() except Exception as e: logger.error(f"请求失败: {url} - {str(e)}") raise这种架构的好处是:
- 功能模块化,便于维护
- 配置与代码分离
- 完善的日志记录
- 易于扩展新功能
在大型爬虫项目中,我还推荐添加:
- 任务调度系统(如APScheduler)
- 监控告警机制
- 数据质量检测模块
- 断点续爬功能
10. 常见问题排查指南
即使经验丰富,爬虫开发中仍会遇到各种问题。以下是典型问题及解决方案:
问题1:返回空数据
- 检查请求头是否完整
- 验证POST参数是否齐全
- 尝试用浏览器开发者工具捕获的原始请求对比
问题2:连接超时
- 增加timeout值
- 检查网络代理设置
- 尝试不同网络环境
问题3:JSON解析失败
- 打印原始响应文本检查格式
- 处理非标准JSON(如BOM头)
- 添加异常捕获
问题4:突然无法抓取
- 检查网站是否改版
- 验证是否有验证码
- 降低请求频率
我习惯准备一个"急救包"脚本,包含各种调试功能:
def debug_request(url, params, headers): print("=== 请求参数 ===") print(f"URL: {url}") print(f"Headers: {json.dumps(headers, indent=2)}") print(f"Body: {params}") with requests.Session() as s: resp = s.post(url, data=params, headers=headers) print("\n=== 响应信息 ===") print(f"Status: {resp.status_code}") print(f"Headers: {dict(resp.headers)}") try: print(f"JSON: {resp.json()}") except: print(f"Raw: {resp.text[:500]}...") return resp这个脚本能快速定位90%的请求问题。记得保存各种异常响应的样本,建立自己的"错误案例库",这对调试非常有帮助。