Click Event Simulation：无需浏览器触发动态数据加载

原创

jackcode

发布于 2025-03-03 16:02:30

2300

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

一、明确目标与前置知识

目标

使用 Python 模拟点击事件，直接发送 HTTP 请求采集拼多多上商品价格和优惠信息。
采用爬虫代理（代理IP）的技术，设置好 Cookie 和 User-Agent，以防止被目标网站屏蔽。
利用多线程技术加速数据采集，提高效率。

前置知识

基本的 Python 编程知识
HTTP 协议与请求头、Cookie 的概念
多线程编程基础（如线程、队列的使用）
代理IP的使用原理

二、按步骤拆解操作

1. 环境准备

在开始编写代码之前，请确保已安装以下 Python 库：

requests：用于发送 HTTP 请求
threading 与 queue：实现多线程任务调度

可使用以下命令安装必要的库（通常 Python 标准库中已自带 threading 和 queue）：

pip install requests

2. 配置代理IP、Cookie 和 User-Agent

在采集过程中，为了防止频繁访问被封，我们使用爬虫代理服务。以下是代理的参考配置（请将示例中的用户名、密码、域名、端口替换为实际的亿牛云提供的参数）。

import requests
import threading
from queue import Queue

# ------------------ 代理IP设置 ------------------
# 配置亿牛云爬虫代理（请替换为实际的代理信息）
proxy_user = "16YUN"       # 代理用户名
proxy_pass = "16IP"       # 代理密码
proxy_domain = "proxy.16yun.cn"  # 代理域名
proxy_port = "8080"             # 代理端口

proxies = {
    "http": f"http://{proxy_user}:{proxy_pass}@{proxy_domain}:{proxy_port}",
    "https": f"http://{proxy_user}:{proxy_pass}@{proxy_domain}:{proxy_port}"
}

# ------------------ 设置Cookie和User-Agent ------------------
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36",  # 模拟浏览器请求
    "Cookie": "your_cookie_here"  # 请将此处替换为实际的Cookie信息
}

3. 模拟点击事件加载数据（无需浏览器）

很多网站使用点击事件来动态加载数据，其实核心在于分析页面加载时实际发送的 HTTP 请求。针对拼多多，假设我们已经通过抓包工具找到了对应的接口（示例接口仅供参考），代码如下：

# ------------------ 定义目标URL及数据采集函数 ------------------
# 目标接口（示例），实际使用中需根据抓包工具确定真实接口
base_url = "https://apiv3.yangkeduo.com/..."

def fetch_product_data(page):
    """
    模拟点击加载数据，采集指定页码的商品信息
    :param page: 页码参数
    """
    # 构造请求URL，假设接口支持页码参数
    url = f"{base_url}?page={page}"
    try:
        # 发送 GET 请求，利用代理IP及设置的headers
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            # 解析返回的JSON数据（示例数据格式，实际可能不同）
            data = response.json()
            # 分析并输出商品价格与优惠信息
            for product in data.get("products", []):
                title = product.get("title")
                price = product.get("price")
                discount = product.get("discount")
                print(f"商品：{title}，价格：{price}，优惠：{discount}")
        else:
            print(f"页面{page}请求失败，状态码：{response.status_code}")
    except Exception as e:
        print(f"请求页面{page}时出现异常：{e}")

4. 实现多线程采集

为了提高采集效率，我们使用 threading 和 queue 实现多线程采集。将待采集页码放入队列中，由多个线程并发执行采集任务。

# ------------------ 多线程采集实现 ------------------
def worker():
    """
    线程工作函数，从队列中取出页码后调用数据采集函数
    """
    while True:
        page = q.get()
        if page is None:
            break  # 队列结束标志，退出线程
        fetch_product_data(page)
        q.task_done()

# 初始化任务队列
q = Queue()
num_threads = 5  # 线程数量，可根据实际情况调整
threads = []

# 启动工作线程
for i in range(num_threads):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)

# 向队列中添加页码任务（假设采集1到10页数据）
for page in range(1, 11):
    q.put(page)

q.join()  # 等待所有任务完成

# 发送结束信号，停止所有线程
for i in range(num_threads):
    q.put(None)
for t in threads:
    t.join()

完整代码如下：

import requests
import threading
from queue import Queue

# ------------------ 代理IP设置 ------------------
# 配置亿牛云爬虫代理（请替换为实际的代理信息）
proxy_user = "16YUN"       # 代理用户名
proxy_pass = "16IP"       # 代理密码
proxy_domain = "proxy.16yun.cn"  # 代理域名
proxy_port = "8080"             # 代理端口

proxies = {
    "http": f"http://{proxy_user}:{proxy_pass}@{proxy_domain}:{proxy_port}",
    "https": f"http://{proxy_user}:{proxy_pass}@{proxy_domain}:{proxy_port}"
}

# ------------------ 设置Cookie和User-Agent ------------------
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                  "(KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36",  # 模拟浏览器请求
    "Cookie": "your_cookie_here"  # 请将此处替换为实际的Cookie信息
}

# ------------------ 定义目标URL及数据采集函数 ------------------
# 目标接口（示例），实际使用中需根据抓包工具确定真实接口
base_url = "https://apiv3.yangkeduo.com/..."

def fetch_product_data(page):
    """
    模拟点击加载数据，采集指定页码的商品信息
    :param page: 页码参数
    """
    # 构造请求URL，假设接口支持页码参数
    url = f"{base_url}?page={page}"
    try:
        # 发送 GET 请求，利用代理IP及设置的headers
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            # 解析返回的JSON数据（示例数据格式，实际可能不同）
            data = response.json()
            # 分析并输出商品价格与优惠信息
            for product in data.get("products", []):
                title = product.get("title")
                price = product.get("price")
                discount = product.get("discount")
                print(f"商品：{title}，价格：{price}，优惠：{discount}")
        else:
            print(f"页面{page}请求失败，状态码：{response.status_code}")
    except Exception as e:
        print(f"请求页面{page}时出现异常：{e}")

# ------------------ 多线程采集实现 ------------------
def worker():
    """
    线程工作函数，从队列中取出页码后调用数据采集函数
    """
    while True:
        page = q.get()
        if page is None:
            break  # 收到结束信号，退出线程
        fetch_product_data(page)
        q.task_done()

# 初始化任务队列
q = Queue()
num_threads = 5  # 线程数量，可根据实际情况调整
threads = []

# 启动工作线程
for i in range(num_threads):
    t = threading.Thread(target=worker)
    t.start()
    threads.append(t)

# 向队列中添加页码任务（假设采集1到10页数据）
for page in range(1, 11):
    q.put(page)

q.join()  # 等待所有任务完成

# 发送结束信号，停止所有线程
for i in range(num_threads):
    q.put(None)
for t in threads:
    t.join()

运行以上代码后，你将看到控制台输出拼多多采集到的各个商品的价格与优惠信息。