HTTPX, eşzamansız desteği ve HTTP/2.'yi içeren modern bir Python HTTP istemcisidir. Bu kılavuz, hem eşitleme hem de eşzamansız CAPTCHA çözümü için bunun CaptchaAI ile nasıl kullanılacağını gösterir.
Gereksinimler
| Gereksinim | Ayrıntılar |
|---|---|
| Python | 3.8+ |
| httpx | 0,24+ |
| CaptchaAI API anahtarı | Buradan bir tane alın |
pip install httpx
Senkronize İstemci
import httpx
import time
import os
class CaptchaAISync:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://ocr.captchaai.com"
self.client = httpx.Client(timeout=30)
def solve(self, params, timeout=300):
params["key"] = self.api_key
# Submit
resp = self.client.get(f"{self.base_url}/in.php", params=params)
text = resp.text
if not text.startswith("OK|"):
raise Exception(f"Submit failed: {text}")
task_id = text.split("|")[1]
# Poll
deadline = time.time() + timeout
poll_params = {"key": self.api_key, "action": "get", "id": task_id}
while time.time() < deadline:
time.sleep(5)
result = self.client.get(
f"{self.base_url}/res.php", params=poll_params
)
if result.text == "CAPCHA_NOT_READY":
continue
if result.text.startswith("OK|"):
return result.text.split("|", 1)[1]
raise Exception(f"Solve failed: {result.text}")
raise TimeoutError(f"Task {task_id} timed out")
def get_balance(self):
resp = self.client.get(f"{self.base_url}/res.php", params={
"key": self.api_key, "action": "getbalance"
})
return float(resp.text)
def close(self):
self.client.close()
# Usage
solver = CaptchaAISync(os.environ["CAPTCHAAI_API_KEY"])
token = solver.solve({
"method": "userrecaptcha",
"googlekey": "6Le-wvkS...",
"pageurl": "https://example.com",
})
print(f"Token: {token[:50]}...")
solver.close()
Eşzamansız İstemci
import httpx
import asyncio
import os
class CaptchaAIAsync:
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://ocr.captchaai.com"
self.client = httpx.AsyncClient(timeout=30)
async def solve(self, params, timeout=300):
params["key"] = self.api_key
# Submit
resp = await self.client.get(
f"{self.base_url}/in.php", params=params
)
text = resp.text
if not text.startswith("OK|"):
raise Exception(f"Submit failed: {text}")
task_id = text.split("|")[1]
# Poll
deadline = asyncio.get_event_loop().time() + timeout
poll_params = {"key": self.api_key, "action": "get", "id": task_id}
while asyncio.get_event_loop().time() < deadline:
await asyncio.sleep(5)
result = await self.client.get(
f"{self.base_url}/res.php", params=poll_params
)
if result.text == "CAPCHA_NOT_READY":
continue
if result.text.startswith("OK|"):
return result.text.split("|", 1)[1]
raise Exception(f"Solve failed: {result.text}")
raise TimeoutError(f"Task {task_id} timed out")
async def get_balance(self):
resp = await self.client.get(f"{self.base_url}/res.php", params={
"key": self.api_key, "action": "getbalance"
})
return float(resp.text)
async def close(self):
await self.client.aclose()
# Usage
async def main():
solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
# Solve multiple concurrently
tasks = [
solver.solve({
"method": "userrecaptcha",
"googlekey": "6Le-wvkS...",
"pageurl": f"https://example.com/page{i}",
})
for i in range(5)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, r in enumerate(results):
if isinstance(r, Exception):
print(f"Page {i}: FAILED - {r}")
else:
print(f"Page {i}: solved ({len(r)} chars)")
await solver.close()
asyncio.run(main())
HTTP/2 Desteği
HTTPX, HTTP/2'yi destekleyerek bağlantı yükünü azaltır:
pip install httpx[http2]
client = httpx.AsyncClient(http2=True, timeout=30)
HTTP/2, istekleri tek bir bağlantı üzerinden çoğullayarak birden fazla CAPTCHA gönderirken ve yoklarken performansı artırır.
CAPTCHA İşleme ile Kazıma Örneği
import httpx
import re
import os
async def scrape_with_captcha(url, solver):
async with httpx.AsyncClient() as client:
# Fetch page
resp = await client.get(url)
html = resp.text
# Check for reCAPTCHA
match = re.search(
r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
)
if not match:
return html
site_key = match.group(1)
token = await solver.solve({
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
})
# Submit form with token
resp = await client.post(url, data={
"g-recaptcha-response": token,
})
return resp.text
async def main():
solver = CaptchaAIAsync(os.environ["CAPTCHAAI_API_KEY"])
content = await scrape_with_captcha("https://example.com", solver)
print(f"Got {len(content)} chars")
await solver.close()
asyncio.run(main())
Karşılaştırma: httpx vs istekler vs aiohttp
| Özellik | httpx (senkronizasyon) | httpx (zaman uyumsuz) | istekler | aiohttp |
|---|---|---|---|---|
| Eşzamansız destek | ⏳ | ✅ | ⏳ | ✅ |
| HTTP/2 | ✅ | ✅ | ⏳ | ⏳ |
| Bağlantı havuzu oluşturma | ✅ | ✅ | ✅ | ✅ |
| API uyumluluğu | istek benzeri | istek benzeri | -" | Farklı |
| Şunun için en iyisi: | Yerleşik değiştirme | Modern eşzamansız kod | Hızlı komut dosyaları | Yüksek eşzamanlılık |
SSS
İstekler üzerinde httpx kullanmalı mıyım?
Yeni projeler için evet. httpx'in isteklerle uyumlu bir API'sinin yanı sıra eşzamansız ve HTTP/2 desteği vardır. Mevcut kod kullanan istekler için geçiş basittir.
httpx aiohttp'den daha mı hızlı?
aiohttp, saf eşzamansız iş yükleri için biraz daha düşük ek yüke sahiptir. httpx, HTTP/2 bağlantıları için daha hızlıdır ve karışık senkronizasyon/async kodu için daha uygundur.
Httpx'i Scrapy ile kullanabilir miyim?
Doğrudan değil - Scrapy, Twisted'ın olay döngüsünü kullanır. httpx'i bağımsız komut dosyalarında veya FastAPI gibi eşzamansız tabanlı çerçevelerle birlikte kullanın.