Python爬虫实战：POST请求与JSON解析攻克药监局动态数据

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?methodgetXkzsList通过检查这个请求的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?methodgetXkzsList headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Content-Type: application/x-www-form-urlencoded; charsetUTF-8 } data { on: true, page: 1, # 当前页码 pageSize: 15, # 每页条数 productName: , conditionType: 1, applyname: , applysn: } response requests.post(url, headersheaders, datadata)这里有几个容易踩坑的地方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(fJSON解析失败: {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( urlbase_url ?methodgetXkzsList, headersheaders, datadata, timeout10 ) 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, encodingutf-8) as f: json.dump(all_data, f, ensure_asciiFalse, indent2)这个流程中我特别加入了几个实用技巧超时设置避免某个请求卡死整个程序异常恢复单页失败不影响整体任务速率控制适当sleep避免被封IP数据校验检查关键字段确保数据质量存储时使用ensure_asciiFalse可以完美保存中文indent参数让生成的JSON文件更易读。记得有次我忘了设置编码打开文件全是\uXXXX的Unicode转义符排查了半天才发现问题。6. 企业详情数据的抓取策略获取企业列表只是第一步每个企业的详细信息才是价值所在。通过分析我们发现详情页也是动态加载的其URL模式为http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?methodgetXkzsByIdid企业ID基于之前获取的企业ID列表我们可以批量抓取详情数据detail_url http://scxk.nmpa.gov.cn:81/xk/itownet/portalAction.do?methodgetXkzsById company_details [] for company in all_data[:50]: # 示例只处理前50家 params {id: company[ID]} try: resp requests.post(detail_url, headersheaders, dataparams) 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;q0.9, }请求频率控制随机延迟更自然import random time.sleep(random.uniform(0.5, 2))代理IP池应对IP封锁proxies { http: http://user:passproxy_ip:port, https: https://user:passproxy_ip:port } response requests.post(url, proxiesproxies)请求失败重试from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1)) def safe_request(url): return requests.post(url, timeout10)我曾因为请求太频繁被封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(stopstop_after_attempt(3)) def fetch_data(url, params, headers): try: response requests.post(url, dataparams, headersheaders) response.raise_for_status() return response.json() except Exception as e: logger.error(f请求失败: {url} - {str(e)}) raise这种架构的好处是功能模块化便于维护配置与代码分离完善的日志记录易于扩展新功能在大型爬虫项目中我还推荐添加任务调度系统如APScheduler监控告警机制数据质量检测模块断点续爬功能10. 常见问题排查指南即使经验丰富爬虫开发中仍会遇到各种问题。以下是典型问题及解决方案问题1返回空数据检查请求头是否完整验证POST参数是否齐全尝试用浏览器开发者工具捕获的原始请求对比问题2连接超时增加timeout值检查网络代理设置尝试不同网络环境问题3JSON解析失败打印原始响应文本检查格式处理非标准JSON如BOM头添加异常捕获问题4突然无法抓取检查网站是否改版验证是否有验证码降低请求频率我习惯准备一个急救包脚本包含各种调试功能def debug_request(url, params, headers): print( 请求参数 ) print(fURL: {url}) print(fHeaders: {json.dumps(headers, indent2)}) print(fBody: {params}) with requests.Session() as s: resp s.post(url, dataparams, headersheaders) print(\n 响应信息 ) print(fStatus: {resp.status_code}) print(fHeaders: {dict(resp.headers)}) try: print(fJSON: {resp.json()}) except: print(fRaw: {resp.text[:500]}...) return resp这个脚本能快速定位90%的请求问题。记得保存各种异常响应的样本建立自己的错误案例库这对调试非常有帮助。