التكامل HTTPX + CaptchaAI

HTTPX هو عميل Python HTTP حديث مع دعم غير متزامن وHTTP/2. يوضح هذا الدليل كيفية استخدامه مع CaptchaAI لحل اختبار CAPTCHA المتزامن وغير المتزامن.

المتطلبات

pip install httpx

العميل المتزامن

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()

عميل غير متزامن

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

يدعم HTTPX HTTP/2، مما يقلل من حمل الاتصال:

pip install httpx[http2]

client = httpx.AsyncClient(http2=True, timeout=30)

يقوم HTTP/2 بإرسال طلبات متعددة عبر اتصال واحد، مما يؤدي إلى تحسين الأداء عند إرسال واستقصاء العديد من اختبارات CAPTCHA.

مثال على الكشط باستخدام معالجة CAPTCHA

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())

المقارنة: httpx مقابل الطلبات مقابل aiohttp

الأسئلة الشائعة

هل يجب علي استخدام httpx على الطلبات؟

بالنسبة للمشاريع الجديدة، نعم. يحتوي httpx على واجهة برمجة تطبيقات متوافقة مع الطلبات بالإضافة إلى دعم غير متزامن وHTTP/2. بالنسبة للتعليمات البرمجية الموجودة باستخدام الطلبات، يكون الترحيل واضحًا ومباشرًا.

هل httpx أسرع من aiohttp؟

يتمتع aiohttp بحمل أقل قليلاً لأحمال العمل غير المتزامنة. يعد httpx أسرع لاتصالات HTTP/2 وأكثر ملاءمة لرمز sync/async المختلط.

هل يمكنني استخدام httpx مع Scrapy؟

ليس بشكل مباشر - يستخدم Scrapy حلقة أحداث Twisted. استخدم httpx في البرامج النصية المستقلة أو مع أطر عمل غير متزامنة مثل FastAPI.

أدلة ذات صلة

• Axios + CaptchaAI: حل CAPTCHA بدون متصفح
• تجريف اختبار CAPTCHA باستخدام بايثون