scraper-builder

Scraper Builder

你正在为用户构建一个可用于生产环境的网页抓取器。你的任务是引导他们从“我想获取网站 X 的数据”开始，最终得到一个可正常运行、稳健可靠的抓取器，并能处理分页、动态内容、反爬保护和数据解析等现实挑战。

关键：始终验证输出

构建抓取器后，务必先用少量样本（1-3 页）运行它，并在扩大规模前向用户展示提取的数据。如果输出为空、格式错误或缺少字段，就继续迭代——修复选择器、切换 API 或调整解析逻辑。无法生成干净数据的抓取器就还没完成。

请在侦察阶段投入足够时间。提前花 2 分钟分析 HTML，可以避免之后花数小时调试。在这里，质量比速度更重要。

此技能的工作方式

此技能会协调 Bright Data 的四个 API，以智能方式构建抓取器。它不会编写脆弱的自定义抓取代码，而是先分析目标网站，再选择最可靠且最具成本效益的数据提取方法。决策流程如下：

1. 是否已有预构建抓取器？ → 使用 Web Scraper API（无需编写解析代码）
2. 页面是否为静态页面／不需要交互？ → 使用 Web Unlocker API（成本最低、最简单）
3. 页面是否需要点击、滚动或 JS 交互？ → 使用 Browser API（完整自动化）
4. 是否需要搜索引擎结果？ → 使用 SERP API

此技能会生成完整、可运行的代码，而不是伪代码或大纲。

---

阶段 1：了解目标

编写任何代码之前，你需要了解用户的需求和目标网站的情况。询问以下问题（用户已回答的问题可跳过）：

1. 哪个网站？ — 目标 URL 或域名
2. 需要哪些数据？ — 所需字段（产品名称、价格、评论等）
3. 抓取范围是什么？ — 单个页面、分类页面、搜索结果，还是整个网站板块？
4. 是否需要分页？ — 是否需要跨多个页面抓取？
5. 数据量有多大？ — 大约有多少条目／页面？（会影响同步或异步方式的选择以及并发策略，参见 references/concurrency-guide.md）
6. 输出格式是什么？ — JSON、CSV、数据库？（未指定时默认为 JSON）
7. 偏好哪种语言？ — Python 还是 Node.js？（未指定时默认为 Python）

不要问得太细。如果用户说“为 Amazon 商品页面构建一个抓取器”，你已经知道：网站是 Amazon，数据是商品详情，范围是商品页面。直接进入下一步。

---

阶段 2：检查预构建抓取器

开始任何自定义工作之前，先检查 Bright Data 是否已有适用于该域名的抓取器。这是最快、最便宜且最可靠的方案。

查看 references/supported-domains.md 中常见预构建抓取器的精选列表。但精选列表可能并不完整——Bright Data 支持 100 多个域名，并会定期添加新的抓取器。如果目标域名不在精选列表中，请查询实时 Dataset List API 进行确认：

curl -H "Authorization: Bearer $BRIGHTDATA_API_KEY" \
     https://api.brightdata.com/datasets/list

这会返回所有可用抓取器及其 dataset_id 和名称。在结果中搜索目标域名。你也可以浏览完整的文档索引 https://docs.brightdata.com/llms.txt，查找特定抓取器的文档和支持的参数。

如果存在预构建抓取器

使用 Web Scraper API 或 Python SDK 平台专用抓取器。这样无需编写解析代码，即可获取结构化 JSON。

Python SDK 方案（首选）：

from brightdata import BrightDataClient

async with BrightDataClient() as client:
    result = await client.scrape.amazon.products(url="https://amazon.com/dp/B0CRMZHDG8")
    if result.success:
        print(result.data)  # Structured product data

CLI 方案（Bright Data CLI 0.4.0+）：

bdata pipelines search "<site or data type>"      # find the dataset_id
bdata pipelines info <dataset_id> <method>        # check input fields
bdata pipelines run <dataset_id> <method> "<url>"

使用预构建抓取器进行批量抓取时，请采用异步触发／轮询／获取模式：

async with BrightDataClient() as client:
    # Trigger without waiting
    job = await client.scrape.amazon.products_trigger(url=url)
    # Poll until ready
    await job.wait(timeout=180, poll_interval=10, verbose=True)
    # Fetch results
    data = await job.fetch()

如果用户需要通过预构建抓取器抓取多个页面，请跳到阶段 5（分页／编排）。

如果没有预构建抓取器

继续进入阶段 3——你需要分析网站并构建自定义抓取器。

---

阶段 3：网站侦察

这是区分可靠爬虫和脆弱爬虫的关键步骤。编写提取代码之前，需要先了解网站的结构。

步骤 3a：获取页面 HTML

使用 Web Unlocker 获取原始 HTML。这样可以判断内容是由服务器渲染还是由客户端渲染，也能获取实际的 DOM 供你分析。

import requests
import os

API_KEY = os.environ["BRIGHTDATA_API_KEY"]
ZONE = os.environ["BRIGHTDATA_UNLOCKER_ZONE"]

response = requests.post(
    "https://api.brightdata.com/request",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "zone": ZONE,
        "url": "https://target-site.com/page",
        "format": "raw"
    }
)
html = response.text

或者使用抓取技能的 shell 脚本：

bash skills/scrape/scripts/scrape.sh "https://target-site.com/page"

步骤 3b：分析 HTML 结构

阅读 references/site-analysis-guide.md，了解详细的分析指南。

查看获取到的 HTML，并确定：

1. HTML 中是否包含内容？ 如果所需数据存在于原始 HTML 中，Web Unlocker 就足够了。如果 HTML 基本上只有空壳和 JS 框架标记（<div id="root"></div>、<div id="__next"></div>、ng-app），则内容由客户端渲染，需要使用 Browser API。

2. 识别可靠的选择器。 找到用于定位数据字段的 CSS 选择器或数据属性。优先级如下（从最可靠到最不可靠）：

   • data-* 属性（例如 [data-testid="product-price"]）——改版后仍然有效
   • 带有特定类名的语义化 HTML（例如 .product-card .price）
   • id 属性——唯一，但可能会变化
   • 结构选择器（例如 div > span:nth-child(2)）——脆弱，应避免

3. 识别数据模式。 页面属于哪种类型：

   • 列表页——重复结构中的多个项目（商品网格、搜索结果）
   • 详情页——单个项目包含多个字段（商品页、个人资料）
   • 分页页——结果分布在多个页面，并带有下一页／上一页控件
   • 无限滚动页——滚动时加载内容（需要 Browser API）
   • API 支持的页面——检查 Network 标签页中的请求模式；有些网站会从 JSON API 获取数据

4. 检查隐藏 API。 许多现代网站会通过 XHR／fetch 调用内部 API 加载数据。如果你在页面源代码或网络活动中发现结构化的 JSON 端点，通常直接通过 Web Unlocker 访问这些端点，比解析 HTML 更简洁。

步骤 3c：确定提取方案

根据分析结果：

发现	方案
内容在 HTML 中，无需交互	Web Unlocker——获取 HTML，并用 BeautifulSoup/Cheerio 解析
内容通过 JSON API 加载	Web Unlocker——直接访问 API 端点
内容需要 JS 渲染	Browser API——渲染后再提取
内容需要点击／滚动／交互	Browser API——自动执行交互
无限滚动分页	Browser API——滚动并收集内容
基于标准 URL 的分页	Web Unlocker——遍历页面 URL
CAPTCHA 较多的网站	Browser API——自动解决 CAPTCHA

---

阶段 4：构建提取器

现在编写实际的提取代码。方案取决于阶段 3 中的判断。

方案 A：Web Unlocker + HTML 解析

最适合静态网站或由服务器渲染 HTML 的网站。这是成本最低、速度最快的方案。

import requests
import os
from bs4 import BeautifulSoup

API_KEY = os.environ["BRIGHTDATA_API_KEY"]
ZONE = os.environ["BRIGHTDATA_UNLOCKER_ZONE"]

def fetch_page(url: str) -> str:
    """Fetch a page through Bright Data Web Unlocker."""
    response = requests.post(
        "https://api.brightdata.com/request",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"zone": ZONE, "url": url, "format": "raw"}
    )
    response.raise_for_status()
    return response.text

def parse_products(html: str) -> list[dict]:
    """Extract product data from HTML. Customize selectors per site."""
    soup = BeautifulSoup(html, "html.parser")
    products = []

    for card in soup.select(".product-card"):  # Adjust selector
        product = {
            "name": card.select_one(".product-title").get_text(strip=True),
            "price": card.select_one(".product-price").get_text(strip=True),
            "url": card.select_one("a")["href"],
            # Add more fields as needed
        }
        products.append(product)

    return products

# Usage
html = fetch_page("https://example.com/products")
products = parse_products(html)

稳健解析的关键模式：

• 始终使用 .get_text(strip=True) 清理空白字符
• 使用 .get("href", "") 而不是 ["href"]，以避免属性缺失时出现 KeyError
• 将各字段的提取分别包装在 try/except 中，避免单个异常项目导致整个抓取失败
• 单独执行价格标准化（去除货币符号并转换为浮点数）

方案 B：Web Unlocker + 直接提取 API 数据

如果发现网站通过 JSON API 端点加载数据，就直接访问该端点。这是最简洁的方案——无需解析 HTML。

import requests
import json
import os

API_KEY = os.environ["BRIGHTDATA_API_KEY"]
ZONE = os.environ["BRIGHTDATA_UNLOCKER_ZONE"]

def fetch_api(api_url: str) -> dict:
    """Fetch a JSON API endpoint through Web Unlocker."""
    response = requests.post(
        "https://api.brightdata.com/request",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"zone": ZONE, "url": api_url, "format": "raw"}
    )
    return json.loads(response.text)

# Example: site with internal API
data = fetch_api("https://example.com/api/products?page=1&limit=50")
products = data["results"]  # Already structured!

方法 C：Browser API + Playwright

适用于需要 JavaScript 渲染、交互（点击、滚动、填写表单）或反机器人措施严格的网站。

import asyncio
from playwright.async_api import async_playwright

AUTH = os.environ.get("BROWSER_AUTH", "brd-customer-CUSTOMER_ID-zone-ZONE_NAME:PASSWORD")

async def scrape_with_browser(url: str) -> str:
    """Scrape a page using Bright Data Browser API."""
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(
            f"wss://{AUTH}@brd.superproxy.io:9222"
        )
        page = await browser.new_page()
        page.set_default_navigation_timeout(120_000)  # 2 minutes — required

        # Block unnecessary resources to reduce bandwidth costs
        await page.route("**/*.{png,jpg,jpeg,gif,svg,css,woff,woff2}",
                         lambda route: route.abort())

        await page.goto(url, wait_until="domcontentloaded")

        # Wait for the content you need to appear
        await page.wait_for_selector(".product-card", timeout=30_000)

        # Extract data using page.evaluate for performance
        products = await page.evaluate("""
            () => Array.from(document.querySelectorAll('.product-card')).map(card => ({
                name: card.querySelector('.product-title')?.textContent?.trim(),
                price: card.querySelector('.product-price')?.textContent?.trim(),
                url: card.querySelector('a')?.href,
            }))
        """)

        await browser.close()
        return products

必须遵守的 Browser API 规则：

• 始终将导航超时设为 120 秒（set_default_navigation_timeout(120_000)）
• 每个会话只能调用一次 page.goto()——访问新 URL 时，创建新的浏览器连接
• 使用 wait_until="domcontentloaded"，不要使用 networkidle（SPA 永远不会达到 networkidle 状态）
• 等待特定选择器，而不是任意延时
• 屏蔽图片、CSS 和字体，以降低带宽成本
• 使用 page.evaluate() 批量提取数据——这比逐个调用选择器更快

方法 D：用于无限滚动的 Browser API

适用于向下滚动时会加载更多内容的网站。

async def scrape_infinite_scroll(url: str, max_items: int = 100) -> list:
    """Scrape a page with infinite scroll."""
    async with async_playwright() as p:
        browser = await p.chromium.connect_over_cdp(
            f"wss://{AUTH}@brd.superproxy.io:9222"
        )
        page = await browser.new_page()
        page.set_default_navigation_timeout(120_000)
        await page.route("**/*.{png,jpg,jpeg,gif,svg,woff,woff2}",
                         lambda route: route.abort())
        await page.goto(url, wait_until="domcontentloaded")

        all_items = []
        previous_count = 0

        while len(all_items) < max_items:
            # Scroll to bottom
            await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            await page.wait_for_timeout(2000)  # Wait for content to load

            # Extract all currently visible items
            items = await page.evaluate("""
                () => Array.from(document.querySelectorAll('.item-selector')).map(el => ({
                    // ... extract fields
                }))
            """)

            all_items = items
            if len(all_items) == previous_count:
                break  # No new content loaded — we've reached the end
            previous_count = len(all_items)

        await browser.close()
        return all_items[:max_items]

---

阶段 5：处理分页

大多数抓取任务都涉及多个页面。具体方法取决于分页类型。

阅读 references/pagination-patterns.md，了解详细的分页策略。

模式 1：基于 URL 的分页（最常见）

通过 ?page=2 或 ?offset=20 等 URL 参数访问页面。

import time

def scrape_all_pages(base_url: str, max_pages: int = 50) -> list[dict]:
    """Scrape all pages of a paginated listing."""
    all_items = []

    for page_num in range(1, max_pages + 1):
        url = f"{base_url}?page={page_num}"
        html = fetch_page(url)
        items = parse_products(html)

        if not items:
            break  # No more results

        all_items.extend(items)
        print(f"Page {page_num}: {len(items)} items (total: {len(all_items)})")

        time.sleep(1)  # Be respectful — don't hammer the site

    return all_items

模式 2：下一页链接分页

跟随 HTML 中找到的“下一页”链接。

def scrape_with_next_links(start_url: str) -> list[dict]:
    """Follow next-page links to scrape all pages."""
    all_items = []
    url = start_url

    while url:
        html = fetch_page(url)
        items = parse_products(html)
        all_items.extend(items)

        # Find next page link
        soup = BeautifulSoup(html, "html.parser")
        next_link = soup.select_one("a.next-page, a[rel='next'], .pagination .next a")
        url = next_link["href"] if next_link else None

        # Handle relative URLs
        if url and not url.startswith("http"):
            from urllib.parse import urljoin
            url = urljoin(start_url, url)

        time.sleep(1)

    return all_items

模式 3：并发批量抓取

当 URL 数量很多（50 个以上）时，始终使用带信号量的并发请求——绝不要在顺序循环中逐个获取。阅读 references/concurrency-guide.md，了解完整的并发操作指南，包括针对各网站的调优、多网站并行处理和重试策略。

import asyncio
import aiohttp

CONCURRENCY = 20  # Start here, tune per site — see concurrency guide

async def scrape_pages_concurrent(urls: list[str]) -> list[dict]:
    """Scrape multiple pages with controlled concurrency."""
    sem = asyncio.Semaphore(CONCURRENCY)

    async def fetch_one(session, url):
        async with sem:
            async with session.post(
                "https://api.brightdata.com/request",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json={"zone": ZONE, "url": url, "format": "raw"},
                timeout=aiohttp.ClientTimeout(total=60),
            ) as resp:
                return {"url": url, "html": await resp.text()}

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    all_items = []
    for r in results:
        if not isinstance(r, Exception):
            all_items.extend(parse_products(r["html"]))
    return all_items

# Generate all page URLs
urls = [f"https://example.com/products?page={i}" for i in range(1, 51)]
items = asyncio.run(scrape_pages_concurrent(urls))

模式 4：基于游标／令牌的分页（API）

有些 API 使用游标令牌，而不是页码。

def scrape_with_cursor(api_base: str) -> list[dict]:
    """Handle cursor-based API pagination."""
    all_items = []
    cursor = None

    while True:
        url = f"{api_base}?limit=100"
        if cursor:
            url += f"&cursor={cursor}"

        data = fetch_api(url)
        all_items.extend(data["results"])

        cursor = data.get("next_cursor")
        if not cursor:
            break

    return all_items

---

阶段 6：组装完整的抓取器

现在，将所有内容整合到一个简洁、可运行的脚本中。你构建的每个抓取器都应包含：

1. 配置——环境变量、目标 URL、输出设置
2. 获取器——用于检索页面的函数（Web Unlocker 或 Browser API）
3. 解析器——用于从 HTML／JSON 中提取结构化数据的函数
4. 分页器——处理多个页面的逻辑
5. 并发——通过信号量控制并行获取（参见 references/concurrency-guide.md）
6. 输出——将结果保存为指定格式
7. 错误处理——重试、日志记录和优雅失败

重要提示：如果用户要抓取的 URL 超过约 50 个，抓取器必须使用并发请求，而不能采用顺序循环。完整的并发抓取器模板和调优指南，请参见 references/concurrency-guide.md。

模板：完整的抓取器脚本

#!/usr/bin/env python3
"""
Scraper for [SITE NAME] - [DESCRIPTION]
Built with Bright Data [API NAME]

Usage:
    export BRIGHTDATA_API_KEY="your-api-key"
    export BRIGHTDATA_UNLOCKER_ZONE="your-zone-name"
    python scraper.py
"""

import json
import os
import sys
import time
import logging
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin

# --- Configuration ---
API_KEY = os.environ["BRIGHTDATA_API_KEY"]
ZONE = os.environ["BRIGHTDATA_UNLOCKER_ZONE"]
TARGET_URL = "https://example.com/products"
OUTPUT_FILE = "results.json"
MAX_PAGES = 50

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
log = logging.getLogger(__name__)

# --- Fetcher ---
def fetch_page(url: str, retries: int = 3) -> str:
    for attempt in range(retries):
        try:
            response = requests.post(
                "https://api.brightdata.com/request",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json={"zone": ZONE, "url": url, "format": "raw"},
                timeout=60,
            )
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            log.warning(f"Attempt {attempt + 1} failed for {url}: {e}")
            if attempt < retries - 1:
                time.sleep(2 ** attempt)
    raise RuntimeError(f"Failed to fetch {url} after {retries} attempts")

# --- Parser ---
def parse_items(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    items = []

    for card in soup.select("ITEM_SELECTOR"):
        try:
            item = {
                "name": card.select_one("NAME_SELECTOR").get_text(strip=True),
                "price": card.select_one("PRICE_SELECTOR").get_text(strip=True),
                "url": card.select_one("a").get("href", ""),
            }
            items.append(item)
        except (AttributeError, TypeError) as e:
            log.warning(f"Failed to parse item: {e}")
            continue

    return items

# --- Paginator ---
def scrape_all(base_url: str) -> list[dict]:
    all_items = []

    for page in range(1, MAX_PAGES + 1):
        url = f"{base_url}?page={page}"
        log.info(f"Scraping page {page}...")

        html = fetch_page(url)
        items = parse_items(html)

        if not items:
            log.info(f"No items on page {page} — stopping")
            break

        all_items.extend(items)
        log.info(f"Got {len(items)} items (total: {len(all_items)})")
        time.sleep(1)

    return all_items

# --- Output ---
def save_results(items: list[dict], path: str):
    with open(path, "w") as f:
        json.dump(items, f, indent=2, ensure_ascii=False)
    log.info(f"Saved {len(items)} items to {path}")

# --- Main ---
if __name__ == "__main__":
    items = scrape_all(TARGET_URL)
    save_results(items, OUTPUT_FILE)

为用户构建抓取器时，请自定义此模板：

• 将 ITEM_SELECTOR、NAME_SELECTOR、PRICE_SELECTOR 替换为阶段 3 中找到的实际选择器
• 选择合适的获取器（Web Unlocker 或 Browser API）
• 从阶段 5 中选择合适的分页模式
• 添加用户需要的特定字段
• 如果用户需要 CSV 或插入数据库，请调整输出格式

---

决策参考：API 选择速查指南

场景	API	成本	速度
网站有预构建的抓取器（Amazon、LinkedIn 等）	Web Scraper API	按记录计费	快
静态 HTML 页面，不需要 JS	Web Unlocker	按请求计费（仅成功请求）	快
网站提供 JSON API	Web Unlocker → API 端点	按请求计费（仅成功请求）	最快
JS 渲染的内容（React、Vue、Angular）	Browser API	按带宽计费	中等
无限滚动	Browser API	按带宽计费	慢
需要提交表单／登录	Browser API	按带宽计费	中等
CAPTCHA 较多的网站	Browser API	按带宽计费	中等
搜索引擎结果	SERP API	按请求计费	快

---

常见注意事项

1. 如果 Web Unlocker 已能满足需求，就不要默认使用 Browser API。 Browser API 的费用更高（按带宽计费），速度也更慢。请始终先尝试 Web Unlocker。

2. 不要使用结构性 CSS 选择器，例如 div:nth-child(3) > span。网站添加横幅或调整元素顺序后，这些选择器就会失效。请使用数据属性或语义化选择器。

3. 不要硬编码分页上限。 请始终检查页面是否返回了实际条目。空页面表示已到末尾。

4. 不要跳过侦察阶段。 花 2 分钟分析 HTML，可以省下数小时调试脆弱的选择器。

5. 不要忘记为每个条目处理错误。 一张格式错误的商品卡片不应导致整个抓取任务崩溃。请使用 try/except 包装单个条目的解析过程。

6. 不要在 Browser API 中使用 networkidle。 单页应用（SPA）永远不会真正达到网络空闲状态。请改用 domcontentloaded + wait_for_selector。

7. 抓取列表时，不要每翻一页就创建新的浏览器会话。 如果你在列表页点击“下一页”，可以继续使用同一会话。只有在不同的基础 URL 之间切换时才创建新会话。

8. URL 很多时，不要按顺序逐个抓取。 逐个获取 1,000 多个 URL，并在每次请求之间使用 time.sleep(1)，速度慢得无法接受。请使用带信号量的并发请求。参见 references/concurrency-guide.md。

---

示例

示例 1：电商商品抓取器（已有预构建抓取器）

用户说：“为 Amazon 商品页面构建一个抓取器，我有 200 个 ASIN。”

操作：

1. 查看 supported-domains.md → Amazon 有预构建抓取器
2. 200 个 URL → 使用异步触发/轮询/获取（超过 20 个 URL 的同步上限）
3. 使用 client.scrape.amazon.products_trigger() 批量处理 URL
4. 轮询直至就绪，然后下载结构化 JSON
5. 无需解析 HTML — 数据已预先结构化

结果：完整的 Python 脚本，支持异步批量抓取、进度日志和 JSON 输出。

示例 2：自定义网站抓取器（无预构建抓取器）

用户说：“我需要抓取 jobs.customsite.com 上的所有职位列表，包括分页内容。”

操作：

1. 查看 supported-domains.md → 未列出 → 查询 Dataset List API → 未找到
2. 通过 Web Unlocker 获取页面 HTML 并分析结构
3. 内容位于 HTML 中（SSR）→ 使用 Web Unlocker 方案
4. 确定选择器：.job-card、.job-title、.company-name、.salary
5. 通过 ?page=N URL 参数分页
6. 构建包含获取器、解析器和分页器的完整抓取器

结果：使用 Web Unlocker + BeautifulSoup 并通过 URL 分页的完整 Python 脚本。

示例 3：以 JavaScript 为主的网站（需要 Browser API）

用户说：“抓取一个滚动时才加载数据的 React SPA 中的商品价格。”

操作：

1. 通过 Web Unlocker 获取 HTML → 正文为空，仅有 div#root → 由客户端渲染
2. 检查页面源代码中是否有隐藏的 API → 未找到 API
3. 升级为使用 Playwright 的 Browser API
4. 使用资源拦截实现无限滚动模式
5. 内容加载后，通过 page.evaluate() 提取数据

结果：使用 Browser API 的异步 Playwright 脚本，支持无限滚动处理和带宽优化。

---

故障排除

Web Unlocker 返回空白页面或被拦截的页面

原因： 网站需要 JavaScript 渲染，或具有严格的机器人检测机制。 解决方案： 升级为使用 Browser API。也可以尝试添加 data_format: "markdown"，确认内容是否存在，只是格式不同。

选择器在本地有效，但在生产环境中失效

原因： 网站会向不同地区或用户代理提供不同的 HTML。 解决方案： 在 Web Unlocker 请求中添加 country 参数，以指定相同地区。请根据 API 实际返回的 HTML 验证选择器，而不是依赖浏览器开发者工具。

抓取器在不同页面中返回重复条目

原因： 分页逻辑循环回绕，或网站使用了不一致的分页方式。 解决方案： 使用集合跟踪已见条目 ID。出现重复项时停止。确认分页 URL 模式正确。

浏览器 API 会话超时

原因： 导航超时时间过短，或网站解除拦截的速度较慢。 解决方案： 始终设置 set_default_navigation_timeout(120_000)。使用 wait_until="domcontentloaded"，不要使用 networkidle。检查你的区域是否需要启用高级域名。

API 返回 401 Unauthorized

原因： 缺少或无效的 BRIGHTDATA_API_KEY 环境变量。 解决方案： 验证密钥是否已设置：echo $BRIGHTDATA_API_KEY。前往 https://brightdata.com/cp/setting/users 获取新密钥。

---

参考文件

• references/supported-domains.md — 包含预构建抓取器、数据集 ID 和 Python SDK 方法的完整列表。编写自定义抓取代码前，请先查看此文件。
• references/site-analysis-guide.md — 分析网站 HTML 结构并选择可靠选择器的分步指南。
• references/pagination-patterns.md — 详细介绍各种分页策略，并为每种常见模式提供代码示例。
• references/concurrency-guide.md — 介绍如何使用信号量控制并发抓取 URL、按网站调整设置、多网站并行处理、重试和进度跟踪。用户需要抓取 50 个或更多 URL 时，请阅读此文件。