Python爬虫实战：5分钟搞定化妆品生产许可证数据抓取（含Ajax动态加载解析）
Python爬虫实战：5分钟搞定化妆品生产许可证数据抓取（含Ajax动态加载解析）
最近在帮一个做美妆行业分析的朋友整理数据，他需要批量获取国内化妆品生产企业的许可证信息。这类数据通常由官方机构公开，但网站往往没有提供便捷的批量下载入口，一页页手动复制显然不现实。这让我想起了几年前自己刚学爬虫时，面对那些“点击下一页才加载内容”的网站，总是一头雾水，不知道数据到底藏在哪里。
实际上，现代网页大量采用Ajax技术动态加载数据，页面的URL不变，内容却通过后台接口悄悄更新。对于数据采集者来说，这既是挑战也是机遇。挑战在于，你不能再用简单的requests.get(页面URL)拿到所有HTML；机遇在于，一旦找到那个“悄悄”传输数据的接口，你就能以更高效、更结构化的方式获取数据，往往还是干净的JSON格式，省去了从HTML中解析的麻烦。
今天，我们就以“国家药品监督管理局”下属的化妆品生产许可信息平台为例，手把手带你走通一个完整的Ajax动态数据抓取流程。整个过程的核心，不是写复杂的代码，而是学会像侦探一样，用浏览器的开发者工具找到数据的真实来源。只要你跟下来，以后遇到类似的动态网站，基本都能举一反三。
1. 侦察阶段：从网页表象到数据接口
动手写代码之前，最关键的一步是搞清楚数据从哪里来。很多新手会直接对看到的网页地址发起请求，结果拿到的HTML里空空如也，这是因为数据是后来通过JavaScript动态填充的。
1.1 识别动态加载
打开目标网站，映入眼帘的是一个企业列表。如果你右键“查看网页源代码”，会发现源码里根本没有这些企业的名字和链接。这是一个典型的动态加载信号。
提示：区分静态和动态加载的一个快速方法是，对比“网页源代码”（
Ctrl+U）和“检查元素”（F12）中Elements标签页的内容。如果后者有数据而前者没有，那数据几乎肯定是动态加载的。
接下来，我们打开浏览器的开发者工具（F12），切换到 Network（网络） 标签页。刷新页面，你会看到一系列网络请求。我们的目标是找到那个真正携带了列表数据的请求。
- 在请求列表中，寻找类型为
XHR或Fetch的请求。这些通常是Ajax请求。 - 在列表页面上进行翻页操作，同时观察Network面板。哪个请求在每次翻页时都会出现，并且其
Response（响应）里包含了新的企业数据，哪个就是我们的目标。
通过这个步骤，我找到了一个关键的请求：
POST http://125.35.6.84:81/xk/itownet/portalAction.do?method=getXkzsList
它的响应是一个JSON字符串，里面包含了企业ID、名称等关键信息。这就是列表数据的源头。
1.2 分析请求详情
点击这个请求，查看它的详细信息，我们需要复制几样东西来模拟浏览器发送请求：
- Headers（请求头）： 特别是
User-Agent，用来伪装成浏览器，避免被简单的反爬机制拦截。 - Payload（负载） 或 Request Body（请求体）： 对于POST请求，这里包含了发送给服务器的参数，比如页码(
page)、每页数量(pageSize)等。
一个典型的请求参数可能如下所示：
{
"on": "true",
"page": "1",
"pageSize": "15",
"productName": "",
"conditionType": "1",
"applyname": "",
"applysn": ""
}
至此，侦察工作完成。我们知道了：
- 数据是通过一个特定的POST接口动态获取的。
- 获取列表需要模拟发送带有分页参数的请求。
- 从列表的JSON响应中，我们可以提取每个企业的唯一ID。
2. 深入挖掘：获取详情数据
列表数据拿到了，但通常我们还需要每个企业的详细许可证信息。点击列表中的企业名称，会跳转到详情页。同样，我们需要用老办法分析这个详情页的数据来源。
2.1 定位详情接口
进入某个企业的详情页，再次打开开发者工具的Network面板，刷新或直接查看加载出的请求。寻找另一个携带了详细数据的XHR请求。
很快，我发现了第二个关键接口：
POST http://125.35.6.84:81/xk/itownet/portalAction.do?method=getXkzsById
它的请求体非常简单，只有一个参数：
{
"id": "ff83aff95c5541cdab5ca6e847514f88"
这个id值，正是我们从列表接口的JSON响应中获取的企业ID。响应同样是一个结构清晰的JSON，包含了企业名称、许可证编号、生产地址、许可项目等所有我们需要的信息。
2.2 构建抓取逻辑链条
现在，整个数据流的逻辑链条就非常清晰了：
- 抓列表： 循环请求列表接口，改变
page参数，获取所有页的企业ID。 - 取详情： 用上一步获取的每一个企业ID，作为参数去请求详情接口。
- 存数据： 将详情接口返回的JSON数据保存下来。
这个“列表-详情”的模式在数据抓取中非常常见，掌握了它，你就攻克了一大类动态网站。
3. 实战编码：从零构建爬虫脚本
理论清晰了，我们开始用Python的requests库将整个过程自动化。我会将代码分成几个函数，让结构更清晰，也便于理解和维护。
3.1 环境准备与基础配置
首先，确保你安装了requests库。如果没有，在命令行执行 pip install requests。
我们创建一个新的Python文件，比如叫 cosmetic_license_spider.py。开头先进行一些基础配置：
import requests
import json
import time
from typing import List, Dict
# 基础配置
BASE_URL = 'http://125.35.6.84:81/xk'
LIST_API = f'{BASE_URL}/itownet/portalAction.do?method=getXkzsList'
DETAIL_API = f'{BASE_URL}/itownet/portalAction.do?method=getXkzsById'
# 请求头，用于模拟浏览器
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Host': '125.35.6.84:81',
'Origin': BASE_URL,
'Referer': f'{BASE_URL}/xk/',
}
这里我定义了基础URL和两个API接口地址。HEADERS中的信息是从浏览器开发者工具中复制并整理过来的，Referer和Origin有时是服务端用于验证请求来源的，加上会更稳妥。
3.2 高效获取企业ID列表
接下来，我们编写获取列表页数据的函数。这个函数需要处理分页。
def fetch_company_ids(start_page: int = 1, end_page: int = 5) -> List[str]:
"""
从列表接口分页获取所有企业的ID
Args:
start_page: 起始页码
end_page: 结束页码
Returns:
企业ID的列表
"""
all_ids = []
for page in range(start_page, end_page + 1):
print(f'正在抓取第 {page} 页列表...')
# 构建POST请求的数据体
post_data = {
'on': 'true',
'page': str(page),
'pageSize': '15', # 每页数量，可根据需要调整
'productName': '',
'conditionType': '1',
'applyname': '',
'applysn': '',
}
try:
# 发送POST请求
response = requests.post(LIST_API, data=post_data, headers=HEADERS, timeout=10)
response.raise_for_status() # 如果状态码不是200，抛出异常
# 解析JSON响应
result = response.json()
if result.get('list'):
# 从每一条记录中提取ID字段
page_ids = [item['ID'] for item in result['list']]
all_ids.extend(page_ids)
print(f' 第 {page} 页获取到 {len(page_ids)} 个ID。')
else:
print(f' 第 {page} 页无数据或数据结构异常。')
except requests.exceptions.RequestException as e:
print(f'请求第 {page} 页时发生错误: {e}')
except json.JSONDecodeError as e:
print(f'解析第 {page} 页JSON响应时发生错误: {e}')
# 礼貌性延时，避免请求过快给服务器带来压力
time.sleep(1)
print(f'列表抓取完成，共获取到 {len(all_ids)} 个企业ID。')
return all_ids
这个函数有几个关键点：
- 错误处理： 使用
try...except包裹网络请求和JSON解析，让程序更健壮。 - 数据提取： 确认响应的JSON结构后，用列表推导式高效提取ID。
- 延时：
time.sleep(1)在每次请求后暂停1秒，这是遵守网络礼仪、避免被封IP的基本操作。
3.3 批量抓取企业详情数据
拿到ID列表后，我们就可以遍历它，去详情接口获取完整数据了。
def fetch_company_details(company_ids: List[str]) -> List[Dict]:
"""
根据企业ID列表，批量获取详情数据
Args:
company_ids: 企业ID列表
Returns:
企业详情字典的列表
"""
all_details = []
total = len(company_ids)
for idx, company_id in enumerate(company_ids, 1):
print(f'正在抓取企业详情 ({idx}/{total}): ID={company_id}')
post_data = {'id': company_id}
try:
response = requests.post(DETAIL_API, data=post_data, headers=HEADERS, timeout=10)
response.raise_for_status()
detail_info = response.json()
all_details.append(detail_info)
except requests.exceptions.RequestException as e:
print(f' 请求ID {company_id} 详情时发生错误: {e}')
# 可以选择记录失败的ID，后续重试
except json.JSONDecodeError as e:
print(f' 解析ID {company_id} 的JSON响应时发生错误: {e}')
# 详情请求也加入延时
time.sleep(0.5)
print(f'详情抓取完成，共获取 {len(all_details)} 条详情数据。')
return all_details
这个函数遍历ID列表，为每个ID构造请求。同样加入了错误处理和延时。enumerate函数用来显示当前进度，在处理大量数据时非常有用。
3.4 数据存储与主流程
数据抓取回来后，我们需要将其持久化保存。JSON格式是个不错的选择，它易于阅读，也方便后续用Python或其他工具处理。
def save_to_json(data: List[Dict], filename: str = 'cosmetic_licenses.json'):
"""将数据保存为JSON文件"""
try:
with open(filename, 'w', encoding='utf-8') as f:
# ensure_ascii=False 确保中文正常显示
# indent=2 让JSON文件有缩进，更美观
json.dump(data, f, ensure_ascii=False, indent=2)
print(f'数据已成功保存至文件: {filename}')
except IOError as e:
print(f'保存文件时发生错误: {e}')
def main():
"""主函数，串联整个抓取流程"""
print("开始化妆品生产许可证数据抓取任务...")
# 步骤1: 获取企业ID列表
company_ids = fetch_company_ids(start_page=1, end_page=3) # 先抓3页作为演示
if not company_ids:
print("未获取到任何企业ID，程序退出。")
return
# 步骤2: 根据ID获取详情
all_details = fetch_company_details(company_ids)
# 步骤3: 保存数据
if all_details:
save_to_json(all_details)
# 简单预览一下第一条数据
print("\n数据预览（第一条记录的部分字段）:")
sample = all_details[0]
preview_fields = ['epsName', 'productSn', 'xkzh', 'qfManagerName']
for field in preview_fields:
print(f" {field}: {sample.get(field, 'N/A')}")
else:
print("未获取到任何详情数据。")
print("抓取任务结束。")
if __name__ == '__main__':
main()
主函数main()清晰地展示了整个工作流。保存数据时使用了json.dump的indent参数，让生成的JSON文件不是紧凑的一行，而是有缩进格式，便于人工查阅。
4. 进阶技巧与避坑指南
基本的爬虫跑起来了，但在实际项目中，你可能会遇到更多问题。下面分享几个进阶技巧和常见坑点。
4.1 处理请求头与反爬策略
我们之前只设置了最基本的User-Agent。有些网站会检查更多的请求头。一个更完整的、模拟Chrome浏览器的请求头可以这样设置：
ADVANCED_HEADERS = {
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Encoding': 'gzip, deflate',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'X-Requested-With': 'XMLHttpRequest', # 表明这是一个Ajax请求
'Host': '125.35.6.84:81',
'Origin': 'http://125.35.6.84:81',
'Referer': 'http://125.35.6.84:81/xk/',
}
其中X-Requested-With: XMLHttpRequest是标识Ajax请求的关键头。如果遇到请求失败，可以尝试将这些头信息全部复制过来。
4.2 使用Session保持会话
如果网站需要登录或使用会话（Session），requests.Session()可以帮你自动管理cookies，让多次请求处于同一个会话中。
def fetch_with_session():
"""使用Session对象进行请求示例"""
session = requests.Session()
session.headers.update(ADVANCED_HEADERS)
# 第一个请求（如果需要登录或初始化会话）
# init_resp = session.get(LOGIN_URL)
# 后续的请求会自动携带cookies
list_resp = session.post(LIST_API, data={'page':'1', 'pageSize':'15'})
# ... 处理响应
对于我们这个案例网站，可能不需要Session，但知道这个工具在应对复杂场景时很有用。
4.3 应对频率限制与IP封禁
大规模抓取时，最常遇到的就是访问频率限制。除了简单的time.sleep，还有更智能的策略：
- 随机延时： 让请求间隔时间不那么规律。
import random time.sleep(random.uniform(0.5, 2.0)) # 随机休眠0.5到2秒 - 代理IP池： 当单个IP被封锁时，轮换使用不同的代理IP是终极解决方案。你可以使用一些付费或免费的代理服务。
PROXIES = { 'http': 'http://your-proxy-ip:port', 'https': 'https://your-proxy-ip:port', } response = requests.post(API_URL, data=post_data, headers=HEADERS, proxies=PROXIES) - 捕获异常并重试： 使用
tenacity或retrying库，为请求添加重试机制，当遇到网络波动或短暂封锁时自动重试几次。
4.4 数据清洗与结构化存储
我们保存的是原始的JSON，但分析时可能需要更结构化的数据，比如存入CSV或数据库。这里演示如何将JSON数据转换为CSV：
import csv
def json_to_csv(json_filename, csv_filename):
"""将JSON文件转换为CSV文件"""
with open(json_filename, 'r', encoding='utf-8') as f:
data = json.load(f)
if not data:
return
# 获取所有可能的字段（取第一条数据的键）
fieldnames = set()
for item in data:
fieldnames.update(item.keys())
fieldnames = list(fieldnames)
with open(csv_filename, 'w', newline='', encoding='utf-8-sig') as f: # utf-8-sig支持Excel中文
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(data)
print(f'CSV文件已生成: {csv_filename}')
这个函数会遍历所有JSON条目，收集所有出现过的字段名作为CSV的表头，确保不遗漏任何信息。
4.5 调试与日志记录
将print语句替换为更专业的日志记录，可以方便地控制输出级别（调试、信息、错误），并将日志保存到文件。
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('spider.log'),
logging.StreamHandler() # 同时输出到控制台
]
)
logger = logging.getLogger(__name__)
# 在代码中使用
logger.info(f'正在抓取第 {page} 页列表...')
logger.error(f'请求发生错误: {e}')
整个流程走下来，你会发现核心的代码其实并不多，大部分工作是在分析和调试。我习惯把这种爬虫脚本保存在一个专门的项目文件夹里，并附上一个README.md文件，记录目标网站、接口分析过程、关键参数以及脚本的使用方法。这样即使过了几个月再回头看，或者分享给同事，也能立刻上手。数据抓取就像解谜，找到那个正确的接口，一切就水到渠成了。下次遇到动态网站，不妨先打开开发者工具的Network标签页看看，惊喜往往就藏在那些XHR请求里。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
10
0- 0
已为社区贡献7条内容
所有评论(0)