应用场景

使用验证码处理进行事件票证监控

票务平台使用验证码来防止自动检查和购买。在监控活动门票可用性(音乐会、体育赛事、剧院)时,您的抓取工具将遇到 reCAPTCHA 挑战、Cloudflare 保护和速率限制。 CaptchaAI 处理验证码解决方案,以便您的显示器可以可靠地检查可用性。

本指南构建了一个具有集成验证码解决功能的工单监控工作流程。


监控工作流程

Configure events → Check availability → CAPTCHA?
                                           ↓ Yes
                                      Solve via CaptchaAI → Retry
                                           ↓ No
                                      Parse availability → Changed?
                                                             ↓ Yes
                                                         Send alert

你需要什么

要求 细节
CaptchaAI API 密钥 验证码网站
Python 3.8+ 使用requests
代理人 推荐自有服务器基础设施
pip install requests

验证码求解器助手

import requests
import time

API_KEY = "YOUR_API_KEY"


def solve_captcha(method, params):
    """Generic CaptchaAI solver for any supported method."""
    params["key"] = API_KEY
    params["json"] = 1

    submit = requests.post("https://ocr.captchaai.com/in.php", data=params).json()

    if submit.get("status") != 1:
        raise RuntimeError(f"Submit error: {submit.get('request')}")

    task_id = submit["request"]
    initial_wait = 10 if method == "turnstile" else 20
    time.sleep(initial_wait)

    for _ in range(30):
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id, "json": 1
        }).json()
        if result.get("status") == 1:
            return result["request"]
        if result.get("request") != "CAPCHA_NOT_READY":
            raise RuntimeError(f"Solve error: {result['request']}")
        time.sleep(5)
    raise TimeoutError("Solve timed out")

票务监控器

from datetime import datetime
import json


class TicketMonitor:
    def __init__(self, proxy=None):
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })
        if proxy:
            self.session.proxies = {
                "http": f"http://{proxy}",
                "https": f"http://{proxy}"
            }
        self.last_status = {}

    def check_event(self, event):
        """Check ticket availability for an event, solving CAPTCHAs if needed."""
        url = event["url"]
        response = self.session.get(url)

        # Handle CAPTCHA if detected
        if "g-recaptcha" in response.text or "recaptcha" in response.text:
            sitekey = self._extract_sitekey(response.text)
            if sitekey:
                token = solve_captcha("userrecaptcha", {
                    "method": "userrecaptcha",
                    "googlekey": sitekey,
                    "pageurl": url
                })
                response = self.session.post(url, data={
                    "g-recaptcha-response": token
                })

        elif "cf-turnstile" in response.text:
            sitekey = self._extract_turnstile_key(response.text)
            if sitekey:
                token = solve_captcha("turnstile", {
                    "method": "turnstile",
                    "sitekey": sitekey,
                    "pageurl": url
                })
                response = self.session.post(url, data={
                    "cf-turnstile-response": token
                })

        # Parse availability
        availability = self._parse_availability(response.text, event)

        # Check for changes
        event_key = event["name"]
        if event_key in self.last_status:
            if availability != self.last_status[event_key]:
                self._send_alert(event, availability)

        self.last_status[event_key] = availability
        return availability

    def _extract_sitekey(self, html):
        if 'data-sitekey="' in html:
            start = html.index('data-sitekey="') + 14
            end = html.index('"', start)
            return html[start:end]
        return None

    def _extract_turnstile_key(self, html):
        if 'data-sitekey="' in html:
            start = html.index('data-sitekey="') + 14
            end = html.index('"', start)
            return html[start:end]
        return None

    def _parse_availability(self, html, event):
        """Parse ticket availability. Customize per ticketing site."""
        available = "sold out" not in html.lower()
        return {
            "event": event["name"],
            "available": available,
            "checked_at": datetime.now().isoformat()
        }

    def _send_alert(self, event, availability):
        """Send availability change notification."""
        status = "AVAILABLE" if availability["available"] else "SOLD OUT"
        print(f"[ALERT] {event['name']}: {status}")

    def monitor_all(self, events):
        """Check all events and return results."""
        results = []
        for event in events:
            try:
                result = self.check_event(event)
                results.append(result)
                print(f"[OK] {event['name']}: {'available' if result['available'] else 'sold out'}")
            except Exception as e:
                print(f"[ERROR] {event['name']}: {e}")
        return results


# Usage
events = [
    {
        "name": "Concert - Madison Square Garden - Aug 15",
        "url": "https://example-tickets.com/event/12345"
    },
    {
        "name": "Basketball Finals - Game 7",
        "url": "https://example-tickets.com/event/67890"
    }
]

monitor = TicketMonitor(proxy="user:pass@proxy.example.com:8080")
results = monitor.monitor_all(events)

for r in results:
    print(json.dumps(r, indent=2))

预期输出:

[OK] Concert - Madison Square Garden - Aug 15: available
[OK] Basketball Finals - Game 7: sold out

调度

定期运行检查:

# Check every 15 minutes
*/15 * * * * cd /path/to/project && python ticket_monitor.py >> /var/log/tickets.log 2>&1

故障排除

问题 原因 处理方式
每张支票上都有频繁的验证码 相同IP,无会话保持 使用 cookie、轮换自有服务器基础设施
经过几次检查后被阻止 速率限制 增加检查间隔,使用代理轮换
可用性状态错误 页面结构已更改 更新_parse_availability方法
验证码解决速度慢 高求解器负载 通过退避实现重试逻辑

常问问题

我应该多久检查一次门票供应情况?

一般监控每 10 - 30 分钟一次。对于高需求事件,每 2 - 5 分钟一次 - 但预计会以更高的频率出现更多验证码。

票务网站使用哪些验证码?

最常见的是 reCAPTCHA v2、Cloudflare Turnstile 和 Cloudflare 验证流程 页面。虚拟候诊室等队列系统可能会使用自定义挑战。

我可以监控多个票务平台吗?

是的。为每个平台的 HTML 结构自定义解析器并添加来自不同站点的事件。

我需要自有服务器基础设施吗?

是的。票务站点积极阻止数据中心 IP。自有服务器基础设施会降低验证码频率。

我如何处理候诊室/queues?

等候室与验证码是分开的。您的监视器应该检测队列页面并等待或重试。 CaptchaAI 解决队列后面出现的验证码。


获取您的 CaptchaAI API 密钥

开始监控门票可用性验证码网站。在事件监控工作流程中自动处理验证码。


相关指南

该文章已禁用评论。