Khi một cổng nhà cung cấp bật CAPTCHA, pipeline giám sát không chậm lại — nó dừng hẳn, và bạn chỉ biết vào sáng hôm sau, lúc báo cáo tồn kho trống một nửa. Luồng xử lý gọn nhất gồm bốn nhịp: crawler nhận diện thử thách trong response, gửi sitekey sang CaptchaAI qua in.php, polling res.php lấy token, rồi POST lại đúng form trên cùng session. Dưới đây là trọn luồng đó bằng Python.
Nguồn dữ liệu nào trong chuỗi cung ứng bật CAPTCHA
| Nhóm nguồn | Loại CAPTCHA | Dữ liệu | Nhịp kiểm tra |
|---|---|---|---|
| Cổng nhà cung cấp | reCAPTCHA v2 | Tồn kho, giá | Hàng ngày |
| Hãng vận chuyển | Cloudflare Turnstile | Cước, ETA | Hàng giờ |
| Catalog nhà sản xuất | CAPTCHA hình ảnh | Thông số, MOQ | Hàng tuần |
| Cổng hải quan | reCAPTCHA v2 | Thuế suất, mã HS | Hàng ngày |
| Cảng vụ | CAPTCHA hình ảnh | Lịch tàu, ùn tắc | 6 giờ một lần |
Với đội mua hàng ở TP.HCM hay nhà máy ở Bình Dương, hai dòng đáng tiền nhất là hải quan và cảng vụ: lịch tàu ở Cát Lái lệch một ngày là kế hoạch sản xuất lệch theo.
Dựng monitor đa nhà cung cấp bằng Python
SupplyChainMonitor duyệt từng nhà cung cấp, phát hiện CAPTCHA trong HTML rồi chọn cách giải theo captcha_type.
import requests
import time
import re
import json
import base64
from datetime import datetime
CAPTCHAAI_KEY = "YOUR_API_KEY"
CAPTCHAAI_URL = "https://ocr.captchaai.com"
def solve_recaptcha(sitekey, pageurl):
resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
"key": CAPTCHAAI_KEY, "method": "userrecaptcha",
"googlekey": sitekey, "pageurl": pageurl, "json": 1,
})
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,
})
data = result.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError("Timeout")
def solve_turnstile(sitekey, pageurl):
resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
"key": CAPTCHAAI_KEY, "method": "turnstile",
"sitekey": sitekey, "pageurl": pageurl, "json": 1,
})
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,
})
data = result.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError("Timeout")
def solve_image(image_bytes):
img_b64 = base64.b64encode(image_bytes).decode()
resp = requests.post(f"{CAPTCHAAI_URL}/in.php", data={
"key": CAPTCHAAI_KEY, "method": "base64",
"body": img_b64, "json": 1,
})
task_id = resp.json()["request"]
for _ in range(20):
time.sleep(3)
result = requests.get(f"{CAPTCHAAI_URL}/res.php", params={
"key": CAPTCHAAI_KEY, "action": "get",
"id": task_id, "json": 1,
})
data = result.json()
if data["request"] != "CAPCHA_NOT_READY":
return data["request"]
raise TimeoutError("Timeout")
class SupplyChainMonitor:
def __init__(self, suppliers, proxy=None):
self.suppliers = suppliers
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",
})
def check_all(self):
"""Check inventory and pricing across all suppliers."""
report = {
"timestamp": datetime.now().isoformat(),
"suppliers": {},
}
for supplier in self.suppliers:
try:
data = self._check_supplier(supplier)
report["suppliers"][supplier["name"]] = {
"status": "success",
"data": data,
}
except Exception as e:
report["suppliers"][supplier["name"]] = {
"status": "error",
"error": str(e),
}
time.sleep(3)
return report
def _check_supplier(self, supplier):
url = supplier["url"]
resp = self.session.get(url, timeout=30)
# Handle CAPTCHA based on type
captcha_type = supplier.get("captcha_type")
if captcha_type and self._has_captcha(resp.text):
resp = self._solve_captcha(resp, url, supplier)
from bs4 import BeautifulSoup
soup = BeautifulSoup(resp.text, "html.parser")
return {
"products": self._extract_inventory(soup),
"last_updated": self._extract_date(soup),
}
def _has_captcha(self, html):
return any(tag in html.lower() for tag in [
'data-sitekey', 'g-recaptcha', 'cf-turnstile', 'captcha',
])
def _solve_captcha(self, resp, url, supplier):
captcha_type = supplier.get("captcha_type", "recaptcha")
sitekey = supplier.get("sitekey", "")
if not sitekey:
match = re.search(r'data-sitekey="([^"]+)"', resp.text)
sitekey = match.group(1) if match else ""
if captcha_type == "turnstile":
token = solve_turnstile(sitekey, url)
return self.session.post(url, data={"cf-turnstile-response": token})
elif captcha_type == "image":
match = re.search(r'src="(/captcha[^"]+)"', resp.text)
if match:
img_resp = self.session.get(url.rstrip("/") + match.group(1))
answer = solve_image(img_resp.content)
return self.session.post(url, data={"captcha": answer})
else:
token = solve_recaptcha(sitekey, url)
return self.session.post(url, data={"g-recaptcha-response": token})
return resp
def _extract_inventory(self, soup):
items = []
for row in soup.select("table.inventory tr, .product-row"):
cols = row.select("td, .col")
if len(cols) >= 3:
items.append({
"sku": cols[0].get_text(strip=True),
"stock": cols[1].get_text(strip=True),
"price": cols[2].get_text(strip=True),
})
return items
def _extract_date(self, soup):
date_el = soup.select_one(".last-updated, .update-time")
return date_el.get_text(strip=True) if date_el else ""
# Configure suppliers
suppliers = [
{
"name": "Supplier A",
"url": "https://supplier-a.example.com/inventory",
"captcha_type": "recaptcha",
"sitekey": "6Lc_xxxxxxx",
},
{
"name": "Carrier B",
"url": "https://carrier-b.example.com/rates",
"captcha_type": "turnstile",
"sitekey": "0x4AAAAAAA_xxx",
},
{
"name": "Manufacturer C",
"url": "https://manufacturer-c.example.com/catalog",
"captcha_type": "image",
},
]
monitor = SupplyChainMonitor(
suppliers=suppliers,
proxy="http://user:[email protected]:5000",
)
report = monitor.check_all()
print(json.dumps(report, indent=2))
Điểm dễ sai nhất là tên field token: reCAPTCHA trả g-recaptcha-response, Turnstile trả cf-turnstile-response, CAPTCHA hình ảnh trả thẳng chuỗi ký tự. Request POST sau đó phải đi trên đúng session đã tải trang.
Theo dõi cước vận chuyển qua Turnstile
Trang tra cước của hãng vận chuyển thường đứng sau Cloudflare Turnstile.
class ShippingRateTracker:
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",
})
def get_rates(self, carrier_url, origin, destination, weight):
"""Fetch shipping rates, handling Turnstile CAPTCHA."""
resp = self.session.get(carrier_url, timeout=30)
sitekey_match = re.search(r'data-sitekey="([^"]+)"', resp.text)
if sitekey_match:
token = solve_turnstile(sitekey_match.group(1), carrier_url)
resp = self.session.post(carrier_url, data={
"origin": origin,
"destination": destination,
"weight": weight,
"cf-turnstile-response": token,
})
if resp.status_code == 200:
return resp.json().get("rates", [])
return []
Turnstile thường có token trong dưới 10 giây với tỷ lệ giải thành công cao, đủ để chạy tuần tự vài chục tuyến.
Cảnh báo khi tồn kho chạm ngưỡng
Hàm dưới đây chỉ cảnh báo đúng lúc SKU tụt dưới ngưỡng, thay vì nhắc lại mỗi giờ.
def monitor_with_alerts(monitor, alert_thresholds, check_interval=3600):
"""Continuously monitor and alert on inventory changes."""
previous_data = {}
while True:
report = monitor.check_all()
for supplier, info in report["suppliers"].items():
if info["status"] != "success":
continue
for product in info["data"].get("products", []):
sku = product["sku"]
stock = product.get("stock", "")
# Parse stock level
try:
stock_qty = int(re.sub(r'\D', '', stock))
except ValueError:
continue
key = f"{supplier}:{sku}"
prev_qty = previous_data.get(key, stock_qty)
threshold = alert_thresholds.get(sku, 10)
if stock_qty < threshold and prev_qty >= threshold:
print(f"ALERT: {supplier} - {sku} dropped to {stock_qty}")
previous_data[key] = stock_qty
time.sleep(check_interval)
Khi lên production, đẩy dòng print sang Slack hoặc email nội bộ kèm tên nhà cung cấp.
Cần bao nhiêu thread cho lịch giám sát
CaptchaAI tính giá theo thread (luồng giải đồng thời), không theo mỗi lần giải. Con số cần thiết phụ thuộc vào số nguồn chạy cùng lúc, không phải tổng số nguồn:
- 30–40 nhà cung cấp rải đều trong ngày: BASIC ($15/tháng, 5 thread).
- 300–500 nguồn gom vào một cửa sổ 15 phút: ADVANCE ($90/tháng, 50 thread).
- Nhiều thị trường, pipeline chạy 24/7: PREMIUM ($170/tháng, 100 thread) trở lên.
Giá niêm yết bằng USD, nên giữ nguyên đơn vị này khi lập dự toán.
Dự phòng khi một nguồn im lặng
- Tách nguồn trọng yếu khỏi nguồn tham khảo, để một lỗi không làm hỏng cả báo cáo.
- Quy định rõ khi nào dữ liệu cũ còn dùng được, khi nào phải báo sự cố.
- Cổng liên tục dựng CAPTCHA là tín hiệu rủi ro nguồn cung, không chỉ lỗi crawler.
Xử lý sự cố thường gặp
| Vấn đề | Nguyên nhân | Cách xử lý |
|---|---|---|
| Parser trả mảng rỗng | Nhà cung cấp đổi giao diện | Cập nhật CSS selector |
| Lượt nào cũng gặp CAPTCHA | Nhịp request quá dày | Giãn khoảng cách kiểm tra |
| Phiên hết hạn giữa chừng | Cổng đặt timeout ngắn | Giữ session xuyên suốt |
Câu hỏi thường gặp
CaptchaAI giải được loại CAPTCHA nào?
reCAPTCHA v2/v3 (kể cả Enterprise), Cloudflare Turnstile và Challenge, GeeTest v3, CAPTCHA hình ảnh/OCR, grid-image, BLS. CaptchaAI không hỗ trợ hCaptcha và FunCaptcha; GeeTest v4 sắp ra mắt; CaptchaFox, Friendly Captcha và Lemin ở giai đoạn beta.
Lấy token Turnstile mất bao lâu?
Dưới 10 giây theo mức công bố. reCAPTCHA v2 dưới 60 giây, nên đặt timeout crawler rộng hơn con số đó.
Làm sao để ít gặp CAPTCHA hơn?
Bám theo tốc độ dữ liệu thực sự đổi: cước theo giờ, tồn kho theo ngày, catalog theo tuần. Thêm độ trễ ngẫu nhiên giữa các nhà cung cấp.
Dữ liệu thu về cần lưu ý gì về tuân thủ?
Chỉ lưu trường phục vụ quyết định mua hàng và giữ log ai chạy job nào. Có dữ liệu cá nhân thì rà lại quy trình theo Nghị định 13/2023/NĐ-CP.
Bài viết liên quan
Đừng để một thử thách CAPTCHA che mất tầm nhìn chuỗi cung ứng — lấy API key CaptchaAI và cho monitor chạy qua mọi cổng nhà cung cấp.
