Como resolver captchas ao realizar web scraping com o Scrapling e o CapSolver

pip install scrapling

pip install "scrapling[fetchers]"
scrapling install

pip install "scrapling[all]"
scrapling install

from scrapling import Fetcher

# Solicitação GET
response = Fetcher.get("https://example.com")

# Solicitação POST com dados
response = Fetcher.post("https://example.com/api", data={"chave": "valor"})

# Acessar resposta
print(response.status)       # Código de status HTTP
print(response.body)         # Bytes brutos
print(response.body.decode()) # Texto decodificado

import requests
import time

CAPSOLVER_API_KEY = "SUA_CHAVE_DE_API"

def solve_captcha(task_type, website_url, website_key, **kwargs):
    """
    Solucionador de captcha genérico usando a API do CapSolver.

    Args:
        task_type: O tipo de tarefa de captcha (ex: "ReCaptchaV2TaskProxyLess")
        website_url: A URL da página com o captcha
        website_key: A chave do site para o captcha
        **kwargs: Parâmetros adicionais específicos do tipo de captcha

    Returns:
        dict: A solução contendo o token e outros dados
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": task_type,
            "websiteURL": website_url,
            "websiteKey": website_key,
            **kwargs
        }
    }

    # Criar a tarefa
    response = requests.post(
        "https://api.capsolver.com/createTask",
        json=payload
    )
    result = response.json()

    if result.get("errorId") != 0:
        raise Exception(f"Falha na criação da tarefa: {result.get('errorDescription')}")

    task_id = result.get("taskId")
    print(f"Tarefa criada: {task_id}")

    # Verificar resultados
    max_attempts = 60  # Máximo de 2 minutos de verificação
    for attempt in range(max_attempts):
        time.sleep(2)

        response = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "taskId": task_id
            }
        )
        result = response.json()

        if result.get("status") == "ready":
            print(f"Captcha resolvido em {(attempt + 1) * 2} segundos")
            return result.get("solution")

        if result.get("errorId") != 0:
            raise Exception(f"Erro: {result.get('errorDescription')}")

        print(f"Esperando... (tentativa {attempt + 1})")

    raise Exception("Tempo esgotado: A resolução do captcha demorou demais")

pip install scrapling

pip install "scrapling[fetchers]"
scrapling install

pip install "scrapling[all]"
scrapling install

from scrapling import Fetcher

# Solicitação GET
response = Fetcher.get("https://example.com")

# Solicitação POST com dados
response = Fetcher.post("https://example.com/api", data={"chave": "valor"})

# Acessar resposta
print(response.status)       # Código de status HTTP
print(response.body)         # Bytes brutos
print(response.body.decode()) # Texto decodificado

import requests
import time

CAPSOLVER_API_KEY = "SUA_CHAVE_DE_API"

def solve_captcha(task_type, website_url, website_key, **kwargs):
    """
    Solucionador de captcha genérico usando a API do CapSolver.

    Args:
        task_type: O tipo de tarefa de captcha (ex: "ReCaptchaV2TaskProxyLess")
        website_url: A URL da página com o captcha
        website_key: A chave do site para o captcha
        **kwargs: Parâmetros adicionais específicos do tipo de captcha

    Returns:
        dict: A solução contendo o token e outros dados
    """
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": task_type,
            "websiteURL": website_url,
            "websiteKey": website_key,
            **kwargs
        }
    }

    # Criar a tarefa
    response = requests.post(
        "https://api.capsolver.com/createTask",
        json=payload
    )
    result = response.json()

    if result.get("errorId") != 0:
        raise Exception(f"Falha na criação da tarefa: {result.get('errorDescription')}")

    task_id = result.get("taskId")
    print(f"Tarefa criada: {task_id}")

    # Verificar resultados
    max_attempts = 60  # Máximo de 2 minutos de verificação
    for attempt in range(max_attempts):
        time.sleep(2)

        response = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={
                "clientKey": CAPSOLVER_API_KEY,
                "taskId": task_id
            }
        )
        result = response.json()

        if result.get("status") == "ready":
            print(f"Captcha resolvido em {(attempt + 1) * 2} segundos")
            return result.get("solution")

        if result.get("errorId") != 0:
            raise Exception(f"Erro: {result.get('errorDescription')}")

        print(f"Esperando... (tentativa {attempt + 1})")

    raise Exception("Tempo esgotado: A resolução do captcha demorou demais")

<div class="g-recaptcha" data-sitekey="6LcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxABCD"></div>

<script src="https://www.google.com/recaptcha/api.js?render=6LcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxABCD"></script>

from scrapling import Fetcher

def scrape_with_recaptcha_v2(target_url, site_key, form_url=None):
    """
    Raspagem de uma página protegida por ReCaptcha v2.

    Args:
        target_url: A URL da página com o captcha
        site_key: A chave de site do ReCaptcha
        form_url: A URL para submeter o formulário (padrão para target_url)

    Returns:
        A resposta da página protegida
    """
    # Resolver o captcha usando o CapSolver
    print("Resolvendo ReCaptcha v2...")
    solution = solve_captcha(
        task_type="ReCaptchaV2TaskProxyLess",
        website_url=target_url,
        website_key=site_key
    )

    captcha_token = solution["gRecaptchaResponse"]
    print(f"Obtido token: {captcha_token[:50]}...")

    # Submeter o formulário com o token de captcha usando o Scrapling
    # Nota: Use Fetcher.post() como método de classe (não método de instância)
    submit_url = form_url or target_url
    response = Fetcher.post(
        submit_url,
        data={
            "g-recaptcha-response": captcha_token,
            # Adicione quaisquer outros campos de formulário necessários pelo site
        }
    )

    return response

# Exemplo de uso
if __name__ == "__main__":
    url = "https://example.com/pagina-protegida"
    site_key = "6LcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxABCD"

    result = scrape_with_recaptcha_v2(url, site_key)
    print(f"Status: {result.status}")
    print(f"Comprimento do conteúdo: {len(result.body)}")  # Use .body para bytes brutos

solution = solve_captcha(
    task_type="ReCaptchaV2TaskProxyLess",
    website_url=target_url,
    website_key=site_key,
    isInvisible=True
)

solution = solve_captcha(
    task_type="ReCaptchaV2EnterpriseTaskProxyLess",
    website_url=target_url,
    website_key=site_key,
    enterprisePayload={
        "s": "valor_s_se_necessário"
    }
)

grecaptcha.execute('6LcxxxxxxxxxxxxxxxxABCD', {action: 'submit'})

from scrapling import Fetcher

def scrape_with_recaptcha_v3(target_url, site_key, page_action="submit", min_score=0.7):
    """
    Raspagem de uma página protegida pelo ReCaptcha v3.

    Args:
        target_url: A URL da página com o captcha
        site_key: A chave de site do ReCaptcha
        page_action: O parâmetro de ação (encontrado em grecaptcha.execute)
        min_score: Pontuação mínima solicitada (0,1-0,9)

    Returns:
        A resposta da página protegida
    """
    print(f"Resolvendo ReCaptcha v3 (ação: {page_action})...")

    solution = solve_captcha(
        task_type="ReCaptchaV3TaskProxyLess",
        website_url=target_url,
        website_key=site_key,
        pageAction=page_action
    )

    captcha_token = solution["gRecaptchaResponse"]
    print(f"Obtido token com pontuação: {solution.get('score', 'N/A')}")

    # Submeter a solicitação com o token usando o método de classe do Scrapling
    response = Fetcher.post(
        target_url,
        data={
            "g-recaptcha-response": captcha_token,
        },
        headers={
            "User-Agent": solution.get("userAgent", "Mozilla/5.0")
        }
    )

    return response

# Exemplo de uso
if __name__ == "__main__":
    url = "https://example.com/api/dados"
    site_key = "6LcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxABCD"

    result = scrape_with_recaptcha_v3(url, site_key, page_action="obterDados")
    print(f"Resposta: {result.body.decode()[:200]}")  # Use .body para conteúdo

solution = solve_captcha(
    task_type="ReCaptchaV3EnterpriseTaskProxyLess",
    website_url=target_url,
    website_key=site_key,
    pageAction=page_action,
    enterprisePayload={
        "s": "parâmetro_s_opcional"
    }
)

<div class="cf-turnstile" data-sitekey="0x4xxxxxxxxxxxxxxxxxxxxxxxxxx"></div>

turnstile.render('#container', {
    sitekey: '0x4xxxxxxxxxxxxxxxxxxxxxxxxxx',
    callback: function(token) { ... }
});

from scrapling import Fetcher

def scrape_with_turnstile(target_url, site_key, action=None, cdata=None):
    """
    Raspagem de uma página protegida pelo Cloudflare Turnstile.

    Args:
        target_url: A URL da página com o captcha
        site_key: A chave de site do Turnstile (começa com 0x4...)
        action: Parâmetro de ação opcional
        cdata: Parâmetro cdata opcional

    Returns:
        A resposta da página protegida
    """
    print("Resolvendo Cloudflare Turnstile...")

    # Construir metadados se fornecido
    metadata = {}
    if action:
        metadata["action"] = action
    if cdata:
        metadata["cdata"] = cdata

    task_params = {
        "task_type": "AntiTurnstileTaskProxyLess",
        "website_url": target_url,
        "website_key": site_key,
    }

    if metadata:
        task_params["metadata"] = metadata

    solution = solve_captcha(**task_params)

    turnstile_token = solution["token"]
    user_agent = solution.get("userAgent", "")

    print(f"Obtido token do Turnstile: {turnstile_token[:50]}...")

    # Submeter com o token usando o método de classe do Scrapling
    headers = {}
    if user_agent:
        headers["User-Agent"] = user_agent

    response = Fetcher.post(
        target_url,
        data={
            "cf-turnstile-response": turnstile_token,
        },
headers=headers
    )

    return response

# Exemplo de uso
if __name__ == "__main__":
    url = "https://example.com/protected"
    site_key = "0x4AAAAAAAxxxxxxxxxxxxxx"

    result = scrape_with_turnstile(url, site_key)
    print(f"Sucesso! Obteve {len(result.body)} bytes")  # Use .body para conteúdo

solution = solve_captcha(
    task_type="AntiTurnstileTaskProxyLess",
    website_url=target_url,
    website_key=site_key,
    metadata={
        "action": "login",
        "cdata": "session_id_ou_dados_customizados"
    }
)

from scrapling import StealthyFetcher
import asyncio

async def scrape_with_stealth_and_recaptcha(target_url, site_key, captcha_type="v2"):
    """
    Combina as funcionalidades anti-bot do StealthyFetcher com o CapSolver para ReCaptcha.

    Args:
        target_url: A URL para scraping
        site_key: A chave do captcha
        captcha_type: "v2" ou "v3"

    Returns:
        O conteúdo da página após resolver o captcha
    """

    # Primeiro, resolva o captcha usando o CapSolver
    print(f"Resolvendo ReCaptcha {captcha_type}...")

    if captcha_type == "v2":
        solution = solve_captcha(
            task_type="ReCaptchaV2TaskProxyLess",
            website_url=target_url,
            website_key=site_key
        )
        token = solution["gRecaptchaResponse"]

    elif captcha_type == "v3":
        solution = solve_captcha(
            task_type="ReCaptchaV3TaskProxyLess",
            website_url=target_url,
            website_key=site_key,
            pageAction="submit"
        )
        token = solution["gRecaptchaResponse"]

    else:
        raise ValueError(f"Tipo de captcha desconhecido: {captcha_type}")

    print(f"Obteve token: {token[:50]}...")

    # Use o StealthyFetcher para comportamento de navegador
    fetcher = StealthyFetcher()

    # Navegue até a página
    page = await fetcher.async_fetch(target_url)

    # Injeta a solução do ReCaptcha usando JavaScript
    await page.page.evaluate(f'''() => {{
        // Encontre o campo g-recaptcha-response e defina seu valor
        let field = document.querySelector('textarea[name="g-recaptcha-response"]');
        if (!field) {{
            field = document.createElement('textarea');
            field.name = "g-recaptcha-response";
            field.style.display = "none";
            document.body.appendChild(field);
        }}
        field.value = "{token}";
    }}''')

    # Encontre e clique no botão de submissão
    submit_button = page.css('button[type="submit"], input[type="submit"]')
    if submit_button:
        await submit_button[0].click()
        # Aguarde a navegação
        await page.page.wait_for_load_state('networkidle')

    # Obtenha o conteúdo final da página
    content = await page.page.content()

    return content

# Wrapper síncrono para uso mais fácil
def scrape_stealth(target_url, site_key, captcha_type="v2"):
    """Wrapper síncrono para o scraper assíncrono do stealth."""
    return asyncio.run(
        scrape_with_stealth_and_recaptcha(target_url, site_key, captcha_type)
    )

# Exemplo de uso
if __name__ == "__main__":
    url = "https://example.com/pagina-altamente-protegida"
    site_key = "6LcxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxABCD"

    content = scrape_stealth(url, site_key, captcha_type="v2")
    print(f"Obteve {len(content)} bytes de conteúdo")

from scrapling import StealthyFetcher
import asyncio

class StealthScraper:
    """Um scraper que mantém sessão entre múltiplas páginas."""

    def __init__(self, api_key):
        self.api_key = api_key
        self.fetcher = None

    async def __aenter__(self):
        self.fetcher = StealthyFetcher()
        return self

    async def __aexit__(self, *args):
        if self.fetcher:
            await self.fetcher.close()

    async def solve_and_access(self, url, site_key, captcha_type="v2"):
        """Resolve o ReCaptcha e acessa a página."""
        global CAPSOLVER_API_KEY
        CAPSOLVER_API_KEY = self.api_key

        # Resolve o ReCaptcha
        task_type = f"ReCaptcha{captcha_type.upper()}TaskProxyLess"
        solution = solve_captcha(
            task_type=task_type,
            website_url=url,
            website_key=site_key
        )
        token = solution["gRecaptchaResponse"]

        # Navegue e injete o token
        page = await self.fetcher.async_fetch(url)
        # ... continue com interação na página

        return page

# Uso
async def main():
    async with StealthScraper("sua_api_key") as scraper:
        page1 = await scraper.solve_and_access(
            "https://example.com/login",
            "site_key_aqui",
            "v2"
        )
        # Sessão é mantida para requisições subsequentes
        page2 = await scraper.solve_and_access(
            "https://example.com/dashboard",
            "outro_site_key",
            "v3"
        )

asyncio.run(main())

import time
import random

def scrape_polite(urls, min_delay=2, max_delay=5):
    """Scrape com atrasos aleatórios para parecer mais humano."""
    results = []
    for url in urls:
        result = scrape_page(url)
        results.append(result)

        # Atraso aleatório entre requisições
        delay = random.uniform(min_delay, max_delay)
        time.sleep(delay)

    return results

def robust_solve_captcha(task_type, website_url, website_key, max_retries=3, **kwargs):
    """Resolve captcha com tentativas automáticas."""
    for attempt in range(max_retries):
        try:
            return solve_captcha(task_type, website_url, website_key, **kwargs)
        except Exception as e:
            print(f"Tentativa {attempt + 1} falhou: {e}")
            if attempt < max_retries - 1:
                time.sleep(5)  # Espere antes de tentar novamente
            else:
                raise

from urllib.robotparser import RobotFileParser

def pode_scrape(url):
    """Verifique se o scraping é permitido pelo robots.txt."""
    rp = RobotFileParser()
    rp.set_url(f"{url}/robots.txt")
    rp.read()
    return rp.can_fetch("*", url)

# O CapSolver suporta tarefas com proxy
solution = solve_captcha(
    task_type="ReCaptchaV2Task",  # Note: sem "ProxyLess"
    website_url=target_url,
    website_key=site_key,
    proxy="http://user:pass@proxy.example.com:8080"
)

import time

class CaptchaCache:
    def __init__(self, ttl=120):  # TTL padrão de 2 minutos
        self.cache = {}
        self.ttl = ttl

    def get_or_solve(self, key, solve_func):
        """Obtenha o token em cache ou resolva um novo."""
        if key in self.cache:
            token, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return token

        token = solve_func()
        self.cache[key] = (token, time.time())
        return token

import asyncio

async def async_solve_captcha(*args, **kwargs):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None,
        lambda: solve_captcha(*args, **kwargs)
    )

# Resolva múltiplos ReCaptchas
solution_v2 = solve_captcha("ReCaptchaV2TaskProxyLess", url, key1)
solution_v3 = solve_captcha("ReCaptchaV3TaskProxyLess", url, key2, pageAction="submit")

# Submeta os tokens usando o método da classe Scrapling
response = Fetcher.post(url, data={
    "g-recaptcha-response": solution_v2["gRecaptchaResponse"],
    "g-recaptcha-response-v3": solution_v3["gRecaptchaResponse"],
})