列表抓取完全指南

从以列表形式显示信息的网站（产品目录、招聘信息、搜索结果）中提取数据非常棘手。网站会将项目分散到多个页面，将它们隐藏在 "加载更多 "按钮后面，或者使用无限滚动。如果没有正确的方法，您的数据集将是不完整的。

本指南涵盖列表抓取基础知识：如何使用 Python 和自动化工具可靠地提取结构化数据。

什么是列表抓取？

列表抓取可从以重复格式显示信息的网页中提取结构化数据。与针对单个页面的一般网络抓取不同，列表抓取侧重于从一个或多个页面中收集多个类似项目（产品卡片、职位列表、业务条目）。

列表抓取为何重要

现代网站以可预测的模式组织数据。每个产品卡片、职位列表或目录条目都遵循相同的 HTML 结构。这使得它们非常适合自动提取。困难在于，当所有项目都分布在分页、无限滚动或动态加载中时，如何访问它们。

列表抓取的关键组成部分

身份验证:查找容纳每个重复项目（产品卡、职位列表）的 HTML 容器。这决定了您可以捕获哪些数据。

提取:将每个容器内的特定字段（名称、价格、位置）映射到您的数据模式。一致的字段映射可确保干净利落的输出。

遍历:通过分页、无限滚动或 "加载更多 "按钮浏览所有项目。这可确保完整的数据收集。

实际应用

电子商务价格监控:通过提取产品名称、价格、评级和可用性状态，跟踪价格变化、库存可用性和竞争对手的产品。

就业市场分析:收集职位名称、薪资范围、所需技能、公司名称和地点 来自工地.将这些数据用于职位提醒、趋势分析和薪酬基准。

商业智能:从目录中提取联系信息、营业时间、客户评价和服务类别，用于潜在客户生成和竞争分析。

内容聚合:抓取新闻网站、博客和内容平台，以建立数据库、跟踪热门话题并监控特定主题。

识别列表类型

在编写代码之前，请先了解目标网站的列表结构。加载方式决定了抓取策略。

标准分页:每页有固定项目，并带有编号链接或 "下一页 "按钮。页码在 URL 中显示为 ?page=2 或 /page/2/.最容易爬行

加载更多按钮:相同的 URL，点击后添加新项目。要么自动点击按钮，要么直接拦截后台 API 请求。

无限滚动:当您向下滚动时，新商品会自动加载。常见于社交媒体和现代电子商务。需要模拟滚动操作或捕捉底层 API 调用。

搜索结果页面:结果取决于查询参数、筛选器和排序。您的爬虫必须在处理底层分页的同时管理这些参数。查看我的文章 最好的 SERP API允许您跳过搜索逻辑，直接通过应用程序接口检索结果。

选择抓取方法

方法 最适合 优点 缺点 HTTP 请求 解析静态 HTML，基于 URL 分页 快速、轻量级、简单 处理 JavaScript 渲染的内容失败 无头浏览器 需要 JavaScript 的动态内容 处理复杂的交互 资源繁重，难以扩展 企业 API 生产系统需要可靠性 受管理的基础设施，内置反僵尸处理 额外成本

方法 1：使用 BeautifulSoup 发送 HTTP 请求

当初始 HTML 响应中存在列表项时，可通过 HTTP 请求和 HTML 解析将其提取出来。

设置

pip install requests beautifulsoup4

基本列表履带

import csv
import time
from urllib.parse import URLjoin
import requests
from bs4 import 美丽汤
BASE_URL = "https://books.toscrape.com/"

def 从页面提取项目(soup, base_url):
    ""从一页中提取所有项目。""
    项目 = []
    
    for 贺卡 in soup.select("article.product_pod"):
        link_el = card.select_one("h3 a")
        price_el = card.select_one("p.price_color")
        rating_el = card.select_one("p.star-rating")
        
        if not link_el 或 not 价格
            continue
        
        项目 = {
            "title": link_el.get("title", "").strip()、
            "detail_url": urljoin(base_url, link_el.get("href", "")),
            "price": price_el.get_text(strip=True),
            "评级": extract_rating(rating_el)
        }
        items.append(item)
    
    return 项目

def 提取评级(评级元素):
    ""从 CSS 类别中提取评级"""
    if not rating_element：
        return ""
    classes = rating_element.get("class", [])
    return " ".join(c for c in 班级 if c != "星级")

def 查找下一页(汤，current_url):
    """查找并返回下一页 URL。""
    next_link = soup.select_one("li.next a")
    if next_link：
        return urljoin(current_url, next_link.get("href", ""))
    return None

def 抓取书籍列表(max_pages=5):
    ""主要爬行功能"""
    url = BASE_URL
    all_items = []
    页数 = 1
    
    while url 和 page_count <= max_pages：
        print(f "抓取页面 {页数｝: {url}")
        
        response = requests.get(url, timeout=15)
        response.raise_for_status()
        
        soup = BeautifulSoup（response.text、 "html.parser")
        items = extract_items_from_page(soup, url)
        print(发现 {len(items)} 页上的项目 {页数｝")
        
        if not items:
            print("未找到物品，停止爬行")
            break
        
        all_items.extend(items)
        url = find_next_page(soup, url)
        page_count = 1
        
        time.sleep(1.5)
    
    print(f "收集的物品总数： {len(all_items)}")
    return 所有项目

def save_results(items, filename="books_catalog.csv"):
    ""将提取的项目保存为 CSV。
    if not items:
        print("无项目可保存")
        return
    
    with open（文件名： "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=items[)0].keys())
        writer.writeheader()
        writer.writerows(items)
    
    print(保存 {len(items)} 项目 {文件名｝")

if __name__ == "__main__":
    books = crawl_book_list(max_pages=5)
    save_results(books)

这段代码将关注点分开：一个函数提取项目，另一个函数查找下一页链接，主循环处理遍历。

方法 2：无头浏览器自动化

当 JavaScript 动态生成列表内容时，请使用真正的浏览器。Playwright 可让您很好地控制浏览器自动化。

设置

pip install playwright
playwright install chromium

处理无限滚动

import csv
import time
from playwright.sync_api import 同步播放器
TARGET_URL = "https://www.scrapingcourse.com/infinite-scrolling/"

def 滚动和等待(page, scroll_pause=2.0):
    ""滚动到底部，等待加载内容。
    page.evaluate("window.scrollTo（0，document.body.scrollHeight）")
    time.sleep(scroll_pause)

def 获取项目数(页面、选择器):
    ""计算当前可见的项目。""
    return page.locator(selector).count()

def 提取项目(page):
    ""从页面中提取所有可见项目。
    项目 = []
    
    for 贺卡 in page.locator(".产品-项目").all():
        项目 = {
            "name": card.locator(".产品名称").inner_text().strip()、
            "price": card.locator(".产品价格").inner_text().strip()、
            "url": card.locator("a").first.get_attribute("href") 或 ""
        }
        items.append(item)
    
    return 项目

def 无限滚动(max_scrolls=15):
    ""抓取无限滚动的列表"""
    with sync_playwright() as playwright：
        browser = playwright.chromium.launch(headless=True)
        page = browser.new_page()
        
        page.goto(TARGET_URL, wait_until="networkidle")
        page.wait_for_selector(".产品-项目", timeout=10000)
        
        previous_count = 0
        stagnant_scrolls = 0
        max_stagnant = 3
        
        for scroll_num in range(max_scrolls)：
            current_count = get_item_count(page、 ".产品-项目")
            print(滚动 {滚动次数   1}: {current_count} 项目可见")
            
            if current_count == previous_count：
                stagnant_scrolls = 1
                if stagnant_scrolls >= max_stagnant：
                    print("多次滚动后没有新项目，停止")
                    break
            else:
                stagnant_scrolls = 0
            
            previous_count = current_count
            scroll_and_wait(page)
        
        items = extract_items(page)
        browser.close()
        
        print(提取 {len(items)} 总项目")
        return 项目

def export_too_csv(items, filename="products.csv"):
    ""将项目导出到 CSV 文件。""
    if not items:
        print("没有要出口的项目")
        return
    
    with open（文件名： "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=items[)0].keys())
        writer.writeheader()
        writer.writerows(items)
    
    print(出口 {len(items)} 项目 {文件名｝")

if __name__ == "__main__":
    产品 = crawl_infinite_scroll(max_scrolls=15)
    export_too_csv(products)

当物品数量在三次滚动尝试中保持不变时，爬虫就会停止。这意味着你已经到达了列表的末尾。

方法 3：采用Bright Data的企业解决方案

对于需要规模和可靠性的生产系统，Bright Data 可提供可管理的基础设施，而无需运营工作。

选项 A：Web Scraper API

布赖特数据公司的 网络抓取 API 提供针对热门网站的预建刮擦工具。它能自动处理 JavaScript 渲染、反僵尸挑战和代理管理。

同步刮削 (多达 20 个 URL 的实时结果）：

import requests
def scrape_urls_sync(urls、数据集 ID、api_key):
    ""使用 Bright Data Web Scraper API 抓取 URL。
    终点 = "https://api.brightdata.com/datasets/v3/scrape"
    
    params = {
        "数据集 ID": dataset_id、
        "格式": "json"
    }
    
    headers = {
        "Authorization": f "承载器 {api_key}",
        "内容类型": "application/json"
    }
    
    有效载荷 = [{"url": url} for url in 网址]
    
    response = requests.post(
        endpoint、
        params=params、
        headers=headers、
        json=payload、
        timeout=60
    )
    response.raise_for_status()
    
    return response.json()

if __name__ == "__main__":
    API_KEY = "YOUR_API_KEY"
    DATASET_ID = "gd_YOUR_DATASET_ID"  # 从 Bright Data 仪表板获取
    
    urls = [
        "https://example.com/product/1",
        "https://example.com/product/2"
    ]
    
    results = scrape_urls_sync(urls、DATASET_ID、API_KEY)
    print(f "Scraped {len(results)} 项目")

异步抓取 (适用于大批量生产）：

import requests
import time
def 触发器_async_scrape(urls、数据集 ID、api_key):
    """启动异步搜刮任务。"""
    终点 = "https://api.brightdata.com/datasets/v3/trigger"
    
    params = {
        "数据集 ID": dataset_id、
        "格式": "json"
    }
    
    headers = {
        "Authorization": f "承载器 {api_key}",
        "内容类型": "application/json"
    }
    
    有效载荷 = [{"url": url} for url in 网址]
    
    response = requests.post(
        endpoint、
        params=params、
        headers=headers、
        json=payload
    )
    response.raise_for_status()
    
    return response.json()["快照 ID"]

def 检查快照状态(快照 ID、api_key):
    ""检查快照是否准备就绪。""
    终点 = f "https://api.brightdata.com/datasets/v3/snapshot/{snapshot_id}"
    
    headers = {"Authorization": f "承载器 {api_key}"}
    
    response = requests.get(endpoint, headers=headers)
    response.raise_for_status()
    
    return response.json()["状态"]

def 下载快照(快照 ID、api_key):
    ""下载完成的快照数据"""
    终点 = f "https://api.brightdata.com/datasets/v3/snapshot/{snapshot_id}"
    
    params = {"格式": "json"}
    headers = {"Authorization": f "承载器 {api_key}"}
    
    response = requests.get(endpoint, params=params, headers=headers)
    response.raise_for_status()
    
    return response.json()

def scrape_urls_async(urls、dataset_id、api_key、poll_interval=10):
    ""通过轮询抓取 URL。""
    snapshot_id = trigger_async_scrape(urls, dataset_id, api_key)
    print(f "任务开始。快照 ID： {snapshot_id}")
    
    while True:
        status = check_snapshot_status(snapshot_id, api_key)
        print(f "状态： {状态｝")
        
        if 状态 == "准备就绪":
            return download_snapshot(snapshot_id, api_key)
        如果 状态 == "失败":
            raise Exception("搜索任务失败")
        
        time.sleep(poll_interval)

if __name__ == "__main__":
    API_KEY = "YOUR_API_KEY"
    DATASET_ID = "gd_YOUR_DATASET_ID"
    
    urls = ["https://example.com/product/"   str(i) for i in range(100)]
    
    results = scrape_urls_async(urls、DATASET_ID、API_KEY)
    print(f "Scraped {len(results)} 项目")

选项 B：浏览器应用程序接口

对于可完全控制浏览器的自定义刮擦，Bright Data 的 浏览器 API 提供内置解封功能的远程浏览器。可通过 Playwright 或 Puppeteer 连接：

import asyncio
from playwright.async_api import async_playwright
获取 = "brd-customer-CUSTOMER_ID-zone-ZONE_NAME:ZONE_PASSWORD"。
async def scrape_with_browser_api(url):
    ""使用 Playwright 的 Bright Data Browser API 进行抓取。
    async with async_playwright() as 剧作家：
        endpoint_url = f "wss://{@brd.superproxy.io">AUTH}@brd.superproxy.io:9222"。
        
        浏览器 = await playwright.chromium.connect_over_cdp(endpoint_url)
        
        try:
            页码 = await browser.new_page()
            await page.goto(url, timeout=120000)
            
            # 您的自定义提取逻辑
            项目 = []
            for 贺卡 in await page.locator(".product-item").all():
                item = {
                    "name"： await card.locator(".name").inner_text()、
                    价格 await card.locator(".price").inner_text()
                }
                items.append(item)
            
            返回项目
        最后
            await browser.close()

if __name__ == "__主____":
    results = asyncio.run(scrape_with_browser_api("https://example.com/products"))
    打印（f"提取 {len(results)} 项目")

选项 C：解锁程序应用程序接口

对于只需解锁 HTML 的简单情况，可使用 解锁程序应用程序接口.它能返回干净的 HTML/JSON，同时自动处理代理、验证码和指纹识别：

import requests
from bs4 import BeautifulSoup
def 带解锁程序的提取(url、api_key、zone_name):
    ""使用 Bright Data Unlocker API 获取 URL 内容。
    终点 = "https://api.brightdata.com/request"
    
    headers = {
        "Authorization": f "承载器 {api_key}",
        "内容类型": "application/json"
    }
    
    有效载荷 = {
        "区"区域名称、
        "url": url、
        "格式": "原始"
    }
    
    response = requests.post(endpoint, headers=headers, json=payload)
    response.raise_for_status()
    
    return response.text

def 带解锁程序的抓取(base_url, api_key, zone_name, max_pages=5):
    ""使用 Unlocker API 抓取分页列表。"""
    all_items = []
    
    for 页码 in range(1, max_pages   1):
        url = f"{base_url}?page={page_num}"
        print(f "获取页面 {page_num}")
        
        html = fetch_with_unlocker(url, api_key, zone_name)
        soup = BeautifulSoup(html、 "html.parser")
        
        项目 = []
        for 贺卡 in soup.select(".产品卡"):
            items.append({
                "name": card.select_one(".名称").get_text(strip=True),
                "price": card.select_one(".价格").get_text(strip=True)
            })
        
        if not items:
            break
        
        all_items.extend(items)
    
    return 所有项目

最佳做法

尊重费率限制:增加请求之间的延迟。对于大多数网站，1-2 秒即可。对于较大的抓取，可将请求时间分散。

处理错误:会出现网络问题和意外页面结构。请使用 try-except 块，并在保存数据前进行检查。

检查数据质量:定期查看提取的数据。缺少字段或截断列表意味着你的选择器或遍历逻辑有问题。

适应变化:网站经常更新 HTML 结构。使用易于更新的选择器构建爬网程序，并定期对其进行测试。

摘要

列表抓取可将分散的网络数据转化为结构化数据集。方法的选择取决于目标网站的技术和规模需求。

从静态列表的 HTTP 请求开始。需要 JavaScript 渲染时，使用无头浏览器。需要可靠性、扩展性和较少维护时，可考虑使用 Bright Data 等企业 API。

基本原理保持不变：识别重复模式、持续提取、完整遍历