Python + Requests库爬取动态Ajax分页数据的完整流程_Python

引言

在当今的互联网环境中，许多网站采用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 pd
def 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 useragent
def 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.futures
def 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 none
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
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)