-
Notifications
You must be signed in to change notification settings - Fork 37
/
NoCaptchaTask.py
156 lines (136 loc) · 8.28 KB
/
NoCaptchaTask.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
import requests
import time
import asyncio
import aiohttp
from .config import create_task_url, get_result_url, app_key, user_agent_data
class NoCaptchaTask:
def __init__(self, anticaptcha_key, proxyAddress, proxyPort, sleep_time=5, proxyType = 'http', **kwargs):
"""
Модуль отвечает за решение NoCaptcha.
userAgent рандомно берётся из актульного списка браузеров-параметров
:param anticaptcha_key: ключ от АнтиКапчи
:param proxyAdress: Адрес прокси-сервера
:param proxyPort: Порт сервера
:param proxyType: Тип прокси http/socks5/socks4
:param sleeptime: Время ожидания решения
:param kwargs: Необязательные параметры, можно переопределить userAgent
"""
if sleep_time < 5:
raise ValueError(f'Параметр `sleep_time` должен быть не менее 5. Вы передали - {sleep_time}')
self.sleep_time = sleep_time
# Пайлоад для создания задачи
self.task_payload = {"clientKey": anticaptcha_key,
"task":
{
"type": "NoCaptchaTask",
"userAgent": user_agent_data,
"proxyType": proxyType,
"proxyAddress": proxyAddress,
"proxyPort": proxyPort,
},
"softId": app_key
}
# пайлоад для получения ответа сервиса
self.result_payload = {"clientKey": anticaptcha_key}
# заполнить пайлоад остальными аргументами
if kwargs:
for key in kwargs:
self.task_payload['task'].update({key: kwargs[key]})
# Работа с капчей
def captcha_handler(self, websiteURL, websiteKey, **kwargs):
"""
Метод получает ссылку на страницу, где расположена капча, и ключ капчи
:param: websiteURL: Ссылка на страницу с капчёй
:param: websiteKey: Ключ капчи(как его получить - описано в документаии на сайте антикапчи)
return: Возвращает ответ сервера в виде JSON(ответ так же можно глянуть в документации антикапчи)
"""
self.task_payload['task'].update({"websiteURL": websiteURL,
"websiteKey": websiteKey})
# отправляем реквест, в ответ получаем JSON содержащий номер решаемой капчи
captcha_id = requests.post(create_task_url, json=self.task_payload, **kwargs).json()
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id['errorId'] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# Ждем решения капчи
time.sleep(self.sleep_time)
while True:
captcha_response = requests.post(get_result_url, json=self.result_payload)
if captcha_response.json()["errorId"] == 0:
if captcha_response.json()["status"] == "processing":
time.sleep(self.sleep_time)
else:
return captcha_response.json()
else:
return captcha_response.json()
class aioNoCaptchaTask:
def __init__(self, anticaptcha_key, proxyAddress, proxyPort, sleep_time=5, proxyType = 'http', **kwargs):
"""
Модуль отвечает за решение NoCaptcha.
userAgent рандомно берётся из актульного списка браузеров-параметров
:param anticaptcha_key: ключ от АнтиКапчи
:param proxyAdress: Адрес прокси-сервера
:param proxyPort: Порт сервера
:param proxyType: Тип прокси http/socks5/socks4
:param sleeptime: Время ожидания решения
:param kwargs: Необязательные параметры, можно переопределить userAgent
"""
self.sleep_time = sleep_time
# Пайлоад для создания задачи
self.task_payload = {"clientKey": anticaptcha_key,
"task":
{
"type": "FunCaptchaTask",
"userAgent": user_agent_data,
"proxyType": proxyType,
"proxyAddress": proxyAddress,
"proxyPort": proxyPort,
},
"softId": app_key
}
# пайлоад для получения ответа сервиса
self.result_payload = {"clientKey": anticaptcha_key}
# заполнить пайлоад остальными аргументами
if kwargs:
for key in kwargs:
self.task_payload['task'].update({key: kwargs[key]})
# Работа с капчей
async def captcha_handler(self, websiteURL, websiteKey):
"""
Метод получает ссылку на страницу, где расположена капча, и ключ капчи
:param: websiteURL: Ссылка на страницу с капчёй
:param: websiteKey: Ключ капчи(как его получить - описано в документаии на сайте антикапчи)
return: Возвращает ответ сервера в виде JSON(ответ так же можно глянуть в документации антикапчи)
"""
self.task_payload["task"].update({"websiteURL": websiteURL,
"websiteKey": websiteKey})
# отправляем реквест, в ответ получаем JSON содержащий номер решаемой капчи
captcha_id = requests.post(create_task_url, json=self.task_payload).json()
async with aiohttp.ClientSession() as session:
async with session.post(create_task_url, json=self.task_payload) as resp:
captcha_id = await resp.json()
# Проверка статуса создания задачи, если создано без ошибок - извлекаем ID задачи, иначе возвращаем ответ сервера
if captcha_id["errorId"] == 0:
captcha_id = captcha_id["taskId"]
self.result_payload.update({"taskId": captcha_id})
else:
return captcha_id
# Ждем решения капчи
await asyncio.sleep(self.sleep_time)
# Отправляем запрос на статус решения капчи.
async with aiohttp.ClientSession as session:
while True:
async with session.post(get_result_url, json=self.result_payload) as resp:
json_result = await resp.json()
# Если нет ошибки - проверяем статус капчи
if json_result["errorId"] == 0:
# Если еще не решена, ожидаем
if json_result["status"] == "processing":
await asyncio.sleep(self.sleep_time)
# Иначе возвращаем ответ
else:
return json_result
else:
return json_result