Python 爬虫实战：抓取 XHR/Fetch 异步加载的动态数据

传统静态网页：
浏览器请求 → 服务器返回完整HTML（含数据） → 渲染显示

动态网页（AJAX）：
浏览器请求 → 服务器返回空HTML骨架
          → 浏览器执行JS → JS发起XHR/Fetch请求API
          → 服务器返回JSON数据 → JS渲染到页面

"""
Python 爬虫：抓取 XHR/Fetch 异步加载数据
核心思路：分析浏览器 Network 面板中的 XHR 请求，
         用 requests 直接模拟 API 调用，获取 JSON 数据。
"""

import requests
import csv
import time
import random
from typing import Optional


class AjaxSpider:
    """异步数据爬虫"""

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/120.0.0.0 Safari/537.36"
            ),
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            # 如果接口需要 Referer 校验，取消下面注释并填入真实值
            # "Referer": "https://example.com/list",
            # 如果接口需要登录态，取消下面注释并填入真实 Cookie
            # "Cookie": "your_cookie_here",
        })

    def fetch_page(self, page: int, page_size: int = 20) -> Optional[dict]:
        """
        请求单页数据（GET 方式）

        实际使用时，将 url 和 params 替换为从浏览器
        Network → XHR 中捕获的真实接口地址和参数。
        """
        url = "https://wlt.east-car.com/posts"
        params = {
            "_page": page,
            "_limit": page_size,
        }
        try:
            resp = self.session.get(url, params=params, timeout=10)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            print(f"[错误] 第 {page} 页请求失败: {e}")
            return None

    def fetch_page_post(self, page: int, page_size: int = 20) -> Optional[dict]:
        """
        请求单页数据（POST 方式，JSON 传参）

        很多接口使用 POST + JSON body 传参，例如：
        {
            "page": 1,
            "pageSize": 20,
            "keyword": "python"
        }
        """
        url = "https://example.com/api/data"
        payload = {
            "page": page,
            "pageSize": page_size,
        }
        try:
            resp = self.session.post(
                url,
                json=payload,
                timeout=10
            )
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            print(f"[错误] 第 {page} 页请求失败: {e}")
            return None

    def crawl(self, max_pages: int = 5, page_size: int = 20) -> list:
        """
        爬取多页数据，返回所有数据的列表
        """
        all_items = []
        for page in range(1, max_pages + 1):
            print(f"正在抓取第 {page}/{max_pages} 页 ...")
            data = self.fetch_page(page, page_size)

            if data is None:
                print(f"  第 {page} 页获取为空，跳过")
                continue

            # jsonplaceholder 返回的是列表，直接追加
            # 如果接口返回的是 {"code": 0, "data": {"list": [...]}} 结构，
            # 需要根据实际结构调整，例如：
            #   items = data.get("data", {}).get("list", [])
            if isinstance(data, list):
                all_items.extend(data)
            elif isinstance(data, dict):
                # 常见嵌套结构，按需调整路径
                items = (
                    data.get("data")
                    or data.get("result")
                    or data.get("items")
                    or data.get("list")
                    or []
                )
                all_items.extend(items if isinstance(items, list) else [items])
            else:
                print(f"  第 {page} 页返回数据类型异常: {type(data)}")

            # 随机延时，模拟真人浏览节奏，降低被封风险
            time.sleep(random.uniform(0.5, 1.5))

        print(f"\n抓取完成，共获取 {len(all_items)} 条数据")
        return all_items

    def save_to_csv(self, items: list, filename: str = "output.csv") -> None:
        """
        将数据保存为 CSV 文件
        """
        if not items:
            print("没有数据可保存")
            return

        # 自动提取所有字段名作为表头
        fieldnames = list(items[0].keys())
        with open(filename, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(items)
        print(f"数据已保存至 {filename}")


def main():
    spider = AjaxSpider()

    # 抓取 5 页数据，每页 20 条
    items = spider.crawl(max_pages=5, page_size=20)

    # 打印前 3 条数据预览
    for i, item in enumerate(items[:3], 1):
        print(f"\n--- 第 {i} 条 ---")
        for key, value in item.items():
            print(f"  {key}: {value}")

    # 保存为 CSV
    spider.save_to_csv(items, "posts.csv")


if __name__ == "__main__":
    main()

from playwright.sync_api import sync_playwright
import json

def crawl_with_playwright(url: str):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page()
        # 拦截 XHR 响应，直接捕获 JSON 数据
        api_data = []
        def handle_response(response):
            if "/api/" in response.url and response.request.resource_type == "xhr":
                try:
                    api_data.append(response.json())
                except Exception:
                    pass
        page.on("response", handle_response)
        page.goto(url, wait_until="networkidle")
        browser.close()
    return api_data

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

def crawl_with_selenium(url: str):
    options = Options()
    options.add_argument("--headless")
    driver = webdriver.Chrome(options=options)
    driver.get(url)
    # 等待动态内容加载完成
    WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, ".data-item"))
    )
    html = driver.page_source  # 此时 HTML 已包含渲染后的数据
    driver.quit()
    return html