Python curl_cffi库从入门到精通详解

pip install curl_cffi --upgrade

from curl_cffi import requests
r = requests.get("https://tools.scrapfly.io/api/fp/ja3", impersonate="chrome")
print(r.json())  # 应返回包含JA3指纹信息的JSON

from curl_cffi import requests

# 模拟Chrome 110的TLS指纹
response = requests.get(
    "https://httpbin.org/get",
    impersonate="chrome110",
    params={"key": "value"},
    headers={"User-Agent": "Custom Agent"}
)

print(response.status_code)
print(response.text)

# 发送JSON数据
payload = {"name": "John", "age": 30}
response = requests.post(
    "https://httpbin.org/post",
    json=payload,
    impersonate="chrome110"
)

# 发送文件
mp = curl_cffi.CurlMime()
mp.addpart(
    name="file",
    content_type="application/octet-stream",
    filename="test.txt",
    local_path="./test.txt"
)
response = requests.post("https://httpbin.org/post", multipart=mp)

proxies = {
    "http": "http://localhost:3128",
    "https": "socks5h://localhost:9050"
}

response = requests.get(
    "https://example.com",
    proxies=proxies,
    impersonate="chrome110"
)

with curl_cffi.Session() as session:
    # 自动保存cookies
    session.get("https://httpbin.org/cookies/set/sessionid/123")
    response = session.get("https://httpbin.org/cookies")
    print(response.json())

def on_message(ws, message):
    print(f"Received: {message}")

with curl_cffi.Session() as session:
    ws = session.ws_connect(
        "wss://echo.websocket.org",
        on_message=on_message
    )
    ws.send("Hello, WebSocket!")
    ws.run_forever()

import asyncio
from curl_cffi.requests import AsyncSession

async def safe_request():
    max_retries = 3
    for attempt in range(max_retries):
        try:
            async with AsyncSession() as session:
                response = await session.get("https://example.com")
                response.raise_for_status()
                return response
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 指数退避

asyncio.run(safe_request())

response = requests.get("https://example.com", verify=False)

curl = curl_cffi.Curl()
curl_cffi.setopt(curl, CURLOPT_SSLVERSION, 6)  # TLS 1.3
curl_cffi.setopt(curl, CURLOPT_SSL_CIPHER_LIST, "TLS_AES_256_GCM_SHA384")