Como posso extrair dados de um site para um arquivo CSV usando o Python Playwright (ou alternativas), evitando erros de acesso e melhorando a velocidade?

Antes mesmo de escrever seu código de scraping, sempre reserve um tempo para entender a página da web.

Neste caso, com base na visualização do código-fonte da página e na análise da aba de rede nas ferramentas de desenvolvimento, não há nada dinâmico aqui.

Meu primeiro instinto foi usar solicitações HTTP simples com um agente de usuário, mas elas são bloqueadas pelo servidor, então o Playwright é uma opção razoável.

Mas como os dados são estáticos, não há necessidade de hibernar , e você também pode usar o headless (com um agente de usuário ), desabilitar o JS e usar o predicado de navegação mais rápido "commit".

Aqui está uma reescrita inicial:

import asyncio
import pandas as pd  # 2.2.2
from io import StringIO
from playwright.async_api import async_playwright  # 1.48.0


URL = "<Your URL>"
UA = (
    "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/130.0.0.0 Mobile Safari/537.3"
)


async def scrape_all_pages(base_url, max_pages=10):
    markets = []

    async with async_playwright() as p:
        browser = await p.chromium.launch()
        context = await browser.new_context(user_agent=UA, java_script_enabled=False)
        page = await context.new_page()

        for page_num in range(1, max_pages + 1):
            url = f"{base_url}?page={page_num}"
            await page.goto(url, wait_until="commit")
            html = await page.content()
            markets.extend(pd.read_html(StringIO(html)))

    return pd.concat(markets, ignore_index=True)


async def main():
    df = await scrape_all_pages(URL, max_pages=10)
    df = df.dropna(how="all")
    print(df)


asyncio.run(main())

Hora original:

real 1m17.661s
user 0m3.348s
sys  0m1.190s

Tempo de reescrita:

real 0m6.912s
user 0m1.417s
sys  0m0.785s

Temos uma aceleração de 13x. A maior melhoria foi remover os sleeps desnecessários.

Se você estiver raspando mais páginas do que isso, adicionar uma fila de tarefas para aumentar o paralelismo pode ajudar. Mas para apenas 10 páginas, a sobrecarga e a complexidade do código de adicionar paralelismo não valem a pena, então vou pular isso por enquanto.

Além disso, você pode ter adicionado sleeps para evitar limitação de taxa do servidor. Esse é um bom motivo para sleep, mas então você não conseguirá obter um speedup. Usar um cluster de proxy residencial contornaria isso, mas é muito trabalhoso de configurar, também fora do escopo da questão.

Acho que isso resolverá seu problema.

• O navegador agora está no modo headless
• Imite os cabeçalhos para evitar a detecção de bots
• Aguarde o carregamento do documento
• Tempo de sono ajustado para 3
• Salve os dados em um arquivo CSV.

Não se esqueça de inserir o caminho para o arquivo CSV (linhas 9 e 68)

import asyncio
from playwright.async_api import async_playwright
import pandas as pd
from io import StringIO
import os

URL = "https://www.coingecko.com/en/coins/1/markets/spot"
  
os.makedirs('<PATH_TO_CSV_FILE>', exist_ok=True)


# Set up a function to add headers for mimicking a browser
def get_headers():
    return {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36",
        "Accept-Language": "en-US,en;q=0.9",
        "Accept-Encoding": "gzip, deflate, br"
    }

# Function to fetch a page with retries if needed
async def fetch_page(page, url):
    print(f"Fetching: {url}")
    await page.goto(url, wait_until="domcontentloaded")  # Wait until the DOM is fully loaded
    await asyncio.sleep(3)  # Adjust sleep time based on the website's response time
    return await page.content()

# Scrape all pages and save them to a CSV file
async def scrape_all_pages(url, max_pages=10):
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)  # Headless mode for better performance
        context = await browser.new_context(
            viewport={"width": 1280, "height": 900},
            user_agent=get_headers()["User-Agent"]  # Use a proper user-agent to avoid blocking
        )
        page = await context.new_page()

        markets = []
        for page_num in range(1, max_pages + 1):
            page_url = f"{url}?page={page_num}"
            html = await fetch_page(page, page_url)
            dfs = pd.read_html(StringIO(html))  # Parse tables
            markets.extend(dfs)

        await page.close()
        await context.close()
        await browser.close()

    return pd.concat(markets, ignore_index=True)

# Run the asynchronous function in the event loop
def run_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop and loop.is_running():
        return asyncio.create_task(coro)
    else:
        return asyncio.run(coro)

# Main function to orchestrate scraping
async def main():
    max_pages = 10
  
    df = await scrape_all_pages(URL, max_pages)
    df = df.dropna(how='all')  # Drop rows where all values are NaN
    df.to_csv('<PATH_TO_CSV_FILE>/scraped_data.csv', index=False)  # Save the data to a CSV file
    print("Data saved to 'scraped_data.csv'")

# Execute the main function
run_async(main())