Một job theo dõi giá hỏng hiếm khi báo lỗi: nó vẫn chạy đúng lịch, vẫn ghi log "OK", chỉ có cột giá bắt đầu rỗng vì sàn trả về trang CAPTCHA. Cách sửa là dạy con bot coi thử thách là một bước bình thường: phát hiện CAPTCHA, trích sitekey, gửi task sang CaptchaAI, nhận token, gửi lại request.
Trước tiên: bot của bạn có biết mình đang dính CAPTCHA không?
Response trả HTTP 200, parser chạy trót lọt và trả chuỗi rỗng — không có exception để bắt. Vòng lặp theo dõi cần kiểm tra trước khi parse: data-sitekey, g-recaptcha hoặc cf-turnstile là đủ tin cậy. Log mỗi lần gặp CAPTCHA: tỷ lệ tăng vọt báo nhịp quét quá dày.
Loại CAPTCHA nào đang chờ bạn trên từng sàn
| Nền tảng | Loại CAPTCHA | Kích hoạt khi |
|---|---|---|
| Amazon | reCAPTCHA v2 | Truy cập dày |
| Walmart | reCAPTCHA v3 + Cloudflare | Điểm bot thấp |
| Target | Cloudflare Turnstile | Truy cập tự động |
| eBay | reCAPTCHA v2 | Trang tìm kiếm |
Cả ba loại trên đều được CaptchaAI hỗ trợ, cùng Cloudflare Challenge, GeeTest v3 và CAPTCHA ảnh/OCR. CaptchaAI không giải hCaptcha và FunCaptcha (Arkose Labs); GeeTest v4 sắp ra mắt; CaptchaFox, Friendly Captcha, Lemin đang beta. Xác định loại CAPTCHA của sàn trước khi viết integration.
Class RetailMonitor: vòng lặp tự xử lý CAPTCHA
RetailMonitor gói trọn luồng: tải trang, phát hiện thử thách, gửi task tới in.php, polling res.php, gửi lại request kèm token.
import requests
import time
import re
import json
from datetime import datetime
from bs4 import BeautifulSoup
CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"
def solve_captcha(method, sitekey, pageurl, **kwargs):
data = {
"key": CAPTCHAAI_KEY, "method": method,
"googlekey": sitekey, "pageurl": pageurl, "json": 1,
}
data.update(kwargs)
resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data=data)
task_id = resp.json()["request"]
for _ in range(60):
time.sleep(5)
result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
"key": CAPTCHAAI_KEY, "action": "get",
"id": task_id, "json": 1,
})
r = result.json()
if r["request"] != "CAPCHA_NOT_READY":
return r["request"]
raise TimeoutError("Timeout")
class RetailMonitor:
def __init__(self, proxy=None):
self.session = requests.Session()
if proxy:
self.session.proxies = {"http": proxy, "https": proxy}
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
})
def check_product(self, url):
"""Check single product's price and availability."""
resp = self.session.get(url, timeout=30)
# Handle CAPTCHA
if self._has_captcha(resp.text):
resp = self._solve_and_retry(resp.text, url)
soup = BeautifulSoup(resp.text, "html.parser")
return {
"url": url,
"title": self._text(soup, "h1, .product-title, #productTitle"),
"price": self._text(soup, ".price, .a-price .a-offscreen, .prod-price"),
"availability": self._text(soup, "#availability, .stock-status, .fulfillment"),
"in_stock": self._check_stock(soup),
"timestamp": datetime.now().isoformat(),
}
def monitor_products(self, product_urls, interval_sec=1800):
"""Continuously monitor products for changes."""
history = {}
while True:
for url in product_urls:
try:
current = self.check_product(url)
# Check for changes
prev = history.get(url)
if prev:
changes = self._detect_changes(prev, current)
if changes:
self._alert(current["title"], changes)
history[url] = current
time.sleep(3)
except Exception as e:
print(f"Error checking {url}: {e}")
print(f"Cycle complete: {len(product_urls)} products checked")
time.sleep(interval_sec)
def track_prices(self, product_urls, output_file="prices.json"):
"""Single price check across all products."""
results = []
for url in product_urls:
try:
data = self.check_product(url)
results.append(data)
time.sleep(3)
except Exception as e:
results.append({"url": url, "error": str(e)})
with open(output_file, "w") as f:
json.dump(results, f, indent=2)
print(f"Tracked {len(results)} products → {output_file}")
return results
def _has_captcha(self, html):
return any(tag in html.lower() for tag in [
'data-sitekey', 'g-recaptcha', 'cf-turnstile', 'captcha',
])
def _solve_and_retry(self, html, url):
match = re.search(r'data-sitekey="([^"]+)"', html)
if not match:
return self.session.get(url)
sitekey = match.group(1)
if 'cf-turnstile' in html:
token = solve_captcha("turnstile", sitekey, url)
return self.session.post(url, data={"cf-turnstile-response": token})
else:
token = solve_captcha("userrecaptcha", sitekey, url)
return self.session.post(url, data={"g-recaptcha-response": token})
def _text(self, soup, selector):
el = soup.select_one(selector)
return el.get_text(strip=True) if el else ""
def _check_stock(self, soup):
stock_el = soup.select_one("#availability, .stock-status")
if stock_el:
text = stock_el.get_text(strip=True).lower()
return "in stock" in text or "available" in text
return None
def _detect_changes(self, prev, current):
changes = []
if prev["price"] != current["price"]:
changes.append(f"Price: {prev['price']} → {current['price']}")
if prev["in_stock"] != current["in_stock"]:
status = "In Stock" if current["in_stock"] else "Out of Stock"
changes.append(f"Stock: → {status}")
return changes
def _alert(self, title, changes):
print(f"ALERT [{title}]: {', '.join(changes)}")
# Usage
monitor = RetailMonitor(
proxy="http://user:[email protected]:5000"
)
products = [
"https://store.example.com/product/abc123",
"https://store.example.com/product/def456",
"https://store.example.com/product/ghi789",
]
# One-time price check
results = monitor.track_prices(products)
# Or continuous monitoring (every 30 min)
# monitor.monitor_products(products, interval_sec=1800)
Hai chi tiết quyết định code chạy hay không: tên trường token phải khớp loại (cf-turnstile-response, g-recaptcha-response), và time.sleep(3) là biến rẻ nhất để giảm CAPTCHA. Cần API key để chạy đoạn trên.
Quét cả danh mục thay vì từng sản phẩm
Quét theo trang danh mục rẻ hơn: mỗi trang có hàng chục sản phẩm mà chỉ tốn một lần giải.
def scan_category(base_url, category, max_pages=20):
"""Scan an entire product category for stock status."""
monitor = RetailMonitor(
proxy="http://user:[email protected]:5000"
)
all_products = []
for page in range(1, max_pages + 1):
url = f"{base_url}/{category}?page={page}"
resp = monitor.session.get(url, timeout=30)
if monitor._has_captcha(resp.text):
resp = monitor._solve_and_retry(resp.text, url)
soup = BeautifulSoup(resp.text, "html.parser")
items = soup.select(".product-card, .s-result-item")
if not items:
break
for item in items:
all_products.append({
"name": monitor._text(item, ".product-name, .a-text-normal"),
"price": monitor._text(item, ".price, .a-price"),
"stock": monitor._text(item, ".stock, .a-color-success"),
"url": item.select_one("a")["href"] if item.select_one("a") else "",
})
time.sleep(3)
return all_products
Vòng lặp dừng khi hết item, tránh quét mù 20 trang trên danh mục chỉ có bốn.
Ví dụ Việt Nam: đối chiếu giá một SKU trên Shopee, Lazada và Tiki
Bài toán quen thuộc với team e-commerce ở TP.HCM: cùng một mã SKU của bạn niêm yết trên Shopee, Lazada và Tiki, mỗi sáng cần biết giá công khai đứng ở đâu. Mỗi sàn một cơ chế, nên hàm dưới tạo RetailMonitor riêng:
def compare_product_across_stores(product_name, stores):
"""Compare prices across retailers for the same product."""
results = []
for store in stores:
monitor = RetailMonitor(proxy=store.get("proxy"))
search_url = f"{store['base_url']}/search?q={product_name}"
try:
resp = monitor.session.get(search_url, timeout=30)
if monitor._has_captcha(resp.text):
resp = monitor._solve_and_retry(resp.text, search_url)
soup = BeautifulSoup(resp.text, "html.parser")
first_result = soup.select_one(".product-card, .s-result-item")
if first_result:
results.append({
"store": store["name"],
"price": monitor._text(first_result, ".price"),
"in_stock": "in stock" in first_result.get_text().lower(),
})
except Exception as e:
results.append({"store": store["name"], "error": str(e)})
time.sleep(5)
results.sort(key=lambda x: x.get("price", "zzzz"))
return results
Chỉ lấy dữ liệu công khai, phục vụ nghiên cứu thị trường hoặc danh mục của chính bạn. Nếu bản ghi kèm thông tin người bán, Nghị định 13/2023/NĐ-CP là lý do để tối giản dữ liệu lưu lại.
Nhịp quét: cân giữa dữ liệu tươi và số lần dính CAPTCHA
| Nhóm sản phẩm | Nhịp kiểm tra |
|---|---|
| Hàng mở bán giới hạn | Mỗi 1 phút |
| Điện tử | Mỗi 30 phút |
| Thời trang | Mỗi 2 giờ |
| Hàng tiêu dùng nhanh | Mỗi 4 giờ |
| Hàng phổ thông | Mỗi 12 giờ |
Phép tính thực tế: 1.000 sản phẩm quét mỗi 30 phút, ~5% số trang bung CAPTCHA. Vì các lần giải rải đều, số CAPTCHA đồng thời hiếm khi vượt vài luồng — BASIC ($15/tháng, 5 thread) thường đủ, vì thứ bạn mua là thread song song chứ không phải số lần giải.
Bảng chẩn đoán khi số liệu bắt đầu sai
| Vấn đề | Nguyên nhân | Cách xử lý |
|---|---|---|
| Trang nào cũng bung CAPTCHA | Request dồn dày | Tăng time.sleep |
| Giá sai hoặc rỗng | Giá do JS render | Dùng Selenium/Playwright |
| Trang "robot check" | Điểm rủi ro cao | Gửi header thực tế |
| Thiếu trường dữ liệu | Sàn đổi HTML | Cập nhật selector |
Câu hỏi thường gặp
Làm sao biết response là trang CAPTCHA chứ không phải trang sản phẩm?
Kiểm tra HTML trước khi parse. Đừng dựa vào HTTP status — trang CAPTCHA vẫn trả 200.
Giải CAPTCHA xong rồi mà request vẫn bị chặn, vì sao?
Thường do sai tên trường token hoặc gửi lại quá muộn. Token có thời hạn ngắn — gửi request ngay khi res.php trả kết quả.
CaptchaAI có giải được hCaptcha không?
Không. hCaptcha và FunCaptcha (Arkose Labs) không được hỗ trợ, GeeTest v4 sắp ra mắt. reCAPTCHA v2/v3, Turnstile, GeeTest v3, CAPTCHA ảnh thì có.
Giá niêm yết bằng USD, thanh toán thế nào?
Bảng giá là USD, từ BASIC ($15/tháng, 5 thread) tới VIP-3 ($7.500/tháng, 5.000 thread), trả bằng thẻ quốc tế. Chi phí cố định theo gói, không tăng theo số CAPTCHA.
Đọc thêm
Lấy API key CaptchaAI và chạy RetailMonitor thử trên năm sản phẩm.
