Создание Flask Callback-сервера для данных LinkedInасинхронный краулинг с webhook и MySQL

При краулинге в большом объёме медленнее всего ждать. Синхронный скрапер отправляет запрос, блокируется до рендеринга и возврата тяжело защищённой страницы, парсит её и только тогда переходит к следующему URL. Выстройте за этим тысячи URL, и большую часть времени скрипт будет простаивать. Асинхронный паттерн переворачивает всё: вы передаёте пакет URL краулеру, он выполняет медленный рендеринг на собственной инфраструктуре и POST-запросом отправляет каждый готовый результат на webhook, который вы контролируете. Это руководство строит именно такую систему: Flask callback-сервер, получающий результаты от асинхронного Crawlbase Crawler и сохраняющий их в MySQL.

Целевая площадка примера, LinkedIn, но сначала прочитайте предупреждение ниже, потому что то, что вы храните, важнее того, как вы это храните. Это руководство намеренно ограничено публичными, неперсональными данными: полями страниц компаний и публичным текстом вакансий, но не профилями отдельных участников. Реальная учебная ценность здесь, архитектура: асинхронный краулинг плюс callback-сервер плюс база данных, и этот паттерн работает одинаково для любого публичного источника.

Что вы создадите

Небольшая асинхронная система краулинга на Python с тремя движущимися частями и общей базой данных MySQL. Вместо одного блокирующего скрипта работа разделена так, чтобы медленный краулинг происходил вне вашей машины, а результаты поступали по мере готовности:

• Скрипт асинхронных запросов Crawler, отправляющий список публичных URL в асинхронный Crawlbase Crawler и сохраняющий идентификатор запроса (RID) для каждого.
• Flask callback-сервер, получающий каждый готовый краулинг как HTTP POST, распаковывающий его и сохраняющий сырой payload.
• Процессор, читающий сохранённые payload по расписанию, извлекающий публичные поля и записывающий структурированные строки.
• MySQL-схема с таблицей отслеживания запросов и таблицами для публичных хранимых полей.

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

Почему асинхронный подход и почему callback-сервер

Страница компании LinkedIn или публичная вакансия рендерится на стороне клиента и защищена агрессивными антибот-мерами, поэтому один запрос медленный и нередко встречает вызов. При синхронном подходе для длинного списка URL скрипт блокируется на каждом по очереди. Асинхронный Crawler принимает ваш URL, немедленно возвращает идентификатор запроса, а затем выполняет медленный рендеринг и повторные попытки на собственной инфраструктуре. Когда получен чистый ответ, результат POST-запросом отправляется на ваш webhook.

Именно push-модель требует callback-сервера. Ваш endpoint не опрашивает и не ждёт; он просто находится в готовности, и каждый результат приземляется по завершении, в каком бы порядке ни завершались краулинги. Движок Crawler отправляет тело сжатым gzip, поэтому endpoint должен распаковать его перед чтением. Разделение запроса, получения и обработки на три скрипта позволяет системе поглощать большой пакет без блокировки какого-либо шага другим. Если нужна более широкая информация о движке, см. руководство по извлечению данных с Crawlbase Crawler.

Предварительные требования

Несколько вещей для подготовки. Всё делается быстро.

Python 3.8 или выше. Проверьте командой python3 --version. Если нет, установите с python.org.

MySQL 8. Работающий MySQL-сервер, к которому вы можете подключиться локально. Официальное руководство по установке охватывает все платформы.

Аккаунт Crawlbase и обычный (TCP) токен. Зарегистрируйтесь, откройте панель управления и скопируйте токен. LinkedIn обслуживается обычным краулером запросов, поэтому здесь используйте TCP-токен, а не JavaScript. Обращайтесь с токеном как с паролем и не добавляйте его в систему контроля версий.

Способ открыть localhost. Crawler отправляет результаты на публичный URL, поэтому в процессе разработки нужен туннель вроде ngrok для доступа к вашему локальному Flask-приложению.

Настройка проекта

Создайте изолированное виртуальное окружение, затем установите нужные библиотеки.

python3 -m venv .venv
source .venv/bin/activate

pip install Flask mysql-connector-python pyyaml requests SQLAlchemy

В Windows активируйте командой .venv\Scripts\activate вместо строки с source. Четыре библиотеки выполняют основную работу: Flask, это webhook-сервер, SQLAlchemy с mysql-connector-python работает с базой данных, requests отправляет запросы краулинга, а pyyaml читает токен из файла настроек. Создайте файл settings.yml рядом со скриптами для хранения токена и имени вашего Crawler.

token: YOUR_CRAWLBASE_TOKEN
crawler: linkedin-public-crawler

Шаг 1: Проектирование MySQL-схемы

У схемы две задачи: отслеживать каждый запрос краулинга на протяжении его жизненного цикла и хранить публичные поля. Создайте пользователя, базу данных и таблицы. Выполните это в командной строке MySQL.

CREATE USER 'linkedincrawler'@'localhost' IDENTIFIED BY 'linked1nS3cret';
CREATE DATABASE linkedin_crawler_db;
GRANT ALL PRIVILEGES ON linkedin_crawler_db.* TO 'linkedincrawler'@'localhost';
USE linkedin_crawler_db;

Теперь таблицы. Таблица crawl_requests является управляющей таблицей для всего асинхронного процесса: каждый отправляемый URL получает одну строку, отслеживаемую по полю status по мере перехода через состояния waiting, затем received, затем processed. Столбец crawlbase_rid связывает строку с идентификатором запроса, возвращаемым Crawler, это единственный ключ для сопоставления входящего callback с инициировавшим его запросом.

CREATE TABLE IF NOT EXISTS `crawl_requests` (
  `id` INT AUTO_INCREMENT PRIMARY KEY,
  `url` TEXT NOT NULL,
  `status` VARCHAR(30) NOT NULL,
  `crawlbase_rid` VARCHAR(255) NOT NULL
);

CREATE INDEX `idx_crawl_requests_status` ON `crawl_requests` (`status`);
CREATE INDEX `idx_crawl_requests_rid` ON `crawl_requests` (`crawlbase_rid`);

Целевые таблицы хранят только публичные, неперсональные данные компании. По одной строке на страницу компании плюс дочерняя таблица для публичных вакансий, на которые ссылается эта страница. Нигде нет столбца с именем человека, должностью или текстом профиля. Именно это является конкретным воплощением границы приватности в самой схеме.

CREATE TABLE IF NOT EXISTS `company_pages` (
  `id` INT AUTO_INCREMENT PRIMARY KEY,
  `crawl_request_id` INT NOT NULL,
  `company_name` VARCHAR(255),
  `industry` VARCHAR(255),
  `description` TEXT,
  FOREIGN KEY (`crawl_request_id`) REFERENCES `crawl_requests`(`id`)
);

CREATE TABLE IF NOT EXISTS `company_job_postings` (
  `id` INT AUTO_INCREMENT PRIMARY KEY,
  `company_page_id` INT NOT NULL,
  `title` VARCHAR(255),
  `location` VARCHAR(255),
  `description` TEXT,
  FOREIGN KEY (`company_page_id`) REFERENCES `company_pages`(`id`)
);

Шаг 2: Определение ORM

Сопоставьте таблицы с классами Python через SQLAlchemy, чтобы остальной код работал с объектами, а не с сырым SQL. Сохраните в lib/database.py. Классы точно отражают схему: CrawlRequest для отслеживания, CompanyPage для публичных полей компании и дочерний JobPosting для каждой публичной вакансии.

from typing import List
from sqlalchemy import ForeignKey, create_engine
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class CrawlRequest(Base):
    __tablename__ = 'crawl_requests'
    id: Mapped[int] = mapped_column(primary_key=True)
    url: Mapped[str]
    status: Mapped[str]
    crawlbase_rid: Mapped[str]
    company_page: Mapped['CompanyPage'] = relationship(back_populates='crawl_request')

class CompanyPage(Base):
    __tablename__ = 'company_pages'
    id: Mapped[int] = mapped_column(primary_key=True)
    company_name: Mapped[str]
    industry: Mapped[str]
    description: Mapped[str]
    crawl_request_id: Mapped[int] = mapped_column(ForeignKey('crawl_requests.id'))
    crawl_request: Mapped['CrawlRequest'] = relationship(back_populates='company_page')
    job_postings: Mapped[List['JobPosting']] = relationship(back_populates='company_page')

class JobPosting(Base):
    __tablename__ = 'company_job_postings'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    location: Mapped[str]
    description: Mapped[str]
    company_page_id: Mapped[int] = mapped_column(ForeignKey('company_pages.id'))
    company_page: Mapped['CompanyPage'] = relationship(back_populates='job_postings')

def create_database_session():
    url = 'mysql+mysqlconnector://linkedincrawler:linked1nS3cret@localhost:3306/linkedin_crawler_db'
    engine = create_engine(url, echo=True)
    return Session(engine)

create_database_session возвращает сессию, которую импортируют все остальные скрипты. Строка подключения содержит пользователя, пароль, хост и базу данных, созданные на шаге 1; измените их здесь при необходимости.

Шаг 3: Отправка URL в асинхронный Crawler

Этот скрипт читает список публичных URL, отправляет каждый в асинхронный Crawler и сохраняет возвращённый RID со статусом waiting. Ключевые параметры: callback=true сообщает Crawler POST-ить результаты, а не возвращать их inline; crawler= называет Crawler, который вы создадите в панели управления. Сохраните как crawl.py, а публичные URL компаний и вакансий поместите по одному на строку в urls.txt.

import requests
import urllib.parse
import json
import yaml
from json import JSONDecodeError
from lib.database import CrawlRequest, create_database_session

settings = yaml.safe_load(open('settings.yml'))
token = settings.get('token')
crawler = settings.get('crawler')

if not token or not crawler:
    print('Set your token and crawler name in settings.yml')
    exit()

urls = open('urls.txt', 'r').readlines()
api = 'https://api.crawlbase.com?token={0}&callback=true&crawler={1}&url={2}&autoparse=true'
session = create_database_session()

for url in urls:
    url = url.strip()
    if not url:
        continue
    encoded = urllib.parse.quote(url, safe='')
    api_url = api.format(token, crawler, encoded)
    print(f'Requesting crawl for {url}')
    try:
        response = requests.get(api_url)
        rid = json.loads(response.text)['rid']
        request_row = CrawlRequest(url=url, crawlbase_rid=str(rid), status='waiting')
        session.add(request_row)
        session.commit()
    except JSONDecodeError:
        print(f'Could not decode response for {url}')

print('Done pushing crawl requests.')

Каждый вызов возвращает небольшое JSON-тело вида {"rid": 12341234}. Скрипт сохраняет этот RID в crawl_requests со статусом waiting и сразу переходит к следующему URL, не блокируясь на реальном краулинге. Параметр autoparse=true просит Crawler возвращать структурированные поля, а не сырой HTML, это то, что читает процессор на шаге 6. В этом и есть весь смысл асинхронной модели: отправка ста URL занимает секунды, а медленная работа происходит в другом месте.

Шаг 4: Создание Flask callback-сервера

Это сердце системы. Crawler POST-ит каждый готовый результат на единственный маршрут. Ваша задача, валидировать запрос, распаковать тело и сохранить payload, чтобы процессор мог его подхватить. Crawler отправляет RID в заголовке с именем rid и два заголовка статуса: PC-Status (статус Crawlbase) и Original-Status (статус целевого сайта). Вы сохраняете только результаты, где оба равны 200. Сохраните как callback_server.py.

import gzip
import os
from flask import Flask, request
from lib.database import CrawlRequest, create_database_session

app = Flask(__name__)
session = create_database_session()
os.makedirs('./data', exist_ok=True)

def header_status(name):
    value = request.headers.get(name)
    return int(value.split(',')[0]) if value else None

@app.route('/crawlbase_crawler_callback', methods=['POST'])
def crawlbase_crawler_callback():
    rid = request.headers.get('rid')
    encoding = request.headers.get('Content-Encoding')

    if rid is None:
        return ('', 204)
    if rid == 'dummyrequest':
        print('Callback server is working')
        return ('', 204)
    if header_status('PC-Status') != 200 or header_status('Original-Status') != 200:
        return ('', 204)

    crawl_request = session.query(CrawlRequest).filter_by(crawlbase_rid=rid, status='waiting').first()
    if crawl_request is None:
        print(f'No waiting request for rid {rid}')
        return ('', 204)

    body = request.data
    if encoding == 'gzip':
        try:
            body = gzip.decompress(body)
        except OSError:
            pass

    with open(f'./data/{rid}.json', 'wb') as f:
        f.write(body)

    crawl_request.status = 'received'
    session.commit()
    print(f'Received rid {rid}')
    return ('', 201)

if __name__ == '__main__':
    app.run(port=5000)

Разберём проверки, потому что каждая из них важна. Отсутствующий rid означает, что запрос поступил не от Crawler, он отбрасывается. RID со значением dummyrequest, это тестовый пинг, который платформа отправляет для проверки доступности endpoint; он логируется и обрабатывается досрочно. Проверка статуса игнорирует всё, что не является чистым 200 по обоим фронтам. Затем вы ищете RID в crawl_requests со статусом waiting: если такой строки нет, callback не соответствует вашему запросу и игнорируется. Только после всех этих проверок вы распаковываете и сохраняете тело, затем переключаете строку в состояние received. Endpoint никогда не блокируется; он записывает файл и немедленно возвращает ответ, оставаясь отзывчивым даже при потоке callback-ов.

Шаг 5: Открытие сервера и регистрация Crawler

Crawler нужен публичный URL для отправки результатов. При работающем Flask-приложении на порту 5000 откройте туннель.

python callback_server.py
ngrok http 5000

ngrok выводит публичный HTTPS URL. Полный callback-маршрут, это URL плюс путь, например https://your-subdomain.ngrok.io/crawlbase_crawler_callback. Убедитесь, что endpoint работает, с помощью тестового пинга до использования Crawler.

curl -i -X POST 'http://localhost:5000/crawlbase_crawler_callback' \
  -H 'rid: dummyrequest' \
  -H 'Content-Type: gzip/json' \
  -H 'Content-Encoding: gzip'

В логе Flask должно появиться сообщение Callback server is working. Теперь перейдите в панель управления Crawlbase, откройте страницу создания Crawler, дайте Crawler то же имя, что указано в settings.yml, и вставьте полный ngrok callback URL. LinkedIn обслуживается обычным (TCP) Crawler, поэтому выберите этот тип. После сохранения Crawler знает, куда отправлять результаты.

Шаг 6: Обработка полученных payload-ов в структурированные строки

Callback-сервер только сохраняет сырые payload-ы. Отдельный процессор запускается по расписанию, подхватывает всё в статусе received, извлекает публичные поля, записывает структурированные строки и помечает запрос как processed. Разделение получения и обработки означает, что медленная запись в базу данных никогда не блокирует webhook. Сохраните как process.py.

import json
import sched
import time
from lib.database import CrawlRequest, CompanyPage, JobPosting, create_database_session

INTERVAL_SECONDS = 60
BATCH_LIMIT = 10

def process():
    session = create_database_session()
    received = session.query(CrawlRequest).filter_by(status='received').limit(BATCH_LIMIT).all()

    if not received:
        print('No received requests to process.')
        return

    for req in received:
        with open(f'./data/{req.crawlbase_rid}.json') as f:
            data = json.load(f)

        page = CompanyPage(
            company_name=data.get('name'),
            industry=data.get('industry'),
            description=data.get('description'),
        )
        page.crawl_request_id = req.id
        session.add(page)

        for job in data.get('jobs', []):
            posting = JobPosting(
                title=job.get('title'),
                location=job.get('location'),
                description=job.get('description'),
            )
            posting.company_page = page
            session.add(posting)

        req.status = 'processed'

    session.commit()

def process_and_reschedule():
    process()
    scheduler.enter(INTERVAL_SECONDS, 1, process_and_reschedule)

if __name__ == '__main__':
    scheduler = sched.scheduler(time.monotonic, time.sleep)
    process_and_reschedule()
    scheduler.run()

Процессор читает только безличные поля компании и вакансий из распарсенного payload: название компании, отрасль, публичное описание, а также название, местоположение и описание каждой публичной вакансии. Он никогда не обращается к полям уровня физического лица, даже если они присутствуют в payload. Поддержание такого узкого списка извлечения, вторая половина границы приватности, дополняющая схему. Цикл sched перезапускает process каждые 60 секунд и обрабатывает не более десяти запросов за проход, что удерживает потребление памяти на постоянном уровне при большом накопленном объёме.

Запуск всего пайплайна

При зарегистрированном Crawler запустите три части, каждую в отдельном терминале с активным виртуальным окружением. Порядок важен: callback-сервер и процессор должны быть запущены до отправки запросов, иначе ранние callback-ы придут без получателя.

# terminal 1: webhook (already running, plus ngrok)
python callback_server.py

# terminal 2: scheduled processor
python process.py

# terminal 3: push the batch
python crawl.py

По мере выполнения crawl.py в crawl_requests появляются строки со статусом waiting. Через несколько минут, по мере того как Crawler завершает каждую страницу, callback-сервер переключает их в received и записывает JSON-файл в папку ./data. На следующем проходе процессор читает эти файлы, заполняет таблицы company_pages и company_job_postings и помечает запросы как processed. За этим можно наблюдать в реальном времени во вкладке мониторинга Crawler в панели управления, которая показывает состояние каждого запроса.

Как выглядят сохранённые данные

После полного запуска в целевых таблицах хранятся чистые, безличные записи компаний. Одна обработанная страница компании выглядит так при чтении обратно как JSON.

{
  "company_name": "Example Robotics",
  "industry": "Industrial Automation",
  "description": "We design warehouse automation systems.",
  "job_postings": [
    {
      "title": "Backend Engineer",
      "location": "Remote, EU",
      "description": "Build and operate our ingestion services."
    }
  ]
}

Каждое поле здесь, это то, что компания публикует о себе. Никаких людей, никаких контактов, никаких профилей. Так и задумано, и именно это делает датасет защитимым.

Масштабирование и передача дополнительного контекста

Архитектура масштабируется без структурных изменений: больший urls.txt означает больше строк waiting, Crawler поглощает очередь, и callback-ы приземляются по мере завершения краулингов. Для сопоставления payload-ов с вашим собственным контекстом прикрепляйте данные с помощью параметра callback_headers при отправке запроса. Crawler эхом возвращает эти заголовки в callback-е, поэтому можно передавать, например, идентификатор пакета, не сохраняя его в URL.

raw_headers = f'BATCH-ID:{batch_id}|SOURCE:public-company-page'
encoded_headers = urllib.parse.quote(raw_headers, safe='')
# append &callback_headers={encoded_headers} to the api url

На принимающей стороне читайте их как обычные заголовки запроса: request.headers.get('BATCH-ID'). Подробнее о поддержании работоспособности крупных запусков против защищённых целей см. руководства о том, как парсить сайты без блокировок, и о создании масштабируемого пайплайна веб-данных.

Законно ли парсить LinkedIn?

Это раздел, который нужно изучить до написания продакшен-кода, а не после. Пользовательское соглашение LinkedIn и его политика в отношении запрещённого программного обеспечения и расширений прямо запрещают парсинг и автоматический сбор данных, и LinkedIn применяет эти условия. Эта позиция остаётся в силе независимо от аккуратности вашего инструментария. Код в этом руководстве обеспечивает техническую сторону; он не делает парсинг LinkedIn совместимым с условиями LinkedIn. Прочитайте Пользовательское соглашение и robots.txt LinkedIn и воспринимайте оба документа как границу своих действий.

Аспект данных не менее важен. Большинство контента LinkedIn является персональными данными об идентифицируемых людях: имена, история работы, заголовки, связи и публикации. По GDPR в Европе и CCPA в Калифорнии обработка персональных данных требует правового основания, и люди имеют права, включая право на удаление своих данных. Здесь есть и реальная судебная практика: в деле hiQ Labs v. LinkedIn суды США рассматривали парсинг публичных профилей в контексте Закона о компьютерном мошенничестве и злоупотреблениях, однако этот процесс был узконаправленным, юрисдикционно-специфичным и не санкционировал парсинг в целом и не отменял договорные условия LinkedIn или законодательство о защите данных. Законность определяется данными, методом, юрисдикцией и соглашениями, которые вас обязывают, поэтому относитесь с подозрением к огульным заявлениям о том, что «публичность означает свободный доступ».

Именно поэтому данное руководство ограничено именно так. Оно хранит только публичную, неперсональную информацию компаний: названия компаний, отрасли, публичные описания и публичный текст вакансий, который компания публикует о себе. Оно никогда не строит профили физических лиц, не обращается к чему-либо за логином и не собирает персональные данные участников. Для любого реального или коммерческого использования правильный путь, официальные API LinkedIn и партнёрские программы, предоставляющие санкционированный, структурированный доступ в рамках условий LinkedIn. Если вашему проекту нужны данные уровня участников, именно этот путь, или официальное соглашение о данных, является ответом, а не более хитрый скрапер. В случае сомнений относительно конкретного использования проконсультируйтесь с квалифицированным юристом. Подробнее об общедоступном подходе см. наш обзор парсинга LinkedIn.

Часто задаваемые вопросы

Зачем использовать асинхронный краулер вместо синхронного скрипта?

Синхронный скрипт блокируется на каждом URL, пока защищённая страница рендерится и возвращается, поэтому длинный список выполняется в основном в режиме простоя. Асинхронный Crawler принимает URL, немедленно возвращает идентификатор запроса и выполняет медленный рендеринг и повторные попытки на собственной инфраструктуре, затем POST-запросом отправляет готовый результат на ваш webhook. Отправка большого пакета занимает секунды, а результаты поступают по мере завершения, а не по одному медленному запросу за раз.

Что на самом деле делает Flask callback-сервер?

Он открывает один POST-маршрут, который Crawler вызывает с каждым готовым результатом. Обработчик читает заголовок rid, проверяет, что и PC-Status, и Original-Status равны 200, убеждается, что RID соответствует запросу в статусе waiting, распаковывает gzip-тело, сохраняет payload на диск и переключает запрос в состояние received. Он возвращается немедленно и никогда не блокируется, поэтому остаётся отзывчивым даже при всплеске callback-ов.

Почему получение и обработка разделены на два скрипта?

Чтобы медленная запись в базу данных не задерживала webhook. Единственная задача callback-сервера, быстро принять и сохранить. Отдельный плановый процессор читает сохранённые payload-ы небольшими пакетами, извлекает публичные поля, записывает структурированные строки и помечает каждый запрос как processed. Разделение двух задач позволяет системе поглощать большой объём callback-ов без обратного давления на каждую из сторон.

Нужен ли JavaScript-токен или обычный?

Обычный (TCP) токен. LinkedIn обслуживается обычным Crawler, поэтому в панели управления выбирается этот тип, а TCP-токен указывается в settings.yml. Асинхронный Crawler всё равно обрабатывает ротацию и повторные попытки за кулисами; тип токена просто говорит ему, какой путь запроса использовать для цели.

Как защитить webhook?

Принимайте только POST-запросы, требуйте секретный токен в пользовательском заголовке или параметре URL, который вы проверяете в каждом вызове, и убедитесь, что ожидаемые заголовки rid, PC-Status и Original-Status присутствуют до того, как доверять payload-у. Избегайте IP-фильтрации, поскольку исходные адреса ротируются и могут меняться без предупреждения. Проверки статуса и RID в примере, начальная точка, а не исчерпывающее решение.

Безопасно ли хранить данные LinkedIn таким образом?

Только если ограничиться публичными, неперсональными данными, как делает это руководство: названия компаний, отрасли, публичные описания и публичный текст вакансий. Хранение профилей участников, имён, связей или других персональных данных влечёт применение Пользовательского соглашения LinkedIn и законов вроде GDPR и CCPA и выходит за рамки данного руководства. Для данных уровня участников или коммерческого использования применяйте официальные API LinkedIn и партнёрские программы, а не скрапер.

IK

Ian Kalvin

Инженер технической поддержки · Crawlbase

Инженер технической поддержки в Crawlbase, пишет с переднего края того, что на самом деле ломается в продакшене при скрейпинге и в прокси-конфигурациях.

NZ

Neil Zamora

Старший архитектор · Crawlbase

Старший архитектор в Crawlbase, сосредоточен на системах, стоящих за краулингом в больших масштабах: ротация прокси, устойчивость к anti-bot и API, скрывающие эту сложность.