-
Notifications
You must be signed in to change notification settings - Fork 0
SDK Usage
Python 客户端的完整用法。只想快速跑通看快速开始， 想照着现成场景改看示例集，要查签名和字段看API 参考。
前提：服务端已经在跑（ipclick run）。客户端只是把任务通过 gRPC 交给它。
三种方式，按"要不要自己管生命周期"来选。
from ipclick import Downloader
with Downloader() as d:
print(d.get("https://example.com").status_code)
# 退出 with 时自动关闭 gRPC 连接不传参数就按配置文件连（默认 127.0.0.1:9528）。要显式指定：
with Downloader(host="10.0.0.5", port=9528, token="...") as d:
...
with Downloader(config_path="/etc/ipclick/ipclick.toml") as d:
...不想在每个函数里传客户端时用它。同一组连接参数只建一个，进程内复用：
from ipclick import get_downloader, close_all_downloaders
d = get_downloader() # 第一次建，后面拿同一个
d2 = get_downloader() # d2 is d
...
close_all_downloaders() # 进程退出前统一关最省事的写法，导入即用，第一次调用时才真的建连接：
from ipclick import downloader
resp = downloader.get("https://example.com")适合脚本和 REPL。长期运行的服务里建议用前两种——生命周期看得见。
create_client()会按[GENERAL].mode决定给你单机客户端还是集群客户端 （ClusterDownloader）。想让"单机还是集群"由配置说话、代码不改，就用它。
with Downloader() as d:
d.get("https://example.com")
d.post("https://api.example.com/items", json={"name": "x"})
d.put("https://api.example.com/items/1", data=b"raw bytes")
d.patch("https://api.example.com/items/1", json={"name": "y"})
d.delete("https://api.example.com/items/1")
d.head("https://example.com")
d.options("https://example.com")get() 的第二个位置参数是 query string：
d.get("https://httpbin.org/get", {"page": 2, "q": "关键词"})
# 等价于 params={"page": 2, "q": "关键词"}post() 的 data 与 json 是两回事：
d.post(url, json={"a": 1}) # Content-Type: application/json，自动序列化
d.post(url, data=b"a=1&b=2") # 原样发送 bytes，Content-Type 自己在 headers 里给
d.post(url, data="纯文本") # str 会按 UTF-8 编码resp = d.get(
"https://example.com",
headers={"User-Agent": "MyBot/1.0", "Accept-Language": "zh-CN"},
cookies={"session": "abc123"},
)cookies 也接受原始 Cookie 头字符串：
d.get(url, cookies="session=abc123; theme=dark")不传 User-Agent 时会自动填一个真实浏览器的。要保持默认伪装就不要覆盖它——
自己填一个 python-requests/2.x 反而更容易被拦。
便捷方法是它的薄包装。参数多的时候直接用它更清楚：
from ipclick import Downloader, HttpMethod
with Downloader() as d:
resp = d.request(
method=HttpMethod.POST,
url="https://api.example.com/search",
headers={"Authorization": "Bearer ..."},
params={"page": 1},
json={"q": "关键词"},
timeout=30,
max_retries=3,
retry_backoff=2.0,
allow_redirects=True,
verify=True,
adapter="curl_cffi",
)method 必须传 HttpMethod 枚举成员（如 HttpMethod.POST）——不像 adapter
那样接受字符串；传 "POST" 这样的字符串会在发送时抛出 ValidationError。
完整参数表见API 参考。
resp = d.get("https://httpbin.org/json")
resp.status_code # int，-1 表示没拿到 HTTP 响应
resp.ok # bool，等价于 is_success()
resp.text # str
resp.content # bytes
resp.json() # dict/list，解析失败抛异常
resp.headers # dict[str, str]
resp.elapsed_ms # int，服务端测的耗时
resp.url # 最终 URL（跟完重定向）
resp.error # str | None
resp.request_uuid # 这次请求的 id，拿它去查链路raise_for_status() 在失败时抛异常，适合"出错就该崩"的脚本：
resp = d.get(url)
resp.raise_for_status() # 4xx/5xx 或传输失败都会抛
data = resp.json()t = resp.trace
t.node_id # 哪台机器执行的
t.adapter # 实际用了哪个适配器
t.attempts # 试了几次（1 = 一次就成）
t.forwarded # 是不是被转发到别的节点
t.queued_ms # 在限流闸门里排了多久排查"为什么慢"先看 queued_ms 和 attempts：前者大是被自己的限流挡住了，
后者大是目标站点在抖。
核心规则：网络问题不抛异常，用法问题抛异常。
resp = d.get("https://does-not-exist.invalid")
# 不抛异常
print(resp.status_code) # -1
print(resp.error) # 具体错误
print(resp.ok) # False所以正常的判断长这样：
resp = d.get(url)
if not resp.ok:
log.warning("抓取失败 %s: %s", url, resp.error)
return None
return resp.text而这些情况会抛异常：
from ipclick import ValidationError, AuthenticationError, AdapterError, URLNotAllowedError
d.get("") # ValidationError：URL 空
d.get("file:///etc/passwd") # ValidationError：前缀不是 http:// / https://
d.get("http://169.254.169.254/") # URLNotAllowedError：服务端 SSRF 准入拒绝
d.get(url, adapter="camoufox") # AdapterError：这个适配器没装
# 令牌不对 # AuthenticationError注意上面第二、三行的区别。 客户端只看 URL 前缀，file:// 连不上服务端就被
ValidationError 挡了；协议白名单、内网地址、云元数据地址这些策略全在服务端，
要等服务端回话才会抛 URLNotAllowedError。所以只写 except URLNotAllowedError
接不住 file:// 那一条。
异常层次见 API 参考。想统一兜住就抓基类 IPClickError。
d.get(url) # 默认 curl_cffi，带 TLS 指纹伪装
d.get(url, adapter="niquests") # HTTP/2 + HTTP/3
d.get(url, adapter="camoufox") # 真浏览器渲染
d.get(url, adapter="browser") # 按平台自动挑一个浏览器引擎六个适配器怎么选见适配器。指定没装的适配器会抛 AdapterError
并带上安装命令，不会静默换一个。
curl_cffi 可以点名模仿某个浏览器版本：
d.get(url, impersonate="chrome131")不传时客户端会自动填 impersonate="chrome"，curl_cffi 再把它解析成当前版本的
Chrome 指纹。没有对应的配置项——改 ipclick.toml 改不动它，要点名版本只能按请求传
impersonate=。可选值随 curl_cffi 版本变化，见适配器。
三种写法：
d.get(url, proxy=True) # 用配置文件里的 [PROXY]
d.get(url, proxy="http://user:pass@1.2.3.4:8080") # 直接给 URLfrom ipclick import ProxyConfig
d.get(url, proxy=ProxyConfig(
scheme="http", host="1.2.3.4", port=8080,
auth_key="user", auth_password="pass",
))proxy=True 但配置里没填代理时会打一条警告并直连——不会静默当成"用了代理"。
账号密码建议走 .env（IPCLICK_PROXY_AUTH_KEY / IPCLICK_PROXY_AUTH_PASSWORD），
见配置体系。
拿渲染后的 DOM，而不是原始 HTML：
resp = d.get("https://spa.example.com", adapter="camoufox")
print(resp.text) # JS 跑完之后的 DOM等待策略、拦截资源、超时这些用 automation_config（JSON 字符串）：
import json
resp = d.get(
"https://spa.example.com",
adapter="camoufox",
automation_config=json.dumps({
"wait_until": "networkidle",
"block_resources": ["image", "font", "media"],
"wait_for_selector": "#content",
"wait_for_timeout": 3000,
}),
)block_resources 是最有效的提速手段——不下图片和字体通常能省掉一半时间。
完整参数与调优见浏览器渲染。
script = """async () => {
document.querySelector('#login').click();
await new Promise(r => setTimeout(r, 1000));
return document.title;
}"""
resp = d.get(url, adapter="camoufox", automation_script=script)脚本有两种写法，但只有一种能用 await：直接写函数体（return document.title;）
时，它会被包成一个同步箭头函数 () => { ... }，函数体里出现 await 就是
JS 语法错（会作为 ValidationError 报回来）。要用 await 就得像上面那样自己写成
async () => {...} 或 async function () {...} 开头——以 function / async /
( / => 开头的脚本会原样透传，不再包装。
⚠️ automation_script绕过 SSRF 准入——脚本里可以自己发请求到任何地址。 它默认关闭（[BROWSER].allow_scripts），只在你信任调用方时打开。 见安全。
响应体不进内存，边收边处理。适合大文件：
with Downloader() as d:
with d.stream("https://example.com/big.zip") as s:
print(s.status_code, s.content_length)
with open("big.zip", "wb") as f:
for chunk in s:
f.write(chunk)
print(s.total_bytes, s.elapsed_ms)要点：
-
一定要用
with（或手动close()）。提前break出循环时，with会取消 底层 RPC；不关的话服务端那次抓取还在跑。 -
s.total_bytes/s.elapsed_ms要消费完才有值——它们在流末尾的 trailer 里。 -
s.is_success()在消费完之后才计入 trailer 里的错误。
网络断了自动用 Range 续传，不重下已经拿到的部分：
from ipclick import Downloader
from ipclick.resume import download_to_file
with Downloader() as d:
result = download_to_file(d, "https://example.com/big.iso", "big.iso")
print(result.total_bytes, result.attempts, result.restarts)attempts > 1 说明中途断过；restarts > 0 说明没能续上、从头重下过——
服务端没给 206，或者压根不支持 Range，之前下的字节全白费了。
两个都为真时是"断过且重下过"，只有 attempts > 1 and restarts == 0 才是
"断过并真的续上了"。带进度回调：
def on_chunk(received: int, total: int) -> None:
pct = received / total * 100 if total else 0
print(f"\r{pct:.1f}%", end="")
download_to_file(d, url, "big.iso", chunk_callback=on_chunk, max_attempts=10)只要字节流不要落盘用 iter_resumable：
from ipclick.resume import iter_resumable
for chunk in iter_resumable(d, url):
process(chunk)服务端不支持 Range 时会从头重下（并在日志里说明），不会静默给你一个残缺文件。
一次 RPC 发多个任务，按完成顺序返回——快的先回，不用等最慢那个：
import uuid_utils as uuid
from ipclick import Downloader, DownloadTask
tasks = [DownloadTask(uuid=str(uuid.uuid7()), url=f"https://example.com/p/{i}") for i in range(100)]
with Downloader() as d:
for resp in d.batch(tasks):
print(resp.request_uuid, resp.url, resp.status_code)每个任务要自己填 uuid。 DownloadTask.uuid 默认是空字符串——真正发出去的
uuid 只在 to_protobuf()（也就是发送那一刻）才生成，不会写回你手上的这个
DownloadTask 对象。按完成顺序收响应、想用 resp.request_uuid 认出它对应哪个
任务时，就必须像上面这样在构造 DownloadTask 时自己传 uuid=；不传的话事后
没有任何办法拿到服务端真正用的那个 uuid。
提前 break 出循环是安全的：那条双向流会被取消，服务端不再继续替你把余下的任务打出去
（2.0.0 之后修的；此前留下一条半开的流，默认 timeout=None 即没有 deadline，
服务端会一直守着未取走的任务直到连接断开）。
DownloadTask 接受和 request() 一样的参数，可以每个任务不同：
tasks = [
DownloadTask(uuid=str(uuid.uuid7()), url="https://a.example.com", adapter="curl_cffi", timeout=10),
DownloadTask(uuid=str(uuid.uuid7()), url="https://b.example.com", adapter="camoufox", timeout=60),
]注意：批量整批发给同一个节点，不跨节点拆分。要打散到多个节点就自己分批， 或者用集群的客户端分发模式。
AsyncDownloader 不在顶层导出，从 ipclick.aio 导入：
import asyncio
from ipclick.aio import AsyncDownloader
async def main():
async with AsyncDownloader() as d:
resp = await d.get("https://example.com")
print(resp.status_code)
asyncio.run(main())并发抓一批：
async def fetch_all(urls: list[str]) -> list[str]:
async with AsyncDownloader() as d:
results = await asyncio.gather(*(d.get(u) for u in urls))
return [r.text for r in results if r.ok]流式与批量都有异步版：
async with AsyncDownloader() as d:
async with await d.stream(url) as s:
async for chunk in s:
...
async for resp in d.batch(tasks):
...同步和异步两条流式路径的语义是一致的——都会在提前退出时取消底层 RPC。
channel 绑事件循环：
AsyncDownloader的连接建立在第一次请求所在的事件循环上。 不要跨asyncio.run()复用同一个实例。
gRPC 连接建起来就该复用——每个请求新建一个客户端会把连接握手的开销加到每次请求上。
# 好：一个客户端打很多请求
with Downloader() as d:
for url in urls:
d.get(url)
# 差：每次都新建
for url in urls:
with Downloader() as d:
d.get(url)长期运行的服务里用 get_downloader() 拿进程级单例，退出时 close_all_downloaders()。
客户端关掉之后再用会抛 ClientClosedError——不会静默重连，因为那会掩盖
"你在用一个本该已经释放的对象"这个问题。