127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
import requests
|
||
import time
|
||
import json
|
||
from typing import Optional, Dict, Any
|
||
|
||
class TwoCaptchaSolver:
|
||
"""2captcha 腾讯滑块验证码处理器"""
|
||
|
||
def __init__(self, api_key: str):
|
||
self.api_key = api_key
|
||
self.base_url = "http://2captcha.com"
|
||
|
||
def solve_tencent_captcha(self, app_id: str, page_url: str, extra_params: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
解决腾讯滑块验证码
|
||
|
||
Args:
|
||
app_id: 腾讯验证码的 app_id
|
||
page_url: 验证码所在页面的 URL
|
||
extra_params: 额外的参数（如果需要）
|
||
|
||
Returns:
|
||
包含验证结果的字典，或 None（如果失败）
|
||
"""
|
||
# 提交验证码任务
|
||
submit_url = f"{self.base_url}/in.php"
|
||
submit_params = {
|
||
"key": self.api_key,
|
||
"method": "tencent",
|
||
"app_id": app_id,
|
||
"pageurl": page_url,
|
||
"json": 1
|
||
}
|
||
|
||
if extra_params:
|
||
submit_params.update(extra_params)
|
||
|
||
try:
|
||
print(f"[2captcha] 提交腾讯验证码任务...")
|
||
response = requests.post(submit_url, data=submit_params)
|
||
result = response.json()
|
||
|
||
if result.get("status") != 1:
|
||
print(f"[2captcha] 提交失败: {result.get('error_text', 'Unknown error')}")
|
||
return None
|
||
|
||
task_id = result.get("request")
|
||
print(f"[2captcha] 任务ID: {task_id}")
|
||
|
||
# 等待并获取结果
|
||
return self._get_result(task_id)
|
||
|
||
except Exception as e:
|
||
print(f"[2captcha] 错误: {str(e)}")
|
||
return None
|
||
|
||
def _get_result(self, task_id: str, max_attempts: int = 60, wait_time: int = 5) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取验证码识别结果
|
||
|
||
Args:
|
||
task_id: 2captcha 任务ID
|
||
max_attempts: 最大尝试次数
|
||
wait_time: 每次尝试之间的等待时间（秒）
|
||
|
||
Returns:
|
||
验证结果字典
|
||
"""
|
||
result_url = f"{self.base_url}/res.php"
|
||
|
||
for attempt in range(max_attempts):
|
||
params = {
|
||
"key": self.api_key,
|
||
"action": "get",
|
||
"id": task_id,
|
||
"json": 1
|
||
}
|
||
|
||
try:
|
||
response = requests.get(result_url, params=params)
|
||
result = response.json()
|
||
|
||
if result.get("status") == 1:
|
||
print("[2captcha] 验证码识别成功")
|
||
# 返回腾讯验证码需要的参数
|
||
return {
|
||
"ticket": result.get("request", {}).get("ticket"),
|
||
"randstr": result.get("request", {}).get("randstr"),
|
||
"raw_response": result
|
||
}
|
||
elif result.get("request") == "CAPCHA_NOT_READY":
|
||
print(f"[2captcha] 等待识别中... ({attempt + 1}/{max_attempts})")
|
||
time.sleep(wait_time)
|
||
else:
|
||
print(f"[2captcha] 获取结果失败: {result.get('error_text', 'Unknown error')}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
print(f"[2captcha] 获取结果时出错: {str(e)}")
|
||
time.sleep(wait_time)
|
||
|
||
print("[2captcha] 超时，未能获取验证码结果")
|
||
return None
|
||
|
||
def report_good(self, task_id: str):
|
||
"""报告验证码正确"""
|
||
try:
|
||
requests.get(f"{self.base_url}/res.php", params={
|
||
"key": self.api_key,
|
||
"action": "reportgood",
|
||
"id": task_id
|
||
})
|
||
print("[2captcha] 已报告验证码正确")
|
||
except:
|
||
pass
|
||
|
||
def report_bad(self, task_id: str):
|
||
"""报告验证码错误"""
|
||
try:
|
||
requests.get(f"{self.base_url}/res.php", params={
|
||
"key": self.api_key,
|
||
"action": "reportbad",
|
||
"id": task_id
|
||
})
|
||
print("[2captcha] 已报告验证码错误")
|
||
except:
|
||
pass |