Créez un agrégateur d'offres d'emploi avec CaptchaAI

Un agrégateur d'offres d'emploi collecte les annonces de plusieurs sites, les nettoie et les range dans une base interrogeable. Dans la pratique, le HTML n'est presque jamais le problème : le vrai point de friction, c'est le reCAPTCHA v2 qui surgit dès que le rythme des requêtes attire l'attention du site. Résolvez ce défi proprement et le reste du pipeline — parsing, normalisation, stockage — redevient de la plomberie Python classique.

Ce tutoriel construit l'agrégateur en quatre briques :

• un modèle de données commun à toutes les sources ;
• une classe de base qui détecte et résout le reCAPTCHA v2 via l'API CaptchaAI ;
• un scraper par source, qui ne décrit que ses sélecteurs CSS ;
• un orchestrateur qui enchaîne les sources et écrit en base.

Le solveur ciblé ici est le reCAPTCHA v2, le type le plus courant sur les portails d'emploi.

Agrégez uniquement des sources autorisées

Avant d'écrire une ligne de code, cadrez le périmètre. Un agrégateur ne doit collecter que des sources dont vous avez le droit d'exploiter les données : votre propre multidiffusion, des partenaires, des flux publics ou des sites dont les conditions autorisent l'agrégation. Vérifiez le fichier robots.txt et les CGU de chaque source.

Côté données, une annonce contient souvent des informations personnelles — nom du recruteur, coordonnées. Pour un lecteur en France, en Belgique ou au Québec, cela relève du RGPD : minimisez les champs collectés, ne stockez que ce qui sert la recherche et documentez la finalité. Le modèle ci-dessous ne conserve volontairement que le titre, l'entreprise, le lieu, l'URL et une fourchette de salaire — aucune donnée de contact.

Vue d'ensemble de l'architecture

Le flux est linéaire : chaque source passe par le même scraper, qui gère le CAPTCHA au besoin ; les résultats sont normalisés dans une structure commune, puis écrits dans une base SQLite unique.

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

Cette forme en entonnoir a un avantage : ajouter une source ne touche jamais au stockage ni à la normalisation. Vous décrivez la nouvelle source — URL et sélecteurs CSS — et vous réutilisez le reste tel quel.

Le modèle de données des offres

Le cœur de l'agrégateur est une structure JobListing unique, partagée par toutes les sources. Chaque site a un HTML différent, mais tous alimentent le même schéma : c'est ce qui rend les annonces comparables et interrogeables. La classe JobDatabase crée la table SQLite et gère l'insertion.

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

Le point clé est la contrainte UNIQUE sur url, couplée à INSERT OR IGNORE : la même offre republiée sur plusieurs pages n'est stockée qu'une fois. La déduplication est donc gratuite, sans requête supplémentaire.

Une classe de base qui gère les CAPTCHA

Toute la logique de résolution vit dans BaseScraper. La méthode fetch récupère la page ; si elle détecte un reCAPTCHA v2 — présence de data-sitekey ou de g-recaptcha —, elle envoie le défi à CaptchaAI, récupère le token, puis rejoue la requête avec ce 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")

Le déroulé côté API est standard : vous envoyez le sitekey et l'URL de la page à in.php avec method=userrecaptcha, vous récupérez un identifiant de tâche, puis vous interrogez res.php jusqu'à obtenir le token. Une fois résolu, le token part dans le champ g-recaptcha-response, exactement comme si un opérateur avait coché la case. Réutiliser self.session d'une requête à l'autre limite la réapparition du CAPTCHA : le site conserve les cookies de session et ne redéclenche pas le défi à chaque page.

Le scraper d'une page de résultats

GenericJobScraper hérite de la classe de base et n'ajoute que ce qui change d'un site à l'autre : le modèle d'URL et les sélecteurs CSS. La pagination boucle jusqu'à max_pages et s'arrête dès qu'une page ne renvoie plus de carte, ce qui évite de tourner à vide.

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

L'extraction du salaire est volontairement tolérante : l'expression régulière capture une fourchette min-max quel que soit le séparateur. Les formats varient beaucoup d'un site à l'autre ; adaptez le motif par source si nécessaire. Les cartes sans titre ni entreprise sont ignorées, ce qui écarte les blocs publicitaires glissés dans les résultats.

Le script d'orchestration

L'orchestrateur relie tout : il parcourt la liste BOARDS, lance un scraper par source et par mot-clé, puis insère les résultats. Un time.sleep(5) entre les sources espace les requêtes et garde le comportement prévisible.

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

Pour agréger plusieurs sources en parallèle, la contrainte n'est pas le prix par résolution mais le nombre de threads. CaptchaAI facture au thread simultané, avec des résolutions illimitées par thread : un thread traite un CAPTCHA à la fois. Le plan BASIC ($15/mois, 5 threads) suffit déjà pour cinq sources en parallèle ; passez au plan STANDARD ($30/mois, 15 threads) si vous ajoutez des sources ou parallélisez aussi les mots-clés. Aucune surfacturation selon le type de CAPTCHA. Pour la production, un worker hébergé chez OVHcloud ou Scaleway garde la collecte proche de vos sources européennes.

Dépannage

FAQ

Quels CAPTCHA CaptchaAI résout-il sur les sites d'emploi ?

Principalement le reCAPTCHA v2, le plus répandu sur les portails d'emploi, mais aussi le reCAPTCHA v3, Cloudflare Turnstile et les CAPTCHA image/OCR. Ce tutoriel utilise le reCAPTCHA v2 via method=userrecaptcha.

Combien de threads faut-il pour agréger plusieurs sources ?

Un thread résout un CAPTCHA à la fois. Pour cinq sources en parallèle, le plan BASIC ($15/mois, 5 threads) suffit ; ajoutez des threads si vous parallélisez aussi les mots-clés ou les pages.

Comment éviter les doublons entre plusieurs sites ?

La contrainte UNIQUE sur l'URL, combinée à INSERT OR IGNORE, écarte automatiquement une offre déjà stockée. Pour dédupliquer sur le contenu — même poste, URLs différentes —, ajoutez une empreinte titre + entreprise + lieu.

Comment rester conforme au RGPD en stockant des offres ?

Ne collectez que des sources autorisées et minimisez les champs : le modèle proposé évite volontairement les données de contact. Documentez la finalité et la durée de conservation, et n'agrégez pas de données personnelles inutiles à la recherche.

Que faire si les pages nécessitent JavaScript pour s'afficher ?

Si les annonces sont rendues côté client, requests ne verra qu'une coquille vide. Passez alors à Selenium ou Playwright pour piloter un vrai navigateur, tout en déléguant la résolution du CAPTCHA à CaptchaAI.

Guides connexes

• Construire un pipeline de génération de leads
• Bonnes pratiques de limitation de débit

Agrégez vos données d'offres d'emploi — démarrez avec CaptchaAI.