Bright Data API 的官方 Python SDK。抓取任何网站、获取 SERP 结果、绕过反爬虫检测和验证码,并访问 100 多个现成数据集。
pip install brightdata-sdk从 Bright Data 控制面板获取 API 令牌:
export BRIGHTDATA_API_TOKEN="your_api_token_here"已通过 CLI 登录? SDK 无需配置即可使用,它会自动回退至 brightdata login 存储的凭据。
此 SDK 原生支持异步。同时也提供同步客户端(参阅同步客户端)。
import asyncio
from brightdata import BrightDataClient
async def main():
async with BrightDataClient() as client:
result = await client.scrape_url("https://example.com")
print(result.data)
asyncio.run(main())async with BrightDataClient() as client:
result = await client.scrape_url("https://example.com")
print(result.data)如需进行非阻塞网页爬虫,请使用 mode="async"。这会触发请求并返回 response_id,SDK 将自动轮询,直至结果就绪:
async with BrightDataClient() as client:
# Triggers request → gets response_id → polls until ready
result = await client.scrape_url(
url="https://example.com",
mode="async",
poll_interval=5, # Check every 5 seconds
poll_timeout=180 # Web Unlocker async can take ~2 minutes
)
print(result.data)
# Batch scraping multiple URLs concurrently
urls = ["https://example.com", "https://example.org", "https://example.net"]
results = await client.scrape_url(url=urls, mode="async", poll_timeout=180)工作方式:
- 向
/unblocker/req发送请求 → 立即返回response_id - 轮询
/unblocker/get_result?response_id=...,直至就绪或超时 - 返回抓取的数据
何时使用异步模式:
- 批量抓取大量 URL
- 继续其他工作的同时在后台处理
性能说明:网络解锁器异步模式通常大约需要 2 分钟才能完成。如需更快地获取单个 URL 的结果,请使用默认同步模式(不传递 mode 参数)。
跨 Google、Bing 和 Yandex 搜索。Google 和 Bing 返回解析后的自然搜索结果(标题/URL/描述/排名);Yandex 没有上游解析器,因此返回原始 HTML。
async with BrightDataClient() as client:
# Google — parsed results
result = await client.search.google(query="python scraping", num_results=10)
for item in result.data:
print(item["title"], item["url"])
# Bing — parsed results (same shape as Google)
result = await client.search.bing(query="python scraping", num_results=10)
for item in result.data:
print(item["title"], item["url"])
# Yandex — raw HTML only (parse it yourself with BeautifulSoup etc.)
result = await client.search.yandex(query="python scraping", num_results=10)
print(f"{len(result.raw_html)} chars of HTML")批量查询:传递字符串列表而不是单个字符串,每项查询都会并发运行,并返回一个 List[SearchResult]。
如需非阻塞 SERP 请求,请使用 mode="async":
async with BrightDataClient() as client:
# Non-blocking - polls for results
result = await client.search.google(
query="python programming",
mode="async",
poll_interval=2, # Check every 2 seconds
poll_timeout=30 # Give up after 30 seconds
)
for item in result.data:
print(item['title'], item['link'])何时使用异步模式:
- 包含大量查询的批量操作
- 继续其他工作的同时在后台处理
- 抓取可能比平时耗时更长时
注意:异步模式使用相同的区域,并返回与同步模式相同的数据结构,无需额外配置!
SDK 包含适用于热门网站的现成爬虫工具:Amazon、LinkedIn、Instagram、Facebook 等。
模式:client.scrape.<platform>.<method>(url)
示例:Amazon
async with BrightDataClient() as client:
# Product details
result = await client.scrape.amazon.products(url="https://amazon.com/dp/B0CRMZHDG8")
# Reviews
result = await client.scrape.amazon.reviews(url="https://amazon.com/dp/B0CRMZHDG8")
# Sellers
result = await client.scrape.amazon.sellers(url="https://amazon.com/dp/B0CRMZHDG8")可用爬虫工具:
client.scrape.amazon:商品、评论、卖家client.scrape.linkedin:个人资料、公司、职位、帖子client.scrape.instagram:个人资料、帖子、评论、Reelsclient.scrape.facebook:帖子、评论、Reels、页面、Marketplace、活动client.scrape.tiktok:个人资料、按关键词/个人资料/URL 获取帖子client.scrape.youtube:视频、频道、评论client.scrape.reddit:帖子、评论、按关键词获取帖子client.scrape.pinterest:帖子、个人资料、按关键词/个人资料获取帖子client.scrape.chatgpt:发送提示词、获取响应client.scrape.perplexity:返回 AI 排名结果的查询client.scrape.digikey:电子元器件商品
通过 Chrome DevTools Protocol(CDP)访问云托管的 Chrome 实例。SDK 会构建连接 URL,你可以使用 Playwright、Puppeteer 或 Selenium 操作浏览器。
from brightdata import BrightDataClient
from playwright.async_api import async_playwright
client = BrightDataClient(
browser_username="brd-customer-<id>-zone-<zone>",
browser_password="<password>",
)
url = client.browser.get_connect_url(country="us") # country is optional
async with async_playwright() as pw:
browser = await pw.chromium.connect_over_cdp(url)
page = await browser.new_page()
await page.goto("https://example.com")
html = await page.content()
await browser.close()适用场景:需要完整浏览器自动化的网站,JS 渲染、登录流程、交互式点击。对于普通 HTML 获取,建议使用 client.scrape_url()。
AI 排名的网页搜索。与 SERP(按搜索引擎顺序返回结果)不同,Discover 接收一个 query 和一个 intent 短语,并按相关性重新排名。还可选择将完整页面内容提取为 Markdown。
async with BrightDataClient() as client:
result = await client.discover(
query="artificial intelligence trends 2026",
intent="latest AI technology developments",
country="us",
num_results=10,
)
for item in result.data:
print(f"[{item['relevance_score']:.2f}] {item['title']} {item['link']}")对于长时间运行的发现任务,请分别触发和轮询:
job = await client.discover_trigger(query="...", intent="...")
result = await job.wait_and_fetch(timeout=60)何时使用 Discover 而不是 SERP:如果需要根据自然语言意图进行实体级相关性排名(例如“查找专注于可持续发展的 AI 公司”),请使用 Discover。如果需要原始搜索引擎结果,请使用 SERP。
运行在 Bright Data 爬虫工作室中构建的自定义采集器。一次调用即可触发任务、轮询直至就绪并返回记录:
async with BrightDataClient() as client:
data = await client.scraper_studio.run(
collector="c_abc123", # your collector ID
input={"url": "https://example.com"}, # collector-specific input
timeout=180,
)
print(f"Got {len(data)} records")如需手动控制生命周期,请使用 client.scraper_studio.trigger() → .status() → .fetch()。
访问 Bright Data 提供的 100 多个现成数据集:来自热门平台的预采集结构化数据。
async with BrightDataClient() as client:
# Filter a dataset — returns a snapshot_id
snapshot_id = await client.datasets.imdb_movies(
filter={"name": "title", "operator": "includes", "value": "black"},
records_limit=5
)
# Download when ready (polls until snapshot is complete)
data = await client.datasets.imdb_movies.download(snapshot_id)
print(f"Got {len(data)} records")
# Quick sample: .sample() auto-discovers fields, no filter needed
# Works on any dataset
snapshot_id = await client.datasets.imdb_movies.sample(records_limit=5)将结果导出到文件:
from brightdata.datasets import export
export(data, "results.json") # JSON
export(data, "results.csv") # CSV
export(data, "results.jsonl") # JSONL可用数据集类别:
- 电子商务:Amazon、Walmart、Shopee、Lazada、Zalando、Zara、H&M、Shein、IKEA、Sephora 等
- 商业情报:ZoomInfo、PitchBook、Owler、Slintel、VentureRadar、Manta
- 职位与人力资源:Glassdoor(公司、评论、职位)、Indeed(公司、职位)、Xing
- 评论:Google 地图、Yelp、G2、Trustpilot、TrustRadius
- 社交媒体:Pinterest(帖子、个人资料)、Facebook 页面
- 房地产:Zillow、Airbnb 和 8 个以上区域平台
- 奢侈品牌:Chanel、Dior、Prada、Balenciaga、Hermes、YSL 等
- 娱乐:IMDB、NBA、Goodreads
查看可用字段:
metadata = await client.datasets.imdb_movies.get_metadata()
for name, field in metadata.fields.items():
print(f"{name}: {field.type}")并发运行多个请求:
import asyncio
from brightdata import BrightDataClient
async def main():
async with BrightDataClient() as client:
urls = ["https://example.com/page1", "https://example.com/page2", "https://example.com/page3"]
tasks = [client.scrape_url(url) for url in urls]
results = await asyncio.gather(*tasks)
asyncio.run(main())对于长时间运行的抓取任务:
async with BrightDataClient() as client:
# Trigger
job = await client.scrape.amazon.products_trigger(url="https://amazon.com/dp/B123")
# Wait for completion
await job.wait(timeout=180)
# Fetch results
data = await job.fetch()对于较简单的使用场景,请使用 SyncBrightDataClient:
from brightdata import SyncBrightDataClient
with SyncBrightDataClient() as client:
result = client.scrape_url("https://example.com")
print(result.data)
# All methods work the same
result = client.scrape.amazon.products(url="https://amazon.com/dp/B123")
result = client.search.google(query="python")详情请参阅 docs/sync_client.md。
检查账户、列出区域并清理未使用的区域:
async with BrightDataClient() as client:
# Verify token + connectivity (never raises — returns False on any error)
if await client.test_connection():
print("Connected")
# Account status, usage stats, credit balance
info = await client.get_account_info()
print(info)
# All zones on this account, with type and usage
zones = await client.list_zones()
for z in zones:
print(z["name"], z["type"])
# Delete a zone — destructive, confirm before calling
await client.delete_zone("zone_to_remove")RuntimeError:无法在异步上下文中使用 SyncBrightDataClient
# Wrong - using sync client in async function
async def main():
with SyncBrightDataClient() as client: # Error!
...
# Correct - use async client
async def main():
async with BrightDataClient() as client:
result = await client.scrape_url("https://example.com")RuntimeError:BrightDataClient 未初始化
# Wrong - forgot context manager
client = BrightDataClient()
result = await client.scrape_url("...") # Error!
# Correct - use context manager
async with BrightDataClient() as client:
result = await client.scrape_url("...")MIT 许可证