Giải captcha nhanh hơn: cách giảm độ trễ trong pipeline
Một bộ giải captcha nhanh là bộ bạn có thể gọi đúng thời điểm, song song, mà không lãng phí yêu cầu. Lời gọi giải của Scarnox trả về khi token đã sẵn sàng, nên thời lượng phụ thuộc vào loại captcha, website mục tiêu và tải hiện tại. Hướng dẫn này không hứa một thời gian cố định. Nó chỉ cho bạn những gì bạn kiểm soát được trong code của mình, cách đo lường, và cách chạy nhiều lần giải cùng lúc, để thời gian bạn bỏ ra được dùng một cách hiệu quả.
Điều gì quyết định thời gian giải
Có ba yếu tố quan trọng, và chỉ một trong số đó nằm trong tay bạn:
- Bản thân captcha. Một số loại thử thách mất nhiều thời gian hơn các loại khác. Đó là thuộc tính của captcha trên website, không phải của code bạn.
- Tải trên bộ giải. Một khoảng thời gian bận rộn có thể làm một lời gọi chậm hơn. Logic thử lại nên xử lý việc đó, chứ không bỏ lời gọi sớm.
- Pipeline của bạn. Giải quá sớm, xử lý tuần tự mọi yêu cầu và chờ các công việc không liên quan đều tốn thời gian mà bạn có thể loại bỏ.
Hãy tập trung nỗ lực vào yếu tố thứ ba. Đó là phần bạn có thể đo lường và thay đổi.
Đo trước khi tối ưu
Bọc lời gọi bằng một bộ đếm thời gian và ghi lại kết quả. Bạn muốn biết thời gian giải và thời gian mã của bạn tiêu tốn xung quanh nó:
import time
import requests
API_KEY = "scarnox_YOURAPIKEY"
URL = "https://api.scarnox.com/api/tasks/create"
def timed_solve(sitekey: str, page_url: str) -> dict:
started = time.perf_counter()
r = requests.post(
URL,
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"captcha_type": "PopularCaptchaTaskProxyless",
"site_key": sitekey,
"site_url": page_url,
},
timeout=185,
)
elapsed = time.perf_counter() - started
data = r.json()
if data.get("errorId") != 0:
return {"ok": False, "error": data.get("errorCode"), "seconds": elapsed}
return {"ok": True, "token": data["solution"]["token"], "seconds": elapsed}
result = timed_solve("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup")
print(result["ok"], round(result["seconds"], 2), "s")
Chạy đoạn này trên các trang bạn quan tâm, vào những thời điểm bạn chạy công việc của mình. Hãy giữ lại các con số. Chúng cho bạn biết một thay đổi có thật sự giúp ích hay không.
Chạy các lần giải song song
Các lần giải độc lập với nhau. Nếu bạn có mười sitekey cần xử lý, chờ từng cái một sẽ lãng phí thời gian mà bạn có thể dành cho những cái khác. Một thread pool là đủ cho hầu hết script, vì mỗi lần giải là một yêu cầu HTTP dành phần lớn thời gian để chờ:
from concurrent.futures import ThreadPoolExecutor
jobs = [
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/signup"),
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/register"),
("10000000-ffff-ffff-ffff-000000000001", "https://example.com/contact"),
]
def run(job):
sitekey, page_url = job
return timed_solve(sitekey, page_url)
# Bắt đầu nhỏ. Chỉ tăng số worker khi lỗi vẫn thấp.
with ThreadPoolExecutor(max_workers=5) as pool:
results = list(pool.map(run, jobs))
for r in results:
status = "ok" if r["ok"] else r["error"]
print(f"{r['seconds']:.2f} s {status}")
Nếu bạn thấy các lỗi trông giống vấn đề giới hạn tần suất hoặc sức chứa khi tăng số worker, hãy giảm xuống. Song song chỉ có ích khi máy chủ theo kịp.
Đừng giải sớm
Đoạn mã nhanh nhất là đoạn mã không chờ một token mà nó chưa cần dùng. Lấy token trước và lưu lại là lỗi thường gặp. Token hết hạn khoảng hai phút sau khi được cấp, nên một token đã lưu thường đã cũ khi bạn cần đến nó. Hãy giải khi form sắp được gửi. Lập luận chi tiết nằm trong captcha token hết hạn: nguyên nhân và cách khắc phục.
Giữ phần còn lại của luồng ngắn
- Giữ một phiên HTTP cho mỗi website để bạn không phải kết nối lại cho mỗi yêu cầu.
- Không chạy xử lý phân tích chậm hay công việc cơ sở dữ liệu giữa lúc giải và lúc gửi.
- Dùng timeout hợp lý cho các yêu cầu của riêng bạn. Lời gọi giải dùng khoảng 185 giây vì nó bị chặn cho đến khi token sẵn sàng. Các lời gọi khác của bạn nên ngắn hơn nhiều.
- Chỉ gửi proxy khi tác vụ cần. Các loại tác vụ proxy nhận trường
proxytheo định dạnghttp://user:pass@host:port, nhưng thêm proxy không tự động làm nhanh hơn.
Tái sử dụng những gì không hết hạn
Một số công việc có thể dùng chung an toàn giữa các lần giải. Hãy để chúng ra khỏi đường đi nóng:
- Sitekey. Đọc sitekey một lần cho mỗi mẫu trang và lưu lại. Nó không hết hạn như token. Chỉ đọc lại khi form bắt đầu thất bại.
- Kết nối HTTP. Một phiên keep-alive tránh bắt tay TLS mới cho mỗi lời gọi. Tạo một phiên cho mỗi thread hoặc mỗi worker, không phải một phiên cho mỗi yêu cầu.
- Cấu hình đã phân tích. Tải danh sách mục tiêu, danh sách proxy và cài đặt một lần khi khởi động, thay vì đọc file bên trong hàm giải.
Những thay đổi này tốn ít công sức và loại bỏ phần overhead xuất hiện trong mọi lời gọi.
Đo phần đuôi, không đo trung bình
Thời gian giải trung bình che giấu những lời gọi chậm làm nghẽn cả một lô. Ghi lại thời gian của từng lần giải, rồi xem trung vị và phân vị thứ 95. Một pipeline kết thúc đúng bằng thời gian của công việc chậm nhất sẽ bị giới hạn bởi phần đuôi, nên một thay đổi nhỏ trong cách thử lại các lời gọi chậm có thể quan trọng hơn bất kỳ thay đổi nào với trung vị:
import statistics
def summarise(seconds: list[float]) -> str:
ordered = sorted(seconds)
p95 = ordered[int(0.95 * (len(ordered) - 1))]
return f"median {statistics.median(ordered):.2f} s, p95 {p95:.2f} s, n={len(ordered)}"
print(summarise([r["seconds"] for r in results]))
So sánh bản tóm tắt trước và sau mỗi thay đổi. Nếu p95 gần như không đổi, thời gian đang bị tiêu tốn ở một chỗ khác trong code của bạn.
Kiểm tra kết quả, không chỉ tốc độ
Một lỗi nhanh vẫn là lỗi. Đếm số lần giải trả về lỗi, và xem tỉ lệ lỗi tăng là tín hiệu để chậm lại. Các lần giải thất bại không bị tính phí, nhưng thời gian dành cho chúng vẫn là thời gian pipeline của bạn bị mất. Mẫu thử lại đầy đủ nằm trong xử lý lỗi bộ giải captcha.