Crie um agregador de lista de empregos com CaptchaAI

Painéis de vagas travam a paginação atrás de um CAPTCHA quando você varre várias páginas. Neste tutorial você monta um agregador em Python que percorre vários quadros de vagas, resolve o CAPTCHA com a API da CaptchaAI, normaliza cargo, empresa e salário e grava tudo em um SQLite pesquisável.

Boa parte dos quadros de vagas remotos devolve de 10 a 20 páginas por busca, e cada página nova pode reapresentar o desafio CAPTCHA — sem automação, alguém teria que resolvê-lo manualmente dezenas de vezes só para reunir os resultados de uma única palavra-chave. Este guia assume que você já tem uma chave de API da CaptchaAI e quer transformar essa varredura repetitiva em um pipeline que roda sozinho, sem intervenção manual a cada página.

Valide os seletores CSS de cada board contra uma página de teste antes de apontar o scraper para uma varredura completa — corrigir um seletor depois de gastar cota de threads sai mais caro do que checar antes.

Arquitetura do pipeline

[Job Board A] ──┐
[Job Board B] ──┼──> Scraper + CAPTCHA Solver ──> Normalizer ──> SQLite DB
[Job Board C] ──┘

Cada board é uma fonte independente; tudo converge para um SQLite deduplicado por URL.

A vantagem de centralizar em um único banco é direta: você roda a mesma busca em vários boards sem cruzar os resultados manualmente depois — a deduplicação já acontece no momento da gravação, pela restrição UNIQUE na coluna url.

Modelo de dados das vagas

# models.py
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import sqlite3
import json


@dataclass
class JobListing:
    title: str
    company: str
    location: str
    url: str
    source: str
    salary_min: Optional[float] = None
    salary_max: Optional[float] = None
    posted_date: Optional[str] = None
    description: str = ""
    tags: list = field(default_factory=list)
    scraped_at: str = field(default_factory=lambda: datetime.now().isoformat())


class JobDatabase:
    def __init__(self, db_path="jobs.db"):
        self.conn = sqlite3.connect(db_path)
        self._create_table()

    def _create_table(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS jobs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                company TEXT NOT NULL,
                location TEXT,
                url TEXT UNIQUE,
                source TEXT,
                salary_min REAL,
                salary_max REAL,
                posted_date TEXT,
                description TEXT,
                tags TEXT,
                scraped_at TEXT
            )
        """)
        self.conn.commit()

    def insert(self, job: JobListing):
        try:
            self.conn.execute(
                """INSERT OR IGNORE INTO jobs
                   (title, company, location, url, source,
                    salary_min, salary_max, posted_date,
                    description, tags, scraped_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (job.title, job.company, job.location, job.url,
                 job.source, job.salary_min, job.salary_max,
                 job.posted_date, job.description,
                 json.dumps(job.tags), job.scraped_at),
            )
            self.conn.commit()
        except sqlite3.IntegrityError:
            pass  # Duplicate URL

    def search(self, keyword, location=None):
        query = "SELECT * FROM jobs WHERE title LIKE ?"
        params = [f"%{keyword}%"]
        if location:
            query += " AND location LIKE ?"
            params.append(f"%{location}%")
        query += " ORDER BY scraped_at DESC"
        cursor = self.conn.execute(query, params)
        return cursor.fetchall()

O dataclass JobListing guarda salary_min e salary_max como campos separados porque a maioria dos anúncios publica uma faixa, não um valor único — assim você filtra por faixa salarial depois sem fazer parsing de string a cada consulta. scraped_at é preenchido automaticamente na criação do objeto, então você sempre sabe quando cada vaga foi coletada, mesmo que ela nunca volte a aparecer numa busca seguinte.

Scraper base com resolução de CAPTCHA integrada

# scraper_base.py
import requests
import re
import time
import os


class BaseScraper:
    API_KEY = os.environ["CAPTCHAAI_API_KEY"]

    def __init__(self, source_name):
        self.source = source_name
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36",
        })

    def fetch(self, url):
        resp = self.session.get(url, timeout=20)

        if self._has_captcha(resp.text):
            token = self._solve_captcha(url, resp.text)
            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            }, timeout=30)

        return resp.text

    def _has_captcha(self, html):
        return "data-sitekey" in html or "g-recaptcha" in html

    def _solve_captcha(self, url, html):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            raise ValueError("No sitekey found")

        sitekey = match.group(1)

        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": url,
            "json": 1,
        }, timeout=30)
        task_id = resp.json()["request"]
        time.sleep(15)

        for _ in range(24):
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.API_KEY, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data.get("status") == 1:
                return data["request"]
            if data["request"] != "CAPCHA_NOT_READY":
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("CAPTCHA solve timeout")

O método fetch() decide sozinho se precisa resolver um desafio: ele procura data-sitekey ou g-recaptcha no HTML da resposta e só chama _solve_captcha quando encontra um dos dois. O laço de consulta tenta por até 24 vezes, com 5 segundos entre tentativas — para reCAPTCHA v2 a CaptchaAI opera com teto de SLA abaixo de 60 s, então esse orçamento de espera cobre o pior caso com folga.

Adaptando o scraper para cada quadro de vagas

# scrapers.py
from bs4 import BeautifulSoup
from scraper_base import BaseScraper
from models import JobListing
import re


class GenericJobScraper(BaseScraper):
    """Scrape a job board search results page."""

    def __init__(self, source_name, base_url, selectors):
        super().__init__(source_name)
        self.base_url = base_url
        self.selectors = selectors

    def scrape_search(self, keyword, location="", max_pages=3):
        jobs = []

        for page in range(1, max_pages + 1):
            url = self.base_url.format(
                keyword=keyword.replace(" ", "+"),
                location=location.replace(" ", "+"),
                page=page,
            )
            html = self.fetch(url)
            page_jobs = self._parse_listings(html)

            if not page_jobs:
                break
            jobs.extend(page_jobs)

        return jobs

    def _parse_listings(self, html):
        soup = BeautifulSoup(html, "html.parser")
        cards = soup.select(self.selectors["card"])
        jobs = []

        for card in cards:
            title_el = card.select_one(self.selectors["title"])
            company_el = card.select_one(self.selectors["company"])
            location_el = card.select_one(self.selectors.get("location", ".location"))
            link_el = card.select_one(self.selectors.get("link", "a"))

            if not title_el or not company_el:
                continue

            salary = self._extract_salary(card.get_text())

            jobs.append(JobListing(
                title=title_el.get_text(strip=True),
                company=company_el.get_text(strip=True),
                location=location_el.get_text(strip=True) if location_el else "",
                url=link_el["href"] if link_el else "",
                source=self.source,
                salary_min=salary[0],
                salary_max=salary[1],
            ))

        return jobs

    def _extract_salary(self, text):
        match = re.search(
            r'\$?([\d,]+)\s*[-–to]+\s*\$?([\d,]+)', text
        )
        if match:
            return (
                float(match.group(1).replace(",", "")),
                float(match.group(2).replace(",", "")),
            )
        return (None, None)

Os seletores CSS ficam em um dicionário selectors — cada novo board vira uma configuração, não uma subclasse.

Orquestrando a coleta com o script principal

# main.py
import time
from models import JobDatabase
from scrapers import GenericJobScraper

BOARDS = [
    {
        "name": "Board A",
        "base_url": "https://board-a.example.com/search?q={keyword}&l={location}&p={page}",
        "selectors": {
            "card": ".job-card",
            "title": ".job-title",
            "company": ".company-name",
            "location": ".job-location",
            "link": "a.job-link",
        },
    },
]


def main():
    db = JobDatabase()
    keywords = ["python developer", "data engineer"]

    for board in BOARDS:
        scraper = GenericJobScraper(board["name"], board["base_url"], board["selectors"])

        for keyword in keywords:
            print(f"Scraping {board['name']} for '{keyword}'...")
            jobs = scraper.scrape_search(keyword, location="Remote")

            for job in jobs:
                db.insert(job)
                print(f"  {job.title} at {job.company}")

            time.sleep(5)

    # Search example
    results = db.search("python", "Remote")
    print(f"\nFound {len(results)} matching jobs")


if __name__ == "__main__":
    main()

Com vários boards em paralelo, o limite passa a ser a concorrência da conta: o plano BASIC (US$ 15/mês, 5 threads) cobre esta coleta sequencial; para rodar boards ao mesmo tempo, um plano com mais threads evita fila.

Agendamento e monitoramento em produção

Depois que o main.py roda sem erro, o próximo passo é tirar você do laço manual: agende a coleta via cron (Linux/macOS) ou o Agendador de Tarefas (Windows), e rode os workers numa região próxima dos boards que você consulta — se a maioria deles atende visitantes do Brasil, um worker na região sa-east-1 da AWS reduz o RTT tanto para o board quanto para a API da CaptchaAI.

Vale acompanhar quatro sinais a cada execução:

• Taxa de timeout do CAPTCHA — se TimeoutError passar a aparecer com frequência, o board pode ter trocado de sitekey ou o volume pede mais threads.
• Tempo médio de resolução — um salto repentino é o primeiro sinal de que vale revisar o plano de threads.
• Vagas novas por execução — se esse número cair a zero por vários ciclos seguidos, confira se os seletores CSS ainda batem com o HTML do board.
• Duplicatas ignoradas pela restrição UNIQUE — um número alto aqui é saudável; mostra que a deduplicação está funcionando, não que algo quebrou.

Solução de problemas

Nenhum desses problemas exige reescrever o scraper — normalmente é ajuste de seletor, de sessão ou de plano de threads.

Conformidade com a LGPD ao coletar vagas públicas

Cargo, empresa, faixa salarial e localização de um anúncio público de vaga costumam ser dados não pessoais — descrevem a posição, não uma pessoa física. Isso muda quando o board expõe o nome, e-mail ou telefone de um recrutador dentro do próprio anúncio: nesse caso, trate esse campo como dado pessoal sob a LGPD e colete só o que o seu caso de uso realmente precisa. Times que atendem Portugal aplicam o mesmo cuidado sob o RGPD.

Perguntas frequentes

Quantas threads da CaptchaAI eu preciso para rodar vários boards ao mesmo tempo?

Uma coleta sequencial roda bem no plano BASIC (5 threads); para boards em paralelo, vale um plano com mais threads.

Como evito duplicar uma vaga que aparece em mais de um board ou página?

A restrição UNIQUE na coluna url resolve isso: INSERT OR IGNORE descarta a segunda ocorrência, sem erro e sem duplicata.

Dá para plugar um novo quadro de vagas sem reescrever o scraper?

Sim. Basta uma entrada em BOARDS com a URL e os seletores CSS do board — GenericJobScraper já interpreta qualquer board nesse formato.

Como reduzo o risco de bloqueio ao coletar de vários boards?

Espace requisições com time.sleep(), alterne user agents e mantenha a sessão consistente.

Por que usar SQLite em vez de um banco mais robusto?

Para o volume de um agregador de vagas — de centenas a poucos milhares de registros por execução — o SQLite basta e elimina a necessidade de um servidor de banco separado. Se o volume crescer muito, trocar por Postgres exige mudar só a classe JobDatabase; o resto do pipeline continua igual.

Guias Relacionados

• Pipeline de geração de leads com CaptchaAI
• Boas práticas de rate limiting em fluxos de resolução de CAPTCHA

Agregue vagas de vários boards em um único banco pesquisável — comece com a CaptchaAI.