Một đội outsourcing ở TP.HCM cần theo dõi giá, đánh giá và tin tuyển dụng của hàng chục đối thủ mỗi ngày trên Shopee, G2, LinkedIn — nhưng cứ vài chục trang lại dính reCAPTCHA hoặc Turnstile, làm gãy script. CaptchaAI giải các CAPTCHA này qua API để pipeline chạy liên tục, không cần canh giải tay.
Nguồn dữ liệu nghiên cứu thị trường và loại CAPTCHA thường gặp
Mỗi nhóm nguồn thường gắn với một loại CAPTCHA đặc trưng — biết trước giúp chọn đúng method:
| Loại nguồn | Ví dụ | Loại CAPTCHA |
|---|---|---|
| Trang đánh giá | G2, Trustpilot, Yelp | reCAPTCHA v2/v3 |
| Trang tuyển dụng | LinkedIn, Indeed | Cloudflare Challenge |
| Danh bạ doanh nghiệp | YellowPages, Crunchbase | Cloudflare Turnstile, reCAPTCHA |
| Mạng xã hội | Twitter/X, Reddit | Nhiều loại khác nhau |
| Cơ sở dữ liệu bằng sáng chế | USPTO, Google Patents | reCAPTCHA v2 |
| Dữ liệu chính phủ | Hồ sơ SEC, dữ liệu điều tra dân số | CAPTCHA hình ảnh |
Xây dựng pipeline thu thập dữ liệu với CaptchaAI
Luồng xử lý gồm bốn bước, lặp lại cho mọi request bị chặn:
- Phát hiện
sitekeytrong HTML của trang vừa fetch - Gửi task tới
in.php, nhận vềtask_id - Polling
res.phpmỗi 5 giây tới khi có token - Gắn token vào request tiếp theo (
g-recaptcha-responsehoặccf-turnstile-response) rồi gửi lại
Cùng một class dùng cho cả trang đánh giá lẫn thư mục doanh nghiệp, chỉ khác hàm parse:
import requests
import time
import re
import csv
import os
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
def solve_captcha(params):
params["key"] = API_KEY
resp = requests.get("https://ocr.captchaai.com/in.php", params=params)
if not resp.text.startswith("OK|"):
raise Exception(f"Submit: {resp.text}")
task_id = resp.text.split("|")[1]
for _ in range(60):
time.sleep(5)
result = requests.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get", "id": task_id,
})
if result.text == "CAPCHA_NOT_READY":
continue
if result.text.startswith("OK|"):
return result.text.split("|", 1)[1]
raise Exception(f"Solve: {result.text}")
raise TimeoutError()
class MarketResearchCollector:
def __init__(self):
self.session = requests.Session()
self.session.headers["User-Agent"] = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/120.0.0.0"
)
def fetch(self, url):
"""Fetch a page, solving CAPTCHAs as needed."""
resp = self.session.get(url)
# Detect reCAPTCHA
match = re.search(
r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text
)
if match:
token = solve_captcha({
"method": "userrecaptcha",
"googlekey": match.group(1),
"pageurl": url,
})
resp = self.session.post(url, data={
"g-recaptcha-response": token,
})
# Detect Turnstile
match = re.search(
r'data-sitekey=["\']([0-9x][A-Za-z0-9_-]+)["\']', resp.text
)
if match and "cf-turnstile" in resp.text:
token = solve_captcha({
"method": "turnstile",
"sitekey": match.group(1),
"pageurl": url,
})
resp = self.session.post(url, data={
"cf-turnstile-response": token,
})
return resp.text
def collect_reviews(self, urls):
"""Collect review data from multiple pages."""
reviews = []
for url in urls:
try:
html = self.fetch(url)
page_reviews = self._parse_reviews(html)
reviews.extend(page_reviews)
print(f" Collected {len(page_reviews)} reviews from {url}")
time.sleep(2) # Polite delay
except Exception as e:
print(f" Error on {url}: {e}")
return reviews
def collect_company_profiles(self, urls):
"""Collect company profile data."""
profiles = []
for url in urls:
try:
html = self.fetch(url)
profile = self._parse_profile(html)
if profile:
profiles.append(profile)
print(f" Collected: {profile.get('name', 'Unknown')}")
time.sleep(2)
except Exception as e:
print(f" Error on {url}: {e}")
return profiles
def _parse_reviews(self, html):
"""Extract review data from HTML."""
reviews = []
# Generic review extraction patterns
for match in re.finditer(
r'class="review-text"[^>]*>(.*?)</div>', html, re.DOTALL
):
reviews.append({
"text": match.group(1).strip()[:500],
})
return reviews
def _parse_profile(self, html):
"""Extract company profile from HTML."""
name = re.search(r'<h1[^>]*>(.*?)</h1>', html)
desc = re.search(
r'class="description"[^>]*>(.*?)</div>', html, re.DOTALL
)
return {
"name": name.group(1).strip() if name else None,
"description": desc.group(1).strip()[:300] if desc else None,
}
def export_csv(self, data, filename):
"""Export collected data to CSV."""
if not data:
return
keys = data[0].keys()
with open(filename, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=keys)
writer.writeheader()
writer.writerows(data)
print(f"Exported {len(data)} records to {filename}")
Chạy thử pipeline
Khởi tạo collector rồi truyền danh sách URL cần thu thập:
collector = MarketResearchCollector()
# Collect competitor reviews
review_urls = [
"https://example-reviews.com/product/competitor-a",
"https://example-reviews.com/product/competitor-b",
"https://example-reviews.com/product/competitor-c",
]
reviews = collector.collect_reviews(review_urls)
collector.export_csv(reviews, "competitor_reviews.csv")
# Collect company profiles
profile_urls = [
"https://example-directory.com/company/alpha-corp",
"https://example-directory.com/company/beta-inc",
]
profiles = collector.collect_company_profiles(profile_urls)
collector.export_csv(profiles, "company_profiles.csv")
Mẹo mở rộng quy mô pipeline
Khi pipeline vượt quá vài chục trang/ngày, siết lại theo các mốc sau:
- Khoảng cách request: 2-5 giây/trang, đừng gửi dồn dập
- Collector đồng thời: 5-10 luồng là đủ cho quy mô vừa
- Nguồn request (proxy): đa dạng hoá khi vượt 100 trang/giờ
- Chống trùng lặp: khấu trừ theo hàm băm trước khi lưu
- Lịch chạy: hàng ngày hoặc hàng tuần cho dữ liệu xu hướng, không cần real-time
Nếu một trang chặn hoàn toàn crawler, kết hợp CaptchaAI với việc đa dạng nguồn request và mẫu request giống hành vi thật thay vì gửi dồn dập — xem bài dùng proxy hiệu quả khi thu thập dữ liệu tránh bị chặn.
Ứng dụng thực tế trong nghiên cứu thị trường
Theo dõi giá đối thủ trên sàn thương mại điện tử
Theo dõi giá, khuyến mãi và tồn kho trên Shopee, Lazada, Tiki hoặc các sàn quốc tế. Dữ liệu thường cần theo dõi:
- Giá niêm yết và giá sau mã giảm giá theo từng SKU
- Số lượng đã bán / tồn kho hiển thị công khai
- Tần suất đổi giá theo khung giờ (flash sale)
Nhiều đội ngũ tại Việt Nam chạy việc này hàng ngày để cập nhật bảng giá B2B — chỉ thu thập dữ liệu công khai trên catalogue, không đụng thông tin cá nhân người mua.
Phân tích cảm nhận thương hiệu qua đánh giá
Thu thập đánh giá, xếp hạng từ các nền tảng review rồi tổng hợp cảm nhận trên nhiều nguồn để so sánh thương hiệu theo thời gian.
Phân tích xu hướng tuyển dụng
Quét tin tuyển dụng trên các job board để nắm xu hướng ngành, cụ thể:
- Mức lương tham khảo theo vị trí và cấp bậc
- Kỹ năng xuất hiện lặp lại trong mô tả công việc
- Số lượng tin đăng mới theo tuần, theo ngành
Theo dõi xu hướng bằng sáng chế
Thu thập hồ sơ bằng sáng chế từ cơ sở dữ liệu công khai để theo dõi hướng đổi mới và R&D đối thủ.
Câu hỏi thường gặp
Giải CAPTCHA cho dữ liệu nghiên cứu thị trường tốn bao nhiêu?
Một dự án nghiên cứu thị trường điển hình kích hoạt CAPTCHA trên khoảng 30% số trang. Với 1.000 trang/ngày, ước tính ~300 lần giải, tổng chi phí khoảng $0,5–3 — gói BASIC ($15/tháng, 5 thread) thường đủ.
Nên dùng polling hay webhook khi thu thập dữ liệu ở quy mô lớn?
Polling (res.php định kỳ) đơn giản, đủ dùng cho vài trăm task/ngày. Ở quy mô hàng nghìn task song song, webhook (pingback) giảm request thừa vì CaptchaAI tự gửi kết quả về URL của bạn.
Thu thập dữ liệu công khai để nghiên cứu thị trường có hợp pháp tại Việt Nam không?
Thường được phép, nhưng vẫn cần kiểm tra điều khoản dịch vụ từng trang.
Nếu dữ liệu chạm tới thông tin cá nhân, áp dụng Nghị định 13/2023/NĐ-CP về bảo vệ dữ liệu cá nhân — tránh thu thập email, số điện thoại hay dữ liệu định danh nếu không có căn cứ pháp lý rõ ràng.
Cần bao nhiêu thread để chạy pipeline 5.000-10.000 trang/ngày?
CaptchaAI tính phí theo số thread đồng thời, không giới hạn số lần giải mỗi thread. Với ~3.000 CAPTCHA/ngày (ước tính từ 10.000 trang, tỷ lệ kích hoạt ~30%), gói STANDARD ($30/tháng, 15 thread) thường đủ cho 10-15 collector song song; nhiều hơn thì chuyển lên ADVANCE ($90/tháng, 50 thread).
