Một trình tổng hợp việc làm gom tin tuyển dụng từ nhiều job board về một chỗ, chuẩn hóa lại rồi cho phép tìm kiếm — nghe đơn giản, nhưng rào cản lớn nhất khi tự động thu thập là thử thách CAPTCHA chặn ngay ở trang kết quả. Bài này dựng một pipeline Python hoàn chỉnh giải quyết đúng vấn đề đó: scrape nhiều board, gặp reCAPTCHA v2 thì để CaptchaAI giải, sau đó lưu dữ liệu sạch vào SQLite để tra cứu.
Luồng xử lý tách thành bốn phần rõ ràng: một model dữ liệu, một lớp scraper biết tự giải CAPTCHA, các scraper riêng cho từng board, và một runner điều phối. Với một team HR-tech hay agency tuyển dụng ở TP.HCM muốn theo dõi thị trường việc làm công khai, kiến trúc này đủ gọn để chạy trên một máy và mở rộng dần khi thêm nguồn. Nếu bạn lưu dữ liệu có thể nhận dạng cá nhân, hãy ghi log tối thiểu và cân nhắc Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân ngay từ khâu thiết kế lưu trữ.
Pipeline gồm bốn phần
- Model dữ liệu —
JobListingvàJobDatabaseđịnh nghĩa cấu trúc một tin và lớp lưu trữ SQLite. - Lớp scraper tự giải CAPTCHA — tải trang và gọi CaptchaAI khi phát hiện reCAPTCHA v2.
- Scraper theo từng board — cấu hình URL và CSS selector cho mỗi nguồn.
- Runner điều phối — chạy toàn bộ và ghi kết quả vào database.
Kiến trúc tổng thể của trình tổng hợp
Dữ liệu đi một chiều: nhiều job board đổ vào một scraper có tích hợp bộ giải CAPTCHA, qua bước chuẩn hóa, rồi vào SQLite. Tách rời từng khối giúp bạn thêm board mới hoặc đổi nơi lưu trữ mà không phải viết lại phần còn lại của pipeline.
[Job Board A] ──┐
[Job Board B] ──┼──> Scraper + CAPTCHA Solver ──> Normalizer ──> SQLite DB
[Job Board C] ──┘
Mô hình dữ liệu cho tin tuyển dụng
JobListing là một dataclass mô tả một tin tuyển dụng: tiêu đề, công ty, địa điểm, khoảng lương, nguồn và thời điểm scrape. JobDatabase bọc SQLite lại — bảng jobs đặt ràng buộc UNIQUE trên cột url, nên hai lần scrape cùng một tin sẽ không tạo bản ghi trùng vì INSERT OR IGNORE lo phần đó. Chuẩn hóa ở đây là thống nhất lương về số, gắn nguồn cho mỗi tin và để search() lọc theo từ khóa lẫn địa điểm, nhờ vậy tin từ nhiều board vẫn so sánh được với nhau.
# 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()
Lớp scraper tự xử lý CAPTCHA
Đây là phần lõi. BaseScraper.fetch() tải trang như bình thường; nếu phát hiện dấu hiệu reCAPTCHA (data-sitekey hoặc g-recaptcha), nó gọi CaptchaAI lấy token rồi POST lại form. Quy trình giải chạy đúng bốn bước quen thuộc của CaptchaAI: trích sitekey từ HTML, gửi task tới in.php với method userrecaptcha kèm googlekey và pageurl, polling res.php mỗi 5 giây cho đến khi có kết quả, rồi đưa token vào field g-recaptcha-response. Vòng lặp 24 lần với time.sleep(5) đặt trần chờ khoảng hai phút trước khi báo timeout — đủ rộng cho reCAPTCHA v2 mà không treo tiến trình vô hạn.
# 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")
Scraper cho từng job board
GenericJobScraper nhận vào một mẫu URL và một bộ CSS selector, nên cùng một class dùng được cho nhiều board khác nhau — thêm board mới chỉ là thêm cấu hình chứ không phải viết code mới. scrape_search() duyệt qua từng trang kết quả và dừng sớm khi một trang không còn thẻ tin nào. Riêng _extract_salary() dùng regex để tách khoảng lương từ text tự do; định dạng lương mỗi board mỗi khác nên đây thường là chỗ bạn cần tinh chỉnh đầu tiên.
# 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)
Chạy toàn bộ pipeline tổng hợp việc làm
main.py ráp mọi thứ lại: khai báo danh sách BOARDS, chạy scraper cho từng từ khóa, ghi kết quả vào database rồi thử một truy vấn tìm kiếm. time.sleep(5) giữa các board là phép lịch sự tối thiểu để không dội request quá dày lên một nguồn. Muốn thêm board, bạn chỉ cần bổ sung một mục vào BOARDS với base_url và bộ selector tương ứng.
# 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()
Xử lý sự cố thường gặp
| Vấn đề | Nguyên nhân | Cách xử lý |
|---|---|---|
| Tin tuyển dụng bị trùng | Cùng một job xuất hiện trên nhiều trang | Khử trùng theo URL nhờ ràng buộc UNIQUE |
| Không tách được mức lương | Định dạng lương không theo chuẩn | Chỉnh lại regex _extract_salary cho từng board |
| Gặp CAPTCHA ở mọi trang | Session không được giữ lại giữa các request | Tái sử dụng self.session cho toàn bộ crawler |
| Không có tin nào sau khi giải | Form CAPTCHA cần JavaScript render | Chuyển sang Selenium + CaptchaAI |
Câu hỏi thường gặp
Pipeline này giải loại CAPTCHA nào?
reCAPTCHA v2. Lớp BaseScraper phát hiện data-sitekey, gửi task tới in.php với method userrecaptcha, polling res.php để lấy token rồi đặt vào field g-recaptcha-response. Nếu job board của bạn dùng loại khác, CaptchaAI cũng hỗ trợ reCAPTCHA v3, Cloudflare Turnstile và GeeTest v3.
Chi phí giải CAPTCHA cho một trình tổng hợp việc làm là bao nhiêu?
CaptchaAI tính tiền theo thread (luồng giải đồng thời), không theo mỗi lần giải. Gói BASIC ($15/tháng, 5 thread) đủ cho một crawler chạy vài board, và mỗi thread giải không giới hạn số CAPTCHA trong tháng. Khi cần scrape song song nhiều board hơn, tăng lên STANDARD ($30/tháng, 15 thread) hoặc ADVANCE ($90/tháng, 50 thread).
Khi nào nên chuyển từ requests sang Selenium?
Khi form CAPTCHA cần JavaScript render hoặc token phải nằm trong một phiên trình duyệt thật. Nếu giải xong mà trang vẫn trả về rỗng, đó là dấu hiệu cần một trình duyệt headless (Selenium hoặc Playwright) kết hợp CaptchaAI thay cho requests.
Lưu và tìm kiếm dữ liệu việc làm thế nào?
JobDatabase dùng SQLite với ràng buộc UNIQUE trên URL để tự khử trùng. Method search() lọc theo từ khóa tiêu đề và địa điểm, sắp xếp theo thời điểm scrape mới nhất — đủ cho một dashboard nội bộ mà không cần thêm hạ tầng.
Hướng dẫn liên quan
Tổng hợp dữ liệu tin tuyển dụng từ nhiều nguồn — bắt đầu với CaptchaAI.
