-
Notifications
You must be signed in to change notification settings - Fork 16
/
main.py
362 lines (309 loc) · 12.8 KB
/
main.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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
import asyncio
from api.qinglong import QlApi, QlOpenApi
from api.send import SendApi
from config import (
jd_login_url,
auto_move,
qinglong_data,
user_datas,
auto_shape_recognition,
)
import json
from loguru import logger
import os
from playwright.async_api import Playwright, async_playwright
import random
import traceback
from typing import Union
from utils.consts import (
supported_types,
supported_colors
)
from utils.tools import (
get_tmp_dir,
get_img_bytes,
get_forbidden_users_dict,
filter_forbidden_users,
save_img,
get_ocr,
get_word,
get_shape_location_by_type,
get_shape_location_by_color,
rgba2rgb,
send_msg,
solve_slider_captcha,
ddddocr_find_bytes_pic
)
"""
基于playwright做的
"""
logger.add(
sink="main.log",
level="DEBUG"
)
async def auto_move_slide(page, retry_times: int = 2):
"""
自动识别移动滑块验证码
"""
from config import slide_difference
for i in range(retry_times):
logger.info(f'第{i + 1}次尝试自动移动滑块中...')
try:
# 查找小图
await page.wait_for_selector('#small_img', state='visible', timeout=3000)
except Exception as e:
# 未找到元素,认为成功,退出循环
logger.info('未找到小图,退出移动滑块')
break
# 获取 src 属性
small_src = await page.locator('#small_img').get_attribute('src')
background_src = await page.locator('#cpc_img').get_attribute('src')
# 获取 bytes
small_img_bytes = get_img_bytes(small_src)
background_img_bytes = get_img_bytes(background_src)
# 获取滑块
slider = await page.wait_for_selector('img.move-img')
await asyncio.sleep(1)
# 获取要移动的长度
distance = ddddocr_find_bytes_pic(small_img_bytes, background_img_bytes)
await asyncio.sleep(1)
# 移动滑块
await solve_slider_captcha(page, slider, distance, slide_difference)
await asyncio.sleep(1)
async def auto_shape(page, retry_times: int = 5):
ocr = get_ocr(beta=True)
"""
自动识别滑块验证码
"""
for i in range(retry_times):
logger.info(f'第{i + 1}次自动识别形状中...')
try:
# 查找小图
await page.wait_for_selector('div.captcha_footer img', state='visible', timeout=3000)
except Exception as e:
# 未找到元素,认为成功,退出循环
logger.info('未找到形状图,退出识别状态')
break
tmp_dir = get_tmp_dir()
background_img_path = os.path.join(tmp_dir, f'background_img.png')
# 获取大图元素
background_locator = page.locator('#cpc_img')
# 获取元素的位置和尺寸
backend_bounding_box = await background_locator.bounding_box()
backend_top_left_x = backend_bounding_box['x']
backend_top_left_y = backend_bounding_box['y']
# 截取元素区域
await page.screenshot(path=background_img_path, clip=backend_bounding_box)
# 获取 图片的src 属性和button按键
word_img_src = await page.locator('div.captcha_footer img').get_attribute('src')
button = page.locator('div.captcha_footer button.sure_btn')
# 找到刷新按钮
refresh_button = page.locator('div.captcha_header img.jcap_refresh')
# 获取文字图并保存
word_img_bytes = get_img_bytes(word_img_src)
rgba_word_img_path = save_img('rgba_word_img', word_img_bytes)
# 文字图是RGBA的,有蒙板识别不了,需要转成RGB
rgb_word_img_path = rgba2rgb('rgb_word_img', rgba_word_img_path)
# 获取问题的文字
word = get_word(ocr, rgb_word_img_path)
if word.find('色') > 0:
target_color = word.split('请选出图中')[1].split('的图形')[0]
if target_color in supported_colors:
logger.info(f'正在点击中......')
# 获取点的中心点
center_x, center_y = get_shape_location_by_color(background_img_path, target_color)
if center_x is None and center_y is None:
logger.info(f'识别失败,刷新中......')
await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
# 得到网页上的中心点
x, y = backend_top_left_x + center_x, backend_top_left_y + center_y
# 点击图片
await page.mouse.click(x, y)
await asyncio.sleep(random.uniform(1, 4))
# 点击确定
await button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
else:
logger.info(f'不支持{target_color},刷新中......')
# 刷新
await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
else:
shape_type = word.split('请选出图中的')[1]
if shape_type in supported_types:
logger.info(f'已找到图形,点击中......')
if shape_type == "圆环":
shape_type = shape_type.replace('圆环', '圆形')
# 获取点的中心点
center_x, center_y = get_shape_location_by_type(background_img_path, shape_type)
if center_x is None and center_y is None:
logger.info(f'识别失败,刷新中......')
await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
# 得到网页上的中心点
x, y = backend_top_left_x + center_x, backend_top_left_y + center_y
# 点击图片
await page.mouse.click(x, y)
await asyncio.sleep(random.uniform(1, 4))
# 点击确定
await button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
else:
logger.info(f'不支持{shape_type},刷新中......')
# 刷新
await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4))
continue
async def get_jd_pt_key(playwright: Playwright, user) -> Union[str, None]:
"""
获取jd的pt_key
"""
try:
from config import headless
except ImportError:
headless = False
browser = await playwright.chromium.launch(headless=headless)
context = await browser.new_context()
try:
page = await context.new_page()
# 关闭Webdriver属性,绕过Webdriver检测
js = """Object.defineProperties(navigator, {webdriver:{get:()=>undefined}});"""
await page.add_init_script(js)
await page.goto(jd_login_url)
await page.get_by_text("账号密码登录").click()
username_input = page.get_by_placeholder("账号名/邮箱/手机号")
await username_input.click()
for u in user:
await username_input.type(u, no_wait_after=True)
await asyncio.sleep(random.random() / 10)
password_input = page.get_by_placeholder("请输入密码")
await password_input.click()
password = user_datas[user]["password"]
for p in password:
await password_input.type(p, no_wait_after=True)
await asyncio.sleep(random.random() / 10)
await page.get_by_role("checkbox").check()
await page.get_by_text("登 录").click()
# 自动识别移动滑块验证码
if auto_move:
# 关键的sleep
await asyncio.sleep(1)
await auto_move_slide(page, retry_times=5)
# 自动验证形状验证码
if auto_shape_recognition:
await asyncio.sleep(1)
await auto_shape(page, retry_times=30)
# 等待验证码通过
logger.info("等待获取cookie...")
await page.wait_for_selector('#msShortcutMenu', state='visible', timeout=120000)
cookies = await context.cookies()
for cookie in cookies:
if cookie['name'] == 'pt_key':
pt_key = cookie["value"]
return pt_key
return None
except Exception as e:
traceback.print_exc()
return None
finally:
await context.close()
await browser.close()
async def get_ql_api(ql_data):
"""
封装了QL的登录
"""
logger.info("开始获取QL登录态......")
# 优化client_id和client_secret
client_id = ql_data.get('client_id')
client_secret = ql_data.get('client_secret')
if client_id and client_secret:
logger.info("使用client_id和client_secret登录......")
qlapi = QlOpenApi(ql_data["url"])
response = qlapi.login(client_id=client_id, client_secret=client_secret)
if response['code'] == 200:
logger.info("client_id和client_secret正常可用......")
return qlapi
else:
logger.info("client_id和client_secret异常......")
qlapi = QlApi(ql_data["url"])
# 其次用token
token = ql_data.get('token')
if token:
logger.info("已设置TOKEN,开始检测TOKEN状态......")
qlapi.login_by_token(token)
# 如果token失效,就用账号密码登录
response = await qlapi.get_envs()
if response['code'] == 401:
logger.info("Token已失效, 正使用账号密码获取QL登录态......")
response = qlapi.login_by_username(ql_data.get("username"), ql_data.get("password"))
if response.status_code != 200:
logger.error(f"账号密码登录失败. response: {response}")
raise Exception(f"账号密码登录失败. response: {response}")
else:
logger.info("Token正常可用......")
else:
# 最后用账号密码
logger.info("正使用账号密码获取QL登录态......")
response = qlapi.login_by_username(ql_data.get("username"), ql_data.get("password"))
if response.status_code != 200:
logger.error(f"账号密码登录失败. response: {response}")
raise Exception(f"账号密码登录失败.response: {response}")
return qlapi
async def main():
try:
qlapi = await get_ql_api(qinglong_data)
send_api = SendApi("ql")
# 拿到禁用的用户列表
response = await qlapi.get_envs()
if response['code'] == 200:
logger.info("获取环境变量成功")
else:
logger.error(f"获取环境变量失败, response: {response}")
raise Exception(f"获取环境变量失败, response: {response}")
user_info = response['data']
# 获取禁用用户
forbidden_users = [x for x in user_info if x['name'] == 'JD_COOKIE' and x['status'] == 1]
if not forbidden_users:
logger.info("所有COOKIE环境变量正常,无需更新")
return
# 获取需要的字段
filter_users_list = filter_forbidden_users(forbidden_users, ['id', 'value', 'remarks', 'name'])
# 生成字典
user_dict = get_forbidden_users_dict(filter_users_list, user_datas)
# 登录JD获取pt_key
async with async_playwright() as playwright:
for user in user_dict:
logger.info(f"开始更新{user}")
pt_key = await get_jd_pt_key(playwright, user)
if pt_key is None:
logger.error(f"获取pt_key失败")
await send_msg(send_api, send_type=1, msg=f"{user} 更新失败")
continue
req_data = user_dict[user]
req_data["value"] = f"pt_key={pt_key};pt_pin={user_datas[user]['pt_pin']};"
logger.info(f"更新内容为{req_data}")
data = json.dumps(req_data)
response = await qlapi.set_envs(data=data)
if response['code'] == 200:
logger.info(f"{user}更新成功")
else:
logger.error(f"{user}更新失败, response: {response}")
await send_msg(send_api, send_type=1, msg=f"{user} 更新失败")
continue
data = bytes(f"[{req_data['id']}]", 'utf-8')
response = await qlapi.envs_enable(data=data)
if response['code'] == 200:
logger.info(f"{user}启用成功")
await send_msg(send_api, send_type=0, msg=f"{user} 更新成功")
else:
logger.error(f"{user}启用失败, response: {response}")
except Exception as e:
traceback.print_exc()
if __name__ == '__main__':
asyncio.run(main())