如何在 Python 爬虫中区分验证码与封锁

py -m venv .venv
.\.venv\Scripts\python.exe -m pip install capskip requests

import re

# A tag whose class attribute names a widget, or an altcha-widget tag. A link, a hidden
# input or a sentence that merely contains the name does not match.
CLS = r"<[^>]*\bclass\s*=\s*[\"'][^\"']*(?<![\w-])%s(?![\w-])"
WIDGET = re.compile(CLS % "(?:g-recaptcha|h-captcha|cf-turnstile|frc-captcha|captchafox)"
                    + r"|<altcha-widget\b", re.I)
TITLE = re.compile(r"<title[^>]*>(.*?)</title>", re.I | re.S)
BLOCK = ("access denied", "error 1020", "unusual traffic", "you have been blocked")

def classify(resp, expect=None):
    """expect: a string only the real page contains, for pages that carry a widget anyway."""
    code, h = resp.status_code, resp.headers
    if h.get("cf-mitigated", "").lower() == "challenge":
        return "cloudflare_challenge"                 # Cloudflare docs
    if 300 <= code < 400:                             # allow_redirects=False callers
        return "unknown"
    html = "html" in h.get("content-type", "").lower()
    text = resp.text[:200000] if html else ""
    low = text.lower()
    if "_cf_chl_opt" in text or "<title>just a moment" in low:
        return "cloudflare_challenge"                 # header stripped by a proxy
    if expect and expect in text:
        return "ok"
    if WIDGET.search(text):
        return "captcha"                              # even on a 429 or a 403
    if code == 429 or (code == 503 and "retry-after" in h):
        return "rate_limited"
    if code == 403 and html and "csrf" not in low:
        return "blocked"                              # JSON 403s fall through
    m = TITLE.search(text)
    head = m.group(1).lower() if m else ""
    small = len(text) < 3000                          # block pages are short
    if code < 400 and html and any(b in head or (small and b in low) for b in BLOCK):
        return "blocked"                              # soft block: 200 with a notice
    return "ok" if code < 400 else "unknown"

import re
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.parse import urljoin

from classify import classify


def retry_delay(resp, fallback):
    raw = resp.headers.get("Retry-After")
    if not raw:
        return fallback
    try:
        return max(1.0, float(raw))                        # delay-seconds
    except ValueError:
        pass
    try:
        when = parsedate_to_datetime(raw)                  # HTTP-date
        return max(1.0, (when - datetime.now(timezone.utc)).total_seconds())
    except (TypeError, ValueError):
        return fallback

def solve_recaptcha_form(session, resp, solver):
    """reCAPTCHA v2 only. For other types, run the detector first."""
    key = re.search(r"data-sitekey=[\"']([^\"']+)", resp.text)
    form = re.search(r"<form[^>]*action=[\"']([^\"']*)", resp.text)
    if "g-recaptcha" not in resp.text or not (key and form):
        raise RuntimeError("not a plain reCAPTCHA v2 form: run the detector first")
    token = solver.recaptcha(sitekey=key.group(1), url=resp.url)["code"]
    return session.post(urljoin(resp.url, form.group(1)),
                        data={"g-recaptcha-response": token}, timeout=30)

def fetch(session, url, solver, rotate, on_challenge,
          tries=4, sleep=time.sleep, expect=None):
    limited = 0
    for _ in range(tries):
        resp = session.get(url, timeout=30)
        kind = classify(resp, expect)
        if kind == "ok":
            return resp
        if kind == "rate_limited":
            limited += 1
            if limited > 2 and "Retry-After" not in resp.headers:
                rotate(session)                    # a bare 429 that outlasts two waits
                limited = 0                        # the new address starts fresh
            else:
                sleep(min(retry_delay(resp, 5.0), 60.0))
        elif kind == "captcha":
            resp = solve_recaptcha_form(session, resp, solver)
            if classify(resp, expect) == "ok":
                return resp
        elif kind == "cloudflare_challenge":
            on_challenge(session, resp)            # yours: a browser, or the Turnstile guide
        elif kind == "blocked":
            rotate(session)                        # never solve a block
        else:
            raise RuntimeError(f"unknown response {resp.status_code} from {url}")
    raise RuntimeError(f"gave up on {url} after {tries} tries")

import requests
from capskip import CapSkip
from policy import fetch

solver = CapSkip(host="127.0.0.1", port=8080)


def rotate(session):
    raise RuntimeError("blocked: change the proxy here, or stop the run")


def on_challenge(session, resp):
    raise RuntimeError("Cloudflare challenge page: needs a browser")


session = requests.Session()
page = fetch(session, "https://example.com/data", solver, rotate, on_challenge)
print(page.status_code, len(page.text))

import requests
from classify import classify

URLS = [
    "https://example.com/",
    "https://httpbin.org/status/429",
    "https://httpbin.org/status/403",
]

for url in URLS:
    resp = requests.get(url, timeout=30)
    print(classify(resp), resp.status_code, url)