HTTP 请求¶
Fetcher 基于高性能的 curl_cffi,提供快速、轻量的 HTTP 请求,以及多种隐身功能。
前提条件
- 已阅读 Fetcher 基础,了解 Response 对象,并知道如何选择 Fetcher。
- 已阅读查询元素,了解如何从 Selector/Response 对象中查找和提取元素。
- 已阅读主要类,了解 Response 从 Selector 继承的属性和方法。
基本用法¶
这个 Fetcher 的主要导入方式与其他 Fetcher 相同:
解析选项的配置方法见这里。通用参数¶
这些请求方法共有一些参数,先介绍它们。
- url:目标 URL。
- stealthy_headers:生成并添加真实浏览器请求头,包括 Google referer,默认开启。
- follow_redirects:控制重定向。默认为
"safe",跟随重定向,但拒绝指向内部或私有 IP 的目标,以防范 SSRF。传入True可不受限制地跟随所有重定向,传入False则完全关闭重定向。 - timeout:每次请求的最长等待时间,单位为秒。默认 30 秒。
- retries:失败请求的最大尝试次数,包含首次尝试。默认 3 次。
- retry_delay:重试之间的等待时间,单位为秒。默认 1 秒。
- impersonate:模拟指定浏览器的 TLS 指纹。可以传入浏览器字符串或列表,例如
"chrome110"、"firefox102"、"safari15_5"表示指定版本,"chrome"、"firefox"、"safari"、"edge"则自动使用最新可用版本。这样,请求在 TLS 层面看起来来自真实浏览器。传入列表时,每次请求随机选择一个。默认使用最新可用的 Chrome 版本。 - http3:使用 HTTP/3 协议发送请求。默认为 False。与
impersonate同时使用可能出现问题。 - cookies:请求使用的 Cookie,可以是
name→value字典或字典列表。 - proxy:此请求使用的代理,所有 HTTP 和 HTTPS 流量都会经过它。格式为
http://username:password@localhost:8030。 - proxy_auth:代理的 HTTP Basic 认证,格式为 (username, password) 元组。
- proxies:代理字典,格式为
{"http": proxy_url, "https": proxy_url}。 - proxy_rotator:用于自动轮换代理的
ProxyRotator实例,不能与proxy或proxies同时使用。 - headers:请求中添加的请求头,可以覆盖
stealthy_headers生成的任意请求头。 - max_redirects:最大重定向次数。默认 30 次,设为 -1 表示不限制。
- verify:是否验证 HTTPS 证书。默认为 True。
- cert:客户端证书和密钥的文件名元组,格式为 (cert, key)。
- selector_config:创建最终
Selector/Response对象时使用的自定义解析参数字典。
注意事项:
- 当前可模拟的浏览器有(
"edge"、"chrome"、"chrome_android"、"safari"、"safari_beta"、"safari_ios"、"safari_ios_beta"、"firefox"、"tor")。 - 可用浏览器及其版本会自动显示在参数补全中,并随
curl_cffi更新。 - 启用
impersonate或stealthy_headers后,Fetcher 会自动生成与所用浏览器版本匹配的真实请求头。
如需进一步自定义,还可以向各方法传入 curl_cffi 支持、而该方法尚未直接提供的参数。
文本编码¶
如果非 UTF-8 页面出现乱码,可以随请求传入 default_encoding="shift_jis",或页面实际使用的编码。只有 Content-Type 响应头未指定字符集时,才会使用这个后备编码。它不会覆盖服务器提供的字符集。
如果服务器提供了错误的字符集,可用 selector_config={"encoding": "shift_jis"} 强制指定正确编码。
HTTP 方法¶
不同方法还有各自的参数,例如 GET 的 params,以及 POST、PUT、DELETE 的 data/json。
通过例子更容易理解:
注意:不支持
OPTIONS和HEAD方法。
GET¶
from scrapling.fetchers import Fetcher
# Basic GET
page = Fetcher.get('https://example.com')
page = Fetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = Fetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = Fetcher.get('https://example.com/search', params={'q': 'query'})
# With headers
page = Fetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = Fetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = Fetcher.get('https://example.com', impersonate='chrome')
# HTTP/3 support
page = Fetcher.get('https://example.com', http3=True)
from scrapling.fetchers import AsyncFetcher
# Basic GET
page = await AsyncFetcher.get('https://example.com')
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', stealthy_headers=True)
page = await AsyncFetcher.get('https://scrapling.requestcatcher.com/get', proxy='http://username:password@localhost:8030')
# With parameters
page = await AsyncFetcher.get('https://example.com/search', params={'q': 'query'})
# With headers
page = await AsyncFetcher.get('https://example.com', headers={'User-Agent': 'Custom/1.0'})
# Basic HTTP authentication
page = await AsyncFetcher.get("https://example.com", auth=("my_user", "password123"))
# Browser impersonation
page = await AsyncFetcher.get('https://example.com', impersonate='chrome110')
# HTTP/3 support
page = await AsyncFetcher.get('https://example.com', http3=True)
page 始终是 Response 对象,也就是一种 Selector,因此可以直接解析:
page.css('.something.something')
page = Fetcher.get('https://api.github.com/events')
>>> page.json()
[{'id': '<redacted>',
'type': 'PushEvent',
'actor': {'id': '<redacted>',
'login': '<redacted>',
'display_login': '<redacted>',
'gravatar_id': '',
'url': 'https://api.github.com/users/<redacted>',
'avatar_url': 'https://avatars.githubusercontent.com/u/<redacted>'},
'repo': {'id': '<redacted>',
...
POST¶
from scrapling.fetchers import Fetcher
# Basic POST
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, params={'q': 'query'})
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = Fetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = Fetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = Fetcher.post('https://example.com/api', json={'key': 'value'})
from scrapling.fetchers import AsyncFetcher
# Basic POST
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, stealthy_headers=True)
page = await AsyncFetcher.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'}, proxy='http://username:password@localhost:8030', impersonate="chrome")
# Another example of form-encoded data
page = await AsyncFetcher.post('https://example.com/submit', data={'username': 'user', 'password': 'pass'}, http3=True)
# JSON data
page = await AsyncFetcher.post('https://example.com/api', json={'key': 'value'})
PUT¶
from scrapling.fetchers import Fetcher
# Basic PUT
page = Fetcher.put('https://example.com/update', data={'status': 'updated'})
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = Fetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = Fetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
from scrapling.fetchers import AsyncFetcher
# Basic PUT
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'})
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.put('https://example.com/update', data={'status': 'updated'}, proxy='http://username:password@localhost:8030')
# Another example of form-encoded data
page = await AsyncFetcher.put("https://scrapling.requestcatcher.com/put", data={'key': ['value1', 'value2']})
DELETE¶
from scrapling.fetchers import Fetcher
page = Fetcher.delete('https://example.com/resource/123')
page = Fetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = Fetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
from scrapling.fetchers import AsyncFetcher
page = await AsyncFetcher.delete('https://example.com/resource/123')
page = await AsyncFetcher.delete('https://example.com/resource/123', stealthy_headers=True, impersonate="chrome")
page = await AsyncFetcher.delete('https://example.com/resource/123', proxy='http://username:password@localhost:8030')
会话管理¶
要用相同配置发送多个请求,可以使用 FetcherSession。它同时支持同步和异步代码,会自动识别并切换会话类型,不需要导入另一个类。
FetcherSession 接受请求方法中的几乎所有参数。可以先为整个会话配置默认值,再轻松覆盖某次请求的设置,如下面的示例所示。
from scrapling.fetchers import FetcherSession
# Create a session with default configuration
with FetcherSession(
impersonate='chrome',
http3=True,
stealthy_headers=True,
timeout=30,
retries=3
) as session:
# Make multiple requests with the same settings and the same cookies
page1 = session.get('https://scrapling.requestcatcher.com/get')
page2 = session.post('https://scrapling.requestcatcher.com/post', data={'key': 'value'})
page3 = session.get('https://api.github.com/events')
# All requests share the same session and connection pool
也可以给 FetcherSession 配置 ProxyRotator,在请求之间自动轮换代理:
from scrapling.fetchers import FetcherSession, ProxyRotator
rotator = ProxyRotator([
'http://proxy1:8080',
'http://proxy2:8080',
'http://proxy3:8080',
])
with FetcherSession(proxy_rotator=rotator, impersonate='chrome') as session:
# Each request automatically uses the next proxy in rotation
page1 = session.get('https://example.com/page1')
page2 = session.get('https://example.com/page2')
# You can check which proxy was used via the response metadata
print(page1.meta['proxy'])
向请求方法直接传入 proxy=,可以覆盖这次请求使用的会话代理或轮换器:
with FetcherSession(proxy='http://default-proxy:8080') as session:
# Uses the session proxy
page1 = session.get('https://example.com/page1')
# Override the proxy for this specific request
page2 = session.get('https://example.com/page2', proxy='http://special-proxy:9090')
异步示例:
async with FetcherSession(impersonate='firefox', http3=True) as session:
# All standard HTTP methods available
response = await session.get('https://example.com')
response = await session.post('https://scrapling.requestcatcher.com/post', json={'data': 'value'})
response = await session.put('https://scrapling.requestcatcher.com/put', data={'update': 'info'})
response = await session.delete('https://scrapling.requestcatcher.com/delete')
import asyncio
from scrapling.fetchers import FetcherSession
# Async session usage
async with FetcherSession(impersonate="safari") as session:
urls = ['https://example.com/page1', 'https://example.com/page2']
tasks = [
session.get(url) for url in urls
]
pages = await asyncio.gather(*tasks)
Fetcher 会通过 FetcherSession 为每次请求创建一个临时会话。
会话的优点¶
- 速度更快:比每次请求都创建会话快 10 倍。
- 保留 Cookie:自动在请求之间管理 Cookie。
- 节省资源:多个请求的内存和 CPU 使用更高效。
- 集中配置:在同一个地方管理请求设置。
示例¶
下面是一些完整示例,帮助初学者开始网页抓取。
基本 HTTP 请求¶
from scrapling.fetchers import Fetcher
# Make a request
page = Fetcher.get('https://example.com')
# Check the status
if page.status == 200:
# Extract title
title = page.css('title::text').get()
print(f"Page title: {title}")
# Extract all links
links = page.css('a::attr(href)').getall()
print(f"Found {len(links)} links")
抓取产品¶
from scrapling.fetchers import Fetcher
def scrape_products():
page = Fetcher.get('https://example.com/products')
# Find all product elements
products = page.css('.product')
results = []
for product in products:
results.append({
'title': product.css('.title::text').get(),
'price': product.css('.price::text').re_first(r'\d+\.\d{2}'),
'description': product.css('.description::text').get(),
'in_stock': product.has_class('in-stock')
})
return results
下载文件¶
from scrapling.fetchers import Fetcher
page = Fetcher.get('https://raw.githubusercontent.com/D4Vinci/Scrapling/main/docs/assets/main_cover.png')
with open(file='main_cover.png', mode='wb') as f:
f.write(page.body)
处理分页¶
from scrapling.fetchers import Fetcher
def scrape_all_pages():
base_url = 'https://example.com/products?page={}'
page_num = 1
all_products = []
while True:
# Get current page
page = Fetcher.get(base_url.format(page_num))
# Find products
products = page.css('.product')
if not products:
break
# Process products
for product in products:
all_products.append({
'name': product.css('.name::text').get(),
'price': product.css('.price::text').get()
})
# Next page
page_num += 1
return all_products
提交表单¶
from scrapling.fetchers import Fetcher
# Submit login form
response = Fetcher.post(
'https://example.com/login',
data={
'username': 'user@example.com',
'password': 'password123'
}
)
# Check login success
if response.status == 200:
# Extract user info
user_name = response.css('.user-name::text').get()
print(f"Logged in as: {user_name}")
提取表格¶
from scrapling.fetchers import Fetcher
def extract_table():
page = Fetcher.get('https://example.com/data')
# Find table
table = page.css('table')[0]
# Extract headers
headers = [
th.text for th in table.css('thead th')
]
# Extract rows
rows = []
for row in table.css('tbody tr'):
cells = [td.text for td in row.css('td')]
rows.append(dict(zip(headers, cells)))
return rows
导航菜单¶
from scrapling.fetchers import Fetcher
def extract_menu():
page = Fetcher.get('https://example.com')
# Find navigation
nav = page.css('nav')[0]
menu = {}
for item in nav.css('li'):
links = item.css('a')
if links:
link = links[0]
menu[link.text] = {
'url': link['href'],
'has_submenu': bool(item.css('.submenu'))
}
return menu
适用场景¶
以下情况适合使用 Fetcher:
- 需要快速发送 HTTP 请求。
- 希望尽量减少额外资源开销。
- 不需要执行 JavaScript,通过请求就能抓取网站。
- 需要一些隐身功能,例如网站有防护,但没有 JavaScript 挑战。
以下情况适合使用 FetcherSession:
- 对同一网站或不同网站发送多个请求。
- 需要在请求之间保留 Cookie 或认证状态。
- 希望通过连接复用提高性能。
- 需要让多个请求使用一致的配置。
- 使用需要会话状态的 API。
以下情况应使用其他 Fetcher:
- 需要浏览器自动化。
- 需要更高级的反机器人应对或隐身功能。
- 需要执行 JavaScript 或与动态内容交互。