빠른 캡차 풀이: 파이프라인의 지연 시간 줄이기
빠른 캡차 솔버는 올바른 순간에, 병렬로, 요청을 낭비하지 않고 호출할 수 있는 솔버입니다. Scarnox 풀이 호출은 토큰이 준비될 때 반환되므로, 소요 시간은 캡차 유형, 대상 사이트, 현재 부하에 따라 달라집니다. 이 가이드는 고정된 시간을 약속하지 않습니다. 대신 여러분의 코드에서 통제할 수 있는 것, 측정하는 방법, 많은 풀이를 동시에 실행하는 방법을 보여 줍니다. 그래서 실제로 쓰는 시간이 잘 쓰이도록 하는 것이 목표입니다.
풀이 시간을 결정하는 요소
세 가지가 중요하며, 그중 여러분의 손에 있는 것은 하나뿐입니다.
- 캡차 자체. 어떤 챌린지 유형은 다른 것보다 오래 걸립니다. 이는 여러분의 코드가 아니라 사이트 캡차의 특성입니다.
- 솔버의 부하. 바쁜 시간대에는 단일 호출이 느려질 수 있습니다. 재시도 로직은 그 호출을 일찍 포기하는 것이 아니라 그 상황을 처리해야 합니다.
- 내 파이프라인. 너무 일찍 풀이하거나, 모든 요청을 직렬로 처리하거나, 관련 없는 작업을 기다리는 것은 모두 제거할 수 있는 시간을 더합니다.
노력은 세 번째 항목에 집중하세요. 측정하고 바꿀 수 있는 부분이 바로 그곳입니다.
최적화하기 전에 측정하기
호출을 타이머로 감싸고 결과를 기록하세요. 풀이 시간과, 그 주변에서 코드가 쓰는 시간을 알고 싶습니다.
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를 열 개 처리해야 한다면, 하나씩 차례로 기다리는 것은 다른 것을 할 수 있었던 시간을 낭비합니다. 각 풀이는 대부분의 시간을 기다리며 보내는 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)
# 작게 시작하세요. 오류가 낮게 유지되는 동안에만 워커 수를 늘리세요.
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}")워커 수를 늘렸을 때 용량이나 제한과 관련된 것처럼 보이는 오류가 나타나면, 워커 수를 줄이세요. 병렬 처리는 서버가 그 속도를 따라갈 때만 도움이 됩니다.
너무 일찍 풀지 않기
가장 빠른 코드는 아직 쓰지 않을 토큰을 기다리지 않는 코드입니다. 토큰을 미리 요청해 저장해 두는 것은 흔한 실수입니다. 토큰은 발급 후 약 2분 뒤에 만료되므로, 저장된 토큰은 필요할 때 이미 오래된 경우가 많습니다. 폼이 막 제출되려는 시점에 풀이하세요. 자세한 이유는 캡차 토큰 만료: 원인과 해결에 있습니다.
나머지 경로는 짧게 유지하기
- 사이트마다 HTTP 세션 하나를 유지해서, 요청마다 다시 연결하지 않도록 하세요.
- 풀이와 제출 사이에 느린 파싱이나 데이터베이스 작업을 넣지 마세요.
- 내 다른 요청에는 적절한 타임아웃을 두세요. 풀이 호출은 토큰이 준비될 때까지 블록되므로 약 185초를 쓰지만, 다른 호출은 훨씬 짧아야 합니다.
- 프록시는 작업에 필요할 때만 보내세요. 프록시 작업 유형은
http://user:pass@host:port형식의proxy필드를 받지만, 프록시를 추가한다고 그 자체로 빨라지지는 않습니다.
만료되지 않는 것은 재사용하기
일부 작업은 여러 풀이에 걸쳐 안전하게 공유할 수 있습니다. 다음은 핫 패스에서 빼 두세요.
- sitekey. 페이지 템플릿마다 한 번 읽고 저장하세요. 토큰처럼 만료되지 않습니다. 폼이 실패하기 시작할 때만 다시 읽으세요.
- HTTP 연결. keep-alive 세션은 호출마다 새로운 TLS 핸드셰이크를 피하게 합니다. 요청마다 하나가 아니라 스레드나 워커마다 세션 하나를 만드세요.
- 파싱된 설정. 대상 목록, 프록시 목록, 설정은 시작할 때 한 번 불러오고, 풀이 함수 안에서 파일을 읽지 마세요.
이런 변경은 비용이 적게 들며, 모든 호출에 나타나는 오버헤드를 없앱니다.
평균이 아니라 꼬리를 측정하기
평균 풀이 시간은 배치를 붙잡는 느린 호출을 가립니다. 각 풀이 시간을 기록한 뒤 중앙값과 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가 거의 움직이지 않는다면, 시간은 코드의 다른 곳에서 쓰이고 있는 것입니다.
속도만이 아니라 결과도 확인하기
빠른 실패도 여전히 실패입니다. 오류를 반환한 풀이 수를 세고, 오류율이 올라가면 속도를 낮추라는 신호로 취급하세요. 실패한 풀이는 청구되지 않지만, 그에 쓰인 시간은 여전히 파이프라인이 잃어버린 시간입니다. 재시도 패턴 전체는 캡차 솔버 오류 처리에 있습니다.