Tutoriales

Cree un agregador de ofertas de empleo con CaptchaAI

Un agregador de ofertas de empleo son cuatro piezas encajadas: un scraper por portal, una capa que resuelva el CAPTCHA sin detener el proceso, un normalizador y una base de datos donde buscar. Aquí las montas en Python con requests, BeautifulSoup, SQLite y la API de CaptchaAI para el reCAPTCHA v2 del camino.

Lo que hunde estos proyectos no es el parseo del HTML: es que el tercer portal pide un CAPTCHA en la página 2 y el script muere ahí, con media recolección hecha. Lo que sigue está ordenado para que ese caso sea un reintento y no el final.


Arquitectura del agregador

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

Cada portal entra por su propio scraper, pero todos comparten la clase base: la que detecta el CAPTCHA, pide el token a CaptchaAI y reenvía la solicitud. Añadir un portal cuesta un diccionario de selectores, no un scraper entero.

Antes de la primera línea de código: revisa los términos de servicio de cada portal y la normativa de protección de datos aplicable (RGPD y LOPDGDD en España, LFPDPPP en México). La responsabilidad legal es tuya.


Paso 1: modelo de datos y almacenamiento

La primera decisión es la clave de deduplicación: la misma oferta aparece en dos portales y otra vez la semana siguiente. Aquí se usa la URL con una restricción UNIQUE, lo más barato cuando cada portal publica su canónico.

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

JobListing es un dataclass a propósito: cada portal devuelve algo distinto y un contrato único obliga a normalizar en la extracción. salary_min y salary_max son opcionales: muchos anuncios no publican banda salarial.

INSERT OR IGNORE junto al UNIQUE sobre url te deja relanzar la misma ejecución sin duplicar nada: por eso puedes programarlo cada mañana sin pensarlo.


Paso 2: la clase base que resuelve el CAPTCHA

Esta es la pieza que separa un scraper de juguete de uno que termina. fetch() hace la solicitud; si la respuesta trae marcas de reCAPTCHA, extrae el sitekey, lo envía a CaptchaAI y reenvía el formulario con el token.

# 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")

Conviene entender tres cosas de este bloque:

  • El flujo es en dos tiempos. Envías la tarea a in.php y consultas res.php hasta que status vale 1. No es una llamada síncrona: el time.sleep(15) inicial evita consultas inútiles al principio.
  • CAPCHA_NOT_READY no es un error. Es el estado normal mientras la tarea se resuelve. Cualquier otro valor en request sí lo es, y el código lanza una excepción en lugar de seguir sondeando.
  • El token viaja en g-recaptcha-response. Va al formulario original de la página, no a un endpoint de CaptchaAI. Con Cloudflare Turnstile el campo es cf-turnstile-response, pero la mecánica es la misma.

La clave API sale de CAPTCHAAI_API_KEY; nunca la escribas en el archivo. La guía de inicio rápido cubre el alta, y el flujo de reCAPTCHA v2 explica cada parámetro de in.php.


Paso 3: el scraper genérico de portales

Con la base resuelta, el scraper concreto solo recorre páginas y convierte tarjetas en objetos JobListing.

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

scrape_search() se detiene en cuanto una página devuelve cero tarjetas. Sin esa comprobación pide siempre max_pages páginas, y cada página de más es otra oportunidad de encontrarte un CAPTCHA.

_extract_salary() es la función que más vas a tocar. La regex por defecto entiende rangos con $, y ahí llega el choque con el mercado hispanohablante: las bandas llegan en euros o en pesos, con otro separador de miles (45.000 € brutos/año frente a $45,000). Trátalo como un parser por portal: ajusta la regex en cada entrada de BOARDS.


Paso 4: el orquestador y la búsqueda

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

El time.sleep(5) entre portales no es decorativo: sin pausa recibes más CAPTCHA, más respuestas 429 y más bloqueos. Con la pausa, el mismo trabajo suele terminar antes.

Un escenario de volumen realista

Supón seis portales de empleo tech en España y México, doce búsquedas guardadas (desarrollador python, data engineer, soporte remoto…), tres páginas por búsqueda y una ejecución diaria: unas 216 páginas, de las que quizá el 10-15 % dispara un CAPTCHA. Entre 20 y 35 resoluciones diarias.

Como CaptchaAI factura por thread concurrente y no por resolución, la variable de coste no es esa cifra sino cuántas resoluciones tienes en vuelo a la vez. Con un agregador secuencial, BASIC ($15/mes, 5 threads) sobra; si paralelizas los portales, STANDARD ($30/mes, 15 threads) da margen. El coste fijo en USD resulta cómodo cuando facturas en moneda volátil.


Cuando algo se rompe

Síntoma Causa habitual Qué hacer
Ofertas duplicadas en la base La misma oferta aparece en varias páginas Ya está cubierto: deduplicación por URL con la restricción UNIQUE
salary_min y salary_max siempre en None Formato de banda salarial no contemplado Ajusta la expresión regular de _extract_salary por portal
CAPTCHA en cada página, no solo en la primera La sesión no persiste entre solicitudes Reutiliza self.session; no lances un requests.get suelto
Cero tarjetas después de resolver el CAPTCHA El formulario depende de JavaScript Pasa ese portal a Selenium o Playwright con CaptchaAI
No sitekey found El sitekey se inyecta por JS, no está en el HTML inicial Renderiza con un navegador headless antes de buscarlo

reCAPTCHA v2 no se resuelve de forma instantánea, y ningún proveedor serio promete lo contrario. Diseña el agregador para tolerar esa espera: que recolecte en segundo plano, nunca dentro de una petición.


Preguntas frecuentes

¿Necesito Selenium o me basta con requests?

Empieza con requests: es más rápido y funciona en la mayoría de portales que sirven los resultados en HTML. Pasa a Selenium o Playwright solo en los que renderizan las ofertas con JavaScript o inyectan el sitekey por script.

¿Qué tipos de CAPTCHA me voy a encontrar en portales de empleo?

Casi siempre reCAPTCHA v2 o v3, y cada vez más Cloudflare Turnstile. CaptchaAI resuelve esos tres, además de GeeTest v3, imagen/OCR y grid. hCaptcha y FunCaptcha (Arkose Labs) no están soportados, y GeeTest v4 figura como próximamente: un portal que use alguno de esos queda fuera.

¿Cuánto cuesta esto al mes?

Depende de la concurrencia, no del número de ofertas. Los planes empiezan en BASIC ($15/mes, 5 threads), con resoluciones ilimitadas por thread. Un agregador secuencial rara vez necesita más; uno paralelo suele quedarse en STANDARD ($30/mes, 15 threads).

¿Puedo guardar los datos en PostgreSQL en lugar de SQLite?

Sí, y es lo razonable en cuanto varias personas consulten el agregador. JobDatabase es la única clase que toca el almacenamiento: cambia la conexión y el CREATE TABLE. Mantén el UNIQUE sobre url, que sostiene la deduplicación.

¿Cada cuánto conviene ejecutarlo?

Una vez al día cubre la mayoría de los casos: las ofertas se publican en horario laboral y no caducan en horas. Ejecutarlo más veces multiplica solicitudes y CAPTCHA sin traer nada nuevo. Si necesitas más frescura, sube la frecuencia solo en las búsquedas que se mueven.


Guías relacionadas


¿Quieres que el agregador llegue al final de la ejecución? Empieza con CaptchaAI.

Los comentarios están deshabilitados para este artículo.