Scarnox · Blog

การแก้ captcha ให้เร็วขึ้น: วิธีลด latency ใน pipeline ของคุณ

captcha solver ที่เร็วคือตัวที่คุณเรียกได้ในจังหวะที่ถูกต้อง พร้อมกันหลายงาน โดยไม่เสียคำขอ การเรียก solve ของ Scarnox จะส่งกลับเมื่อ token พร้อม ดังนั้นระยะเวลาขึ้นอยู่กับประเภทของ captcha เว็บเป้าหมาย และโหลดในขณะนั้น คู่มือนี้ไม่สัญญาเวลาที่แน่นอน แต่แสดงสิ่งที่คุณควบคุมได้ในโค้ดของตัวเอง วิธีวัดผล และวิธีรันการแก้จำนวนมากพร้อมกัน เพื่อให้เวลาที่คุณใช้ถูกใช้อย่างคุ้มค่า

อะไรกำหนดเวลาการแก้

มีสามสิ่งที่สำคัญ และมีเพียงสิ่งเดียวที่อยู่ในมือคุณ

ตัว captcha เอง การท้าทายบางประเภทใช้เวลานานกว่าประเภทอื่น นั่นเป็นคุณสมบัติของ captcha ของเว็บ ไม่ใช่ของโค้ดคุณ
โหลดของ solver ช่วงที่งานหนาแน่นอาจทำให้การเรียกเดียวช้าลง ตรรกะการลองใหม่ควรจัดการเรื่องนี้ ไม่ใช่ยกเลิกการเรียกเร็วเกินไป
ไปป์ไลน์ของคุณเอง การแก้เร็วเกินไป การประมวลผลทีละคำขอ และการรองานที่ไม่เกี่ยวข้อง ล้วนเพิ่มเวลาที่คุณลดได้

ให้ความพยายามของคุณอยู่ที่ข้อที่สาม เพราะเป็นส่วนที่คุณวัดและเปลี่ยนแปลงได้

วัดก่อนจะปรับแต่ง

ห่อการเรียกด้วยตัวจับเวลาและบันทึกผล คุณต้องการรู้ทั้งเวลาการแก้และเวลาที่โค้ดของคุณใช้รอบ ๆ มัน

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")

รันสิ่งนี้กับหน้าที่คุณสนใจ ในเวลาที่คุณรันงานของคุณ เก็บตัวเลขไว้ ตัวเลขเหล่านั้นบอกคุณว่าการเปลี่ยนแปลงช่วยได้จริงหรือไม่

รันการแก้พร้อมกัน

การแก้แต่ละครั้งเป็นอิสระต่อกัน หากคุณมี sitekey สิบตัวที่ต้องจัดการ การรอทีละตัวเสียเวลาที่คุณอาจใช้กับตัวอื่นได้ thread pool เพียงพอสำหรับสคริปต์ส่วนใหญ่ เพราะการแก้แต่ละครั้งคือ HTTP หนึ่งคำขอที่ใช้เวลาส่วนใหญ่ไปกับการรอ

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)

# เริ่มจากจำนวนน้อย เพิ่มจำนวน worker เฉพาะเมื่อ error ยังต่ำอยู่
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}")

หากคุณเห็น error ที่ดูเหมือนปัญหาการจำกัดอัตราหรือความจุเมื่อเพิ่มจำนวน worker ให้ลดลง การทำงานพร้อมกันจะช่วยก็ต่อเมื่อเซิร์ฟเวอร์ตามทันเท่านั้น

อย่าแก้ล่วงหน้า

โค้ดที่เร็วที่สุดคือโค้ดที่ไม่รอ token ที่ยังไม่ได้ใช้ การขอ token ล่วงหน้าแล้วเก็บไว้เป็นความผิดพลาดที่พบบ่อยที่สุด token หมดอายุประมาณสองนาทีหลังออกให้ ดังนั้น token ที่เก็บไว้มักเก่าเกินไปเมื่อคุณต้องการใช้ แก้เมื่อฟอร์มกำลังจะถูกโพสต์ รายละเอียดการให้เหตุผลอยู่ในcaptcha token หมดอายุ: สาเหตุและวิธีแก้

เก็บเส้นทางที่เหลือให้สั้น

ใช้ HTTP session หนึ่งตัวต่อเว็บเพื่อไม่ต้องเชื่อมต่อใหม่ทุกคำขอ
อย่าทำงานประมวลผลที่ช้าหรือการทำงานกับฐานข้อมูลระหว่างการแก้และการโพสต์
ใช้ timeout ที่เหมาะสมกับคำขอของคุณเอง การเรียก solve ใช้ราว 185 วินาทีเพราะบล็อกจนกว่า token จะพร้อม คำขออื่นของคุณควรสั้นกว่ามาก
ส่ง proxy เฉพาะเมื่องานต้องการ ประเภทงาน proxy รับฟิลด์ proxy ในรูปแบบ http://user:pass@host:port แต่การเพิ่ม proxy ไม่ได้ทำให้เร็วขึ้นโดยตัวมันเอง

ใช้ซ้ำสิ่งที่ไม่หมดอายุ

งานบางอย่างสามารถแชร์ข้ามการแก้ได้อย่างปลอดภัย เก็บสิ่งเหล่านี้ให้พ้นจากเส้นทางที่ร้อนแรง

sitekey อ่าน sitekey หนึ่งครั้งต่อ template ของหน้าและเก็บไว้ มันไม่หมดอายุเหมือน token อ่านใหม่เฉพาะเมื่อฟอร์มเริ่มล้มเหลว
การเชื่อมต่อ HTTP session ที่ใช้ keep-alive หลีกเลี่ยงการทำ TLS handshake ใหม่ทุกการเรียก สร้าง session หนึ่งตัวต่อเธรดหรือต่อ worker ไม่ใช่หนึ่งตัวต่อคำขอ
config ที่ parse แล้ว โหลดรายการเป้าหมาย รายการ proxy และการตั้งค่าครั้งเดียวตอนเริ่มต้น แทนการอ่านไฟล์ในฟังก์ชัน solve

การเปลี่ยนแปลงเหล่านี้มีค่าใช้จ่ายน้อยและขจัดภาระที่ปรากฏในทุกการเรียก

วัดส่วนหาง ไม่ใช่ค่าเฉลี่ย

เวลาการแก้เฉลี่ยซ่อนการเรียกที่ช้าซึ่งทำให้ชุดงานค้าง บันทึกเวลาแก้แต่ละครั้ง แล้วดูค่ามัธยฐานและเปอร์เซ็นไทล์ที่ 95 ไปป์ไลน์ที่เสร็จภายในเวลาของงานที่ช้าที่สุดถูกจำกัดด้วยส่วนหาง การปรับวิธีลองใหม่การเรียกที่ช้าเพียงเล็กน้อยจึงอาจสำคัญกว่าการเปลี่ยนค่ามัธยฐาน

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]))

เปรียบเทียบสรุปก่อนและหลังการเปลี่ยนแปลงทุกครั้ง หาก p95 แทบไม่ขยับ แสดงว่าเวลาถูกใช้ไปที่อื่นในโค้ดของคุณ

ตรวจผลลัพธ์ ไม่ใช่แค่ความเร็ว

ความล้มเหลวที่เร็วยังคงเป็นความล้มเหลว นับการแก้ที่ส่ง error และถือว่าอัตรา error ที่เพิ่มขึ้นเป็นสัญญาณให้ชะลอลง การแก้ที่ล้มเหลวไม่ถูกเรียกเก็บเงิน แต่เวลาที่ใช้ไปกับมันก็ยังเป็นเวลาที่ไปป์ไลน์เสียไป รูปแบบการลองใหม่ฉบับเต็มอยู่ในการจัดการข้อผิดพลาดของ captcha solver