Создайте агрегатор списков вакансий с помощью CaptchaAI

Каждый job-борд отдаёт вакансии в своём формате, а часть страниц закрыта проверкой reCAPTCHA v2 — из-за этого один парсер быстро превращается в набор костылей. Здесь мы собираем компактный агрегатор на Python: общий базовый класс парсера решает CAPTCHA через CaptchaAI, приводит вакансии к единой схеме и складывает их в SQLite, по которой удобно искать. К концу статьи у вас будет рабочий каркас из четырёх файлов, к которому легко подключать новые доски.

Сценарий типичный для фрилансера или небольшого агентства: собрать удалённые вакансии python developer и data engineer сразу с нескольких площадок, отфильтровать дубли и держать всё в одной локальной базе. Тарификация CaptchaAI по потокам (тариф BASIC — $15/мес, 5 потоков) делает стоимость предсказуемой даже при биллинге в нестабильной локальной валюте: платите за одновременные потоки, а не за каждую решённую капчу.

Как устроен агрегатор

Схема простая: несколько парсеров тянут страницы результатов, единый слой решения CAPTCHA снимает проверки там, где они встречаются, нормализатор приводит поля к общей модели, а результат уходит в SQLite. Пайплайн проходит четыре стадии:

• Сбор — парсеры листают страницы поиска и забирают HTML в рамках одной сессии.
• Решение CAPTCHA — если страница закрыта reCAPTCHA v2, токен приходит от CaptchaAI, и запрос повторяется.
• Нормализация — карточки приводятся к единой модели JobListing.
• Хранение — вакансии складываются в SQLite с дедупликацией по URL.

Совет: держите слой решения CAPTCHA отдельно от парсеров конкретных досок. Тогда добавление новой площадки не затрагивает логику работы с проверками.

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

Что понадобится перед стартом

Перед запуском подготовьте окружение:

• Python 3.10 или новее и две библиотеки — requests и beautifulsoup4.
• Аккаунт CaptchaAI и API-ключ. Положите ключ в переменную окружения CAPTCHAAI_API_KEY, чтобы он не попал в систему контроля версий — базовый класс читает его именно оттуда.
• Ничего для базы. Отдельная СУБД не нужна: SQLite создаст файл jobs.db при первом запуске.

Весь проект — это четыре небольших файла (models.py, scraper_base.py, scrapers.py, main.py), которые мы соберём по порядку.

Шаг 1: модель данных и хранилище

Начинаем с общей схемы вакансии и обёртки над SQLite. JobListing описывает одну вакансию, а JobDatabase создаёт таблицу, вставляет записи и умеет искать по ключевому слову. Ограничение UNIQUE на поле url и вставка через INSERT OR IGNORE берут на себя дедупликацию — одну и ту же вакансию с разных страниц база просто не задвоит.

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

Шаг 2: базовый класс парсера с решением reCAPTCHA v2

Здесь живёт вся логика работы с CAPTCHA, чтобы конкретные парсеры о ней не думали. Метод fetch сначала запрашивает страницу в рамках одной сессии; если в HTML найден data-sitekey, он вытаскивает sitekey, отправляет задачу на in.php с методом userrecaptcha, опрашивает res.php до готовности и повторно отправляет форму с токеном в поле g-recaptcha-response. Сессия переиспользуется между запросами — это заметно снижает частоту повторных проверок. API-ключ читается из переменной окружения CAPTCHAAI_API_KEY, а не хранится в коде.

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

Шаг 3: парсер страницы результатов

GenericJobScraper наследует базовый класс и знает только про CSS-селекторы конкретной доски. Он листает страницы поиска до max_pages, разбирает карточки через BeautifulSoup и вытягивает зарплату регулярным выражением. Такой подход позволяет описать новую площадку одним словарём селекторов, не трогая код решения CAPTCHA.

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

Шаг 4: точка входа и запуск сборщика

Осталось связать всё вместе. Список BOARDS описывает площадки и их селекторы, main перебирает доски и ключевые слова, сохраняет вакансии в базу и делает паузу между досками. В конце — пример поиска по сохранённым данным. Добавление новой доски сводится к ещё одной записи в BOARDS.

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

Как запускать сбор регулярно

Разовый прогон удобен для проверки, но ценность агрегатора — в свежих данных. Запускайте main.py по расписанию и держите ритм щадящим:

• Расписание. На Linux — cron (например, раз в сутки ночью), на Windows — «Планировщик заданий».
• Темп запросов. Пауза между досками, единая сессия и time.sleep() снижают нагрузку на площадку и вероятность повторных проверок.
• Идемпотентность. Ограничение UNIQUE отсекает дубли, поэтому повторные запуски безопасно дополняют базу, а не задваивают её.

Если вакансий много, разбейте ключевые слова по нескольким запускам, чтобы не держать процесс часами.

Что делать, если что-то ломается

Частые вопросы

Какого тарифа CaptchaAI хватит для агрегатора?

Оплата идёт за одновременные потоки, а число решений в рамках тарифа не ограничено, поэтому ориентируйтесь на параллельность:

• BASIC ($15/мес, 5 потоков) — для ночного сбора с нескольких досок по очереди: поток освобождается сразу после решения.
• ADVANCE ($90/мес, 50 потоков) — если досок много и вы запускаете их параллельно.

Начните с меньшего тарифа и поднимайте его, когда упрётесь в пропускную способность.

Что делать, если вакансии подгружаются через JavaScript?

requests видит только исходный HTML. Если карточки появляются после выполнения скриптов, обычный запрос вернёт пустую страницу. В этом случае переходите на Selenium или Puppeteer, а токен reCAPTCHA v2 по-прежнему получайте через CaptchaAI и подставляйте в форму.

Законно ли собирать вакансии с job-бордов?

Собирайте только те данные, которые вы вправе обрабатывать, и уважайте условия использования площадки. Для читателей в РФ сбор персональных данных регулирует 152-ФЗ «О персональных данных»; для трансграничной аудитории действуют GDPR-подобные требования. Это не юридический совет — сверяйтесь с условиями конкретного сайта.

Как приложение отличает поддерживаемые типы CAPTCHA?

Этот проект рассчитан на reCAPTCHA v2 (метод userrecaptcha, токен g-recaptcha-response). CaptchaAI также решает reCAPTCHA v3, Cloudflare Turnstile и Challenge, GeeTest v3, а также image/OCR и grid-капчи; hCaptcha и FunCaptcha не поддерживаются. Под другой тип проверки поменяйте метод и поля запроса.

Смежные материалы

• Пайплайн лидогенерации
• Практики ограничения частоты запросов

Соберите данные о вакансиях в одном месте — начните с CaptchaAI.