scraperapi-python-sdk

ScraperAPI — Python SDK 最佳实践

要求： Python 3.7+，pip install scraperapi-sdk，以及 SCRAPERAPI_API_KEY 环境变量。

设置

import os
from scraperapi_sdk import ScraperAPIClient

client = ScraperAPIClient(os.environ["SCRAPERAPI_API_KEY"])

切勿将 API 密钥硬编码。每次都从环境变量中读取。

决策指南

编写代码前，先选择合适的调用模式。

情况	模式
单个 URL，需要立即获得响应	client.get() — 同步
20 个以上 URL，或经常超时的单个 URL	Async Jobs API — 使用 requests.post() 调用 async.scraperapi.com/jobs
受支持的平台（Amazon、Google、Walmart、eBay、Redfin）	Structured Data 端点 — api.scraperapi.com/structured/<vertical>
页面通过 JavaScript 加载内容（React、Vue、Angular）	使用 render=True 的 client.get()
网站会拦截标准代理	使用 premium=True 的 client.get()

简单的单 URL 任务使用同步 SDK。批量任务则改用原始 requests 调用异步端点——SDK 不内置异步或批量支持。

基本用法

# Simple scrape — returns HTML as a string
html = client.get(url="https://example.com/")

# With parameters
html = client.get(
    url="https://example.com/",
    params={"render": True, "country_code": "us"}
)

参数参考

渲染

# Render JavaScript before returning HTML
# Use when: page uses React/Vue/Angular, or initial scrape returns empty/partial content
# Cost: +10 credits (standard: 1 → rendered: 10)
html = client.get(url="https://spa-site.com/", params={"render": True})

# Wait for a specific element before capturing (requires render=True)
# Use when: content loads lazily and simple render still misses it
html = client.get(
    url="https://spa-site.com/",
    params={"render": True, "wait_for_selector": ".product-list"}
)

# Capture a screenshot (auto-enables rendering)
html = client.get(url="https://example.com/", params={"screenshot": True})

先不使用 render=True。只有当响应缺少预期内容时才启用；它会增加费用和延迟。

代理和地理定位

# Route through a specific country — no extra cost
html = client.get(url="https://example.com/", params={"country_code": "us"})

# Premium residential/mobile proxies — for sites that block datacenter IPs
# Cost: 10 credits per request (25 with render=True)
html = client.get(url="https://hard-site.com/", params={"premium": True})

# Ultra-premium — for the toughest anti-bot protections
# Cost: 30 credits per request (75 with render=True)
# Note: incompatible with custom headers (keep_headers is ignored)
html = client.get(url="https://hardest-site.com/", params={"ultra_premium": True})

premium 和 ultra_premium 不能同时使用——切勿同时设置。逐步升级： standard (1 cr) → render (10 cr) → premium (10 cr) → premium+render (25 cr) → ultra_premium (30 cr)。

会话（粘性代理）

# Reuse the same proxy IP across multiple requests
# Use for: multi-page flows, pagination, login-state scraping
# Sessions expire 15 minutes after last use; any integer is a valid session ID
html1 = client.get("https://example.com/page1", params={"session_number": 42})
html2 = client.get("https://example.com/page2", params={"session_number": 42})

标头和设备类型

import requests

# Forward your own headers to the target site
# Note: keep_headers is incompatible with ultra_premium=True
html = client.get(
    url="https://example.com/",
    params={"keep_headers": True},
    headers={"Accept-Language": "en-US", "Referer": "https://google.com"}
)

# Emulate mobile or desktop user-agent
html = client.get(url="https://example.com/", params={"device_type": "mobile"})

响应格式

# Default: raw HTML
html = client.get(url="https://example.com/")

# Structured JSON for supported sites (Amazon, Google, etc.)
# autoparse extracts fields like title, price, reviews automatically
data = client.get(url="https://amazon.com/dp/B09V3KXJPB", params={"autoparse": True})

# Markdown — useful for LLM pipelines
text = client.get(url="https://docs.example.com/", params={"output_format": "markdown"})

# Plain text
text = client.get(url="https://example.com/", params={"output_format": "text"})

# Binary content (PDFs, images)
content = client.get(url="https://example.com/report.pdf", params={"binary_target": True})

升级阶梯

始终从最省成本的选项开始。只有当网站拦截了前一档时，才升级。

def scrape_with_escalation(url):
    for params in [
        {},                       # 1 credit
        {"render": True},         # 10 credits
        {"premium": True},        # 10 credits
        {"premium": True, "render": True},  # 25 credits
        {"ultra_premium": True},  # 30 credits
    ]:
        result = client.get(url=url, params=params)
        if result and "<html" in result.lower():
            return result
    return None

记录成功的档位，这样下次访问同一域名时就可以从该档位开始。

Async Jobs（用于批量任务）

同步 SDK 会在每次调用时阻塞（每个 URL 最长 70 秒）。对于 20 个以上 URL，直接使用 requests 调用异步端点——它会在后台提交任务，并支持轮询或接收 Webhook。

import os, requests, time

API_KEY = os.environ["SCRAPERAPI_API_KEY"]

def submit_job(url, extra_params=None):
    payload = {"apiKey": API_KEY, "url": url, "apiParams": extra_params or {}}
    r = requests.post("https://async.scraperapi.com/jobs", json=payload, timeout=10)
    r.raise_for_status()
    return r.json()  # {"id": "...", "status": "running", "statusUrl": "..."}

def poll_job(job, max_wait=120, interval=5):
    deadline = time.time() + max_wait
    while time.time() < deadline:
        r = requests.get(job["statusUrl"], timeout=10)
        data = r.json()
        if data["status"] == "finished":
            return data["response"]["body"]
        if data["status"] == "failed":
            raise RuntimeError(f"Job failed: {job['id']}")
        time.sleep(interval)
    raise TimeoutError(f"Job {job['id']} did not finish in {max_wait}s")

# Batch pattern — submit all, then collect
urls = ["https://example.com/page1", "https://example.com/page2"]
jobs = [submit_job(u) for u in urls]
results = [poll_job(j) for j in jobs]

如果你拥有自己的服务端，请使用 Webhook（"callback": {"type": "webhook", "url": "https://yourapp.com/cb"}），这样就完全不必轮询。请参阅 异步文档。

对于超大批量任务（最多 50,000 个 URL），请使用 https://async.scraperapi.com/batchjobs，并传入 "urls": [...] 数组。

Structured Data 端点

如果目标平台受支持，请使用这些端点代替抓取原始 HTML。它们会返回整洁的 JSON；与使用 render=True 抓取原始内容相比，消耗的积分更少，而且无需编写解析逻辑。

import os, requests

API_KEY = os.environ["SCRAPERAPI_API_KEY"]

def structured_get(vertical, params):
    """Call a structured data endpoint synchronously."""
    url = f"https://api.scraperapi.com/structured/{vertical}"
    r = requests.get(url, params={"api_key": API_KEY, **params}, timeout=70)
    r.raise_for_status()
    return r.json()

# Google SERP
results = structured_get("google/search", {"query": "best Python web scrapers"})

# Amazon product details
product = structured_get("amazon/product", {"asin": "B09V3KXJPB"})

# Walmart product search
items = structured_get("walmart/search", {"query": "standing desk", "tld": "com"})

支持的垂直领域：google/search、google/news、google/jobs、google/shopping、 google/maps、amazon/product、amazon/search、amazon/offers、 walmart/product、walmart/search、walmart/category、 ebay/product、ebay/search、redfin/listing、redfin/search。

有关各垂直领域所需参数，请参阅 结构化数据文档。

错误处理

import os, requests
from scraperapi_sdk import ScraperAPIClient

client = ScraperAPIClient(os.environ["SCRAPERAPI_API_KEY"])

def scrape(url, params=None):
    try:
        result = client.get(url=url, params=params or {})
        return result
    except requests.exceptions.HTTPError as e:
        status = e.response.status_code
        if status == 401:
            raise ValueError("Invalid API key — check SCRAPERAPI_API_KEY") from e
        if status == 403:
            # Blocked or credits exhausted — try escalating proxy tier
            raise RuntimeError("Blocked or out of credits") from e
        if status == 429:
            # Concurrent request limit — back off and retry
            raise RuntimeError("Rate limit — reduce concurrency or switch to async") from e
        if status in (500, 503):
            # Transient failure — retry with exponential backoff
            raise RuntimeError(f"Transient error {status} — retry") from e
        raise

状态码参考：200 表示成功，401 表示密钥无效，403 表示被拦截或积分不足，404 表示未找到目标， 429 表示速率限制，500/503 表示暂时性错误（可安全重试）。ScraperAPI 不会对 500 响应收费。

积分费用参考

请求类型	积分
标准	1
render=True	10
premium=True	10
premium=True + render=True	25
ultra_premium=True	30
ultra_premium=True + render=True	75
Structured Data（大多数垂直领域）	1–5

在任意请求中添加 "max_cost": N，即可限制积分消耗——如果请求所需积分超过 N，则返回 403。

文档

• Python SDK 入门
• 请求参数
• 结构化数据端点
• 异步 API
• 控制面板与额度