-
Notifications
You must be signed in to change notification settings - Fork 0
Recipes
HadesTop edited this page Aug 21, 2026
·
2 revisions
按场景组织的可直接抄的代码。每一节都是"我要干这个"→"这样写"。
参数的确切含义见 SDK 用法 与 API 参考。
所有例子都假设服务端已经在跑（ipclick run）。
from ipclick import Downloader
with Downloader() as d:
resp = d.get("https://example.com")
if resp.ok:
print(resp.text)
else:
print("失败:", resp.error)原始 HTML 是个空壳、内容靠 JS 填的时候：
import json
from ipclick import Downloader
with Downloader() as d:
resp = d.get(
"https://spa.example.com/products",
adapter="camoufox",
automation_config=json.dumps({
"wait_for_selector": ".product-card", # 等到真正要的元素出现
"block_resources": ["image", "font", "media"],
}),
)
print(resp.text)等元素而不是等固定时间——wait_for_selector 比 wait_for_timeout 又快又稳。
block_resources 通常能省一半时间。
异步客户端 + asyncio.gather：
import asyncio
from ipclick.aio import AsyncDownloader
async def fetch_all(urls: list[str]) -> dict[str, str]:
async with AsyncDownloader() as d:
results = await asyncio.gather(*(d.get(u) for u in urls))
return {r.url: r.text for r in results if r.ok}
pages = asyncio.run(fetch_all([f"https://example.com/p/{i}" for i in range(50)]))或者用批量接口，按完成顺序处理（快的先回）：
from ipclick import Downloader, DownloadTask
tasks = [DownloadTask(url=f"https://example.com/p/{i}") for i in range(50)]
with Downloader() as d:
for resp in d.batch(tasks):
if resp.ok:
save(resp.url, resp.text)不想同时打太多，用信号量夹住：
import asyncio
from ipclick.aio import AsyncDownloader
async def crawl(urls: list[str], limit: int = 10) -> list[str]:
sem = asyncio.Semaphore(limit)
async def one(d, url):
async with sem:
r = await d.get(url)
return r.text if r.ok else None
async with AsyncDownloader() as d:
return [t for t in await asyncio.gather(*(one(d, u) for u in urls)) if t]这是客户端侧的自我约束。服务端还有一层按目标域名的闸门， 见按 host 限流——那一层才是"别把对方打挂"的保证。
先登录拿 Cookie，再带着它抓：
from ipclick import Downloader
with Downloader() as d:
login = d.post(
"https://example.com/api/login",
json={"username": "u", "password": "p"},
)
login.raise_for_status()
token = login.json()["token"]
page = d.get(
"https://example.com/me",
headers={"Authorization": f"Bearer {token}"},
)
print(page.json())如果是 Cookie 会话：
cookie = login.headers.get("set-cookie", "")
page = d.get("https://example.com/me", cookies=cookie)请求之间不共享 cookie jar——每个请求相互独立。要保持会话就自己把 Cookie 传下去，如上。
from ipclick import Downloader
from ipclick.resume import download_to_file
def progress(received: int, total: int) -> None:
# total 是 -1 表示目标站点没给 Content-Length，服务端也不知道总长——
# 这时算百分比会得到负数，所以要按 total > 0 分开处理
if total <= 0:
print(f"\r{received / 1e6:.1f} MB（总长未知）", end="")
return
pct = received / total * 100
print(f"\r{received / 1e6:.1f} MB / {total / 1e6:.1f} MB ({pct:.1f}%)", end="")
with Downloader() as d:
r = download_to_file(d, "https://example.com/big.iso", "big.iso",
max_attempts=10, chunk_callback=progress)
print(f"\n完成 {r.total_bytes} 字节，试了 {r.attempts} 次，续传 {r.restarts} 次")import json
from ipclick import Downloader
with Downloader() as d:
with d.stream("https://example.com/data.ndjson") as s:
buf = b""
for chunk in s:
buf += chunk
while b"\n" in buf:
line, buf = buf.split(b"\n", 1)
handle(json.loads(line))记得用 with——提前 break 时它会取消服务端那次抓取。
with Downloader() as d:
# 用配置文件里的 [PROXY]
d.get(url, proxy=True)
# 或直接给
d.get(url, proxy="http://user:pass@1.2.3.4:8080")轮换多个代理：
from itertools import cycle
proxies = cycle([
"http://p1.example.com:8080",
"http://p2.example.com:8080",
])
with Downloader() as d:
for url in urls:
d.get(url, proxy=next(proxies))配好集群之后，用 create_client() 让配置决定形态：
from ipclick import create_client
with create_client() as d: # [GENERAL].mode = "cluster" 时给集群客户端
resp = d.get("https://example.com")
print(resp.trace.node_id) # 实际是哪台机器抓的要点名某台机器，见集群。
按代价从低到高试：
with Downloader() as d:
# 1. 默认就带 TLS 指纹伪装，先原样试
r = d.get(url)
# 2. 换一个指纹版本
if r.status_code == 403:
r = d.get(url, impersonate="chrome131")
# 3. 补齐像真人的请求头
if r.status_code == 403:
r = d.get(url, headers={
"Referer": "https://www.google.com/",
"Accept-Language": "zh-CN,zh;q=0.9",
})
# 4. 真起浏览器
if r.status_code == 403:
r = d.get(url, adapter="camoufox")
# 5. 换出口 IP
if r.status_code == 403:
r = d.get(url, proxy=True)排查思路见故障排查。
默认 4xx/5xx 会让 resp.ok 是 False，但有时这个状态码本身就是你要的答案：
resp = d.get(url, allowed_status_codes=[200, 429])
if resp.status_code == 429:
print("被限速了，交给上层排队重试")
resp = d.get(url) # 404 不必特别声明，它默认就在名单里
if resp.status_code == 404:
print("确认不存在")这也会让适配器不再为这个状态码重试——上面这个例子省掉的就是 429 那三轮退避等待。
只对本来会重试的码（默认 429/500/502/503/504）有意义。拿 404 举例是没用的：
它压根不在重试集合里，列上去省不掉任何等待。另外它也不改变成功判定——
resp.ok 仍然只认 2xx。
每个响应都带 request_uuid，拿它去查：
resp = d.get(url)
print(resp.request_uuid) # 01a01c47-a370-7812-...ipclick trace list --limit 20 --json
ipclick trace stats --json要能查历史就先打开落盘（默认只在内存里）：
[TRACE]
sqlite_enabled = true细节见链路记录。
# 拿状态码
ipclick fetch https://example.com --json | jq -r .status
# 只要正文
ipclick fetch https://example.com > page.html
# POST 一段 JSON
ipclick fetch https://api.example.com/items -X POST \
--json-body '{"name":"x"}' -H 'Authorization: Bearer ...' --json
# 从 stdin 读请求体
cat body.json | ipclick fetch https://api.example.com -X POST -d @- --json
# 下大文件
ipclick fetch https://example.com/big.zip -o big.zip
# 健康检查（退出码即结果，适合放 CI）
ipclick health && echo "服务正常"--json 时 stdout 上有且只有一个 JSON 文档，日志走 stderr，
所以管给 jq 永远安全。退出码含义见命令行。
把上面几条合起来——重试、限流、失败可查：
import asyncio, json, logging
from ipclick.aio import AsyncDownloader
log = logging.getLogger(__name__)
async def scrape(urls: list[str], concurrency: int = 8) -> dict[str, str]:
sem = asyncio.Semaphore(concurrency)
out: dict[str, str] = {}
async def one(d, url: str) -> None:
async with sem:
resp = await d.get(url, timeout=30, max_retries=3)
if resp.ok:
out[url] = resp.text
return
# 普通抓取拿不到，升级成浏览器再试一次
resp = await d.get(url, adapter="camoufox", timeout=90,
automation_config=json.dumps(
{"block_resources": ["image", "font", "media"]}))
if resp.ok:
out[url] = resp.text
else:
log.warning("放弃 %s: %s (uuid=%s)", url, resp.error, resp.request_uuid)
async with AsyncDownloader() as d:
await asyncio.gather(*(one(d, u) for u in urls))
return out
if __name__ == "__main__":
pages = asyncio.run(scrape([f"https://example.com/p/{i}" for i in range(100)]))
print(f"拿到 {len(pages)} 个页面")ipclick skill install # 写到 .claude/skills/ipclick/SKILL.md装完直接说"用 ipclick 抓一下 …"即可。技能包里已经写清楚了输出契约和常见坑， 不必再逐条解释。见命令行 → 给 AI 用。