当前位置：首页 > news >正文

Python + Requests库爬取动态Ajax分页数据

news 2025/7/31 12:05:49

引言

在当今的互联网环境中，许多网站采用Ajax（Asynchronous JavaScript and XML）技术动态加载数据，以提高用户体验。传统的爬虫方法（如直接解析HTML）无法获取这些动态生成的内容，因此需要分析Ajax请求，模拟浏览器发送HTTP请求来获取数据。

本文将介绍如何使用Python + Requests库爬取动态Ajax分页数据，包括：

分析Ajax请求，找到数据接口
模拟请求参数，构造翻页逻辑
解析返回数据（通常是JSON格式）
存储数据（如CSV或数据库）

我们将以某电商网站（模拟案例）为例，演示如何爬取分页商品数据。

1. 分析Ajax请求

1.1 目标网站分析

假设目标网站的商品列表采用Ajax动态加载，URL结构如下：

https://example.com/api/products?page=1&size=10

**page**：当前页码
**size**：每页数据量

1.2 使用浏览器开发者工具

打开Chrome/Firefox开发者工具（F12）
进入Network（网络）选项卡
选择XHR（Ajax请求）
翻页时观察新增的请求，找到数据接口

https://example.com/ajax-analysis.png

1.3 确定请求参数

观察请求的：

URL（是否包含页码参数）
Headers（是否需要**User-Agent**、**Referer**等）
请求方式（GET/POST）
返回数据格式（通常是JSON）

2. Python + Requests 实现爬取

2.1 安装依赖库

2.2 构造请求函数

import requests
import pandas as pddef fetch_ajax_data(page):url = "https://example.com/api/products"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","Referer": "https://example.com/products",}params = {"page": page,"size": 10,}response = requests.get(url, headers=headers, params=params)if response.status_code == 200:return response.json()  # 假设返回的是JSON数据else:print(f"请求失败，状态码：{response.status_code}")return None

2.3 解析数据并翻页

def crawl_ajax_pages(max_pages=5):all_products = []for page in range(1, max_pages + 1):print(f"正在爬取第 {page} 页...")data = fetch_ajax_data(page)if data and "products" in data:all_products.extend(data["products"])else:print(f"第 {page} 页无数据或解析失败")return all_products

2.4 存储数据（CSV）

def save_to_csv(data, filename="products.csv"):df = pd.DataFrame(data)df.to_csv(filename, index=False, encoding="utf-8-sig")print(f"数据已保存至 {filename}")# 执行爬取
if __name__ == "__main__":products = crawl_ajax_pages(max_pages=5)if products:save_to_csv(products)

3. 进阶优化

3.1 处理反爬机制

随机User-Agent：防止被识别为爬虫
请求间隔：避免被封IP
代理IP：应对IP限制

import time
from fake_useragent import UserAgentdef fetch_ajax_data_safe(page):ua = UserAgent()headers = {"User-Agent": ua.random,"Referer": "https://example.com/products",}time.sleep(1)  # 避免请求过快# 其余代码同上...

3.2 异常处理

try:response = requests.get(url, headers=headers, params=params, timeout=10)response.raise_for_status()  # 检查HTTP错误
except requests.exceptions.RequestException as e:print(f"请求异常：{e}")return None

3.3 多线程/异步爬取（提高效率）

import concurrent.futuresdef crawl_with_threads(max_pages=5, workers=3):with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:futures = [executor.submit(fetch_ajax_data, page) for page in range(1, max_pages + 1)]all_products = []for future in concurrent.futures.as_completed(futures):data = future.result()if data:all_products.extend(data.get("products", []))return all_products

4. 完整代码示例

import requests
import pandas as pd
import time
from fake_useragent import UserAgent# 代理服务器配置
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"# 构造代理字典
proxies = {"http": f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}","https": f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}",
}def fetch_ajax_data(page):ua = UserAgent()url = "https://example.com/api/products"headers = {"User-Agent": ua.random,"Referer": "https://example.com/products",}params = {"page": page,"size": 10,}try:time.sleep(1)  # 防止请求过快# 添加proxies参数使用代理response = requests.get(url,headers=headers,params=params,timeout=10,proxies=proxies)response.raise_for_status()return response.json()except requests.exceptions.RequestException as e:print(f"第 {page} 页请求失败：{e}")return Nonedef crawl_ajax_pages(max_pages=5):all_products = []for page in range(1, max_pages + 1):print(f"正在爬取第 {page} 页...")data = fetch_ajax_data(page)if data and "products" in data:all_products.extend(data["products"])else:print(f"第 {page} 页无数据或解析失败")return all_productsdef save_to_csv(data, filename="products.csv"):df = pd.DataFrame(data)df.to_csv(filename, index=False, encoding="utf-8-sig")print(f"数据已保存至 {filename}")if __name__ == "__main__":products = crawl_ajax_pages(max_pages=5)if products:save_to_csv(products)