票务平台使用验证码来防止自动检查和购买。在监控活动门票可用性(音乐会、体育赛事、剧院)时,您的抓取工具将遇到 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 密钥
开始监控门票可用性验证码网站。在事件监控工作流程中自动处理验证码。
相关指南
- 如何使用API解决reCAPTCHA v2
- 如何使用API解决Cloudflare Turnstile
- 通过验证码解决实现ticketing QA 测试
- 使用 CaptchaAI 实现重试逻辑