为 LinkedIn 数据构建 Flask 回调服务器结合 webhook 与 MySQL 的异步爬取

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

pip install Flask mysql-connector-python pyyaml requests SQLAlchemy

token: YOUR_CRAWLBASE_TOKEN
crawler: linkedin-public-crawler

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;

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`)
);

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)

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.')

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)

python callback_server.py
ngrok http 5000

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

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

# 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

{
  "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."
    }
  ]
}

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