chickliu/jd-login
1
0
Fork 0
mirror of https://github.com/zhao-zg/jd-login.git synced 2026-01-16 16:00:43 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Update login.py

Browse Source 
优化代码
This commit is contained in:
zhao-zg 2024-12-06 09:51:50 +08:00 committed by GitHub
parent ca6a0a2300
commit 0ac90fa27b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 232 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

342
login.py
View File

@ -17,10 +17,16 @@ import numpy as np
import base64 import base64
import io import io
import re import re
import logging
# 传参获得已初始化的ddddocr实例 # 传参获得已初始化的ddddocr实例
ocr = None ocr = None
ocrDet = None ocrDet = None
logger = logging.getLogger("login")
simple_format = "[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d] %(message)s"
logging.basicConfig(level=logging.INFO, format=simple_format, datefmt="%Y-%m-%d %H:%M:%S %z")
# 支持的形状类型 # 支持的形状类型
supported_types = [ supported_types = [
"三角形", "三角形",
@ -71,42 +77,43 @@ async def loginPhone(chromium_path, workList, uid, headless):
return await isWrongAccountOrPassword(page, verify=True) return await isWrongAccountOrPassword(page, verify=True)
return False return False
except Exception as e: except Exception as e:
print("isWrongAccountOrPassword " + str(e)) logger.info("isWrongAccountOrPassword " + str(e))
return False return False
# 判断验证码错误 # 判断验证码错误
async def isStillInSMSCodeSentPage(page): async def isStillInSMSCodeSentPage(page):
try: try:
button = await page.querySelector('.getMsg-btn.text-btn.timer.active') if not await page.querySelector('.getMsg-btn.text-btn.timer.active') and await page.querySelector('#authcode'):
if button: return True
return False return False
return await page.querySelector('#authcode')
except Exception as e: except Exception as e:
print("isStillInSMSCodeSentPage " + str(e)) logger.info("isStillInSMSCodeSentPage " + str(e))
return False return False
# 判断验证码超时 # 判断验证码超时
async def needResendSMSCode(page): async def needResendSMSCode(page):
try: try:
return await page.querySelector('.getMsg-btn.text-btn.timer.active'); if await page.querySelector('.getMsg-btn.text-btn.timer.active'):
return True
return False
except Exception as e: except Exception as e:
print("needResendSMSCode " + str(e)) logger.info("needResendSMSCode " + str(e))
return False return False
async def isSendSMSDirectly(page): async def isSendSMSDirectly(page):
try: try:
title = await page.title() title = await page.title()
if title in ['手机语音验证', '手机短信验证']: if title in ['手机语音验证', '手机短信验证']:
print('需要' + title) logger.info('需要' + title)
return True return True
return False return False
except Exception as e: except Exception as e:
print("isSendSMSDirectly " + str(e)) logger.info("isSendSMSDirectly " + str(e))
return False return False
usernum = workList[uid].account usernum = workList[uid].account
print(f"正在登录 {usernum} 的手机号") logger.info(f"正在登录 {usernum} 的手机号")
browser = await launch( browser = await launch(
{ {
@ -135,30 +142,30 @@ async def loginPhone(chromium_path, workList, uid, headless):
sms_sent = False sms_sent = False
while True: while True:
now_time = datetime.datetime.now() now_time = datetime.datetime.now()
print("循环检测中...") logger.info("循环检测中...")
if (now_time - start_time).total_seconds() > 70: if (now_time - start_time).total_seconds() > 70:
print("进入超时分支") logger.info("进入超时分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "登录超时" workList[uid].msg = "登录超时"
break break
elif await page.J("#searchWrapper"): elif await page.J("#searchWrapper"):
print("进入成功获取cookie分支") logger.info("进入成功获取cookie分支")
workList[uid].cookie = await getCookie(page) workList[uid].cookie = await getCookie(page)
workList[uid].status = "pass" workList[uid].status = "pass"
break break
elif await page.xpath('//*[@id="captcha_modal"]'): elif await page.xpath('//*[@id="captcha_modal"]'):
print("进入安全验证分支") logger.info("进入安全验证分支")
if await page.xpath('//*[@id="small_img"]'): if await page.xpath('//*[@id="small_img"]'):
print("进入过滑块分支") logger.info("进入过滑块分支")
workList[uid].status = "pending" workList[uid].status = "pending"
workList[uid].msg = "正在过滑块检测" workList[uid].msg = "正在过滑块检测"
await verification(page) await verification(page)
await page.waitFor(2000) await page.waitFor(2000)
elif await page.xpath('//*[@id="captcha_modal"]/div/div[3]/button'): elif await page.xpath('//*[@id="captcha_modal"]/div/div[3]/button'):
print("进入点形状、颜色验证分支") logger.info("进入点形状、颜色验证分支")
workList[uid].status = "pending" workList[uid].status = "pending"
workList[uid].msg = "正在过形状、颜色检测" workList[uid].msg = "正在过形状、颜色检测"
@ -167,14 +174,14 @@ async def loginPhone(chromium_path, workList, uid, headless):
await page.waitFor(2000) await page.waitFor(2000)
continue continue
elif await page.querySelector('.dialog'): elif await page.querySelector('.dialog'):
print("进入弹出对话框分支") logger.info("进入弹出对话框分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "账号异常，自行检查" workList[uid].msg = "账号异常，自行检查"
break break
if False == sms_sent: if False == sms_sent:
button = await page.querySelector('.getMsg-btn.text-btn.timer.active') button = await page.querySelector('.getMsg-btn.text-btn.timer.active')
if button is None: if button is None:
print("进入直接发短信分支") logger.info("进入直接发短信分支")
if not workList[uid].isAuto: if not workList[uid].isAuto:
workList[uid].status = "SMS" workList[uid].status = "SMS"
workList[uid].msg = "需要短信验证" workList[uid].msg = "需要短信验证"
@ -184,11 +191,11 @@ async def loginPhone(chromium_path, workList, uid, headless):
else: else:
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "自动续期时不能使用短信验证" workList[uid].msg = "自动续期时不能使用短信验证"
print("自动续期时不能使用短信验证") logger.info("自动续期时不能使用短信验证")
break break
else: else:
if await isStillInSMSCodeSentPage(page): if await isStillInSMSCodeSentPage(page):
print("进入验证码错误分支") logger.info("进入验证码错误分支")
IN_SMS_TIMES += 1 IN_SMS_TIMES += 1
if IN_SMS_TIMES % 3 == 0: if IN_SMS_TIMES % 3 == 0:
workList[uid].SMS_CODE = None workList[uid].SMS_CODE = None
@ -197,7 +204,7 @@ async def loginPhone(chromium_path, workList, uid, headless):
await typePhoneSMScode(page, workList, uid) await typePhoneSMScode(page, workList, uid)
elif await needResendSMSCode(page): elif await needResendSMSCode(page):
print("进入验证码超时分支") logger.info("进入验证码超时分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "验证码超时，请重新开始" workList[uid].msg = "验证码超时，请重新开始"
break break
@ -206,13 +213,13 @@ async def loginPhone(chromium_path, workList, uid, headless):
except Exception as e: except Exception as e:
workList[uid].msg = "服务器异常退出" workList[uid].msg = "服务器异常退出"
workList[uid].status = "error" workList[uid].status = "error"
print("异常退出") logger.info("异常退出")
print(e) logger.info(e)
await browser.close() await browser.close()
await deleteSessionDelay(workList, uid) await deleteSessionDelay(workList, uid)
raise e raise e
print("任务完成退出") logger.info("任务完成退出")
await browser.close() await browser.close()
return return
async def loginPassword(chromium_path, workList, uid, headless): async def loginPassword(chromium_path, workList, uid, headless):
@ -231,7 +238,7 @@ async def loginPassword(chromium_path, workList, uid, headless):
return await isWrongAccountOrPassword(page, verify=True) return await isWrongAccountOrPassword(page, verify=True)
return False return False
except Exception as e: except Exception as e:
print("isWrongAccountOrPassword " + str(e)) logger.info("isWrongAccountOrPassword " + str(e))
return False return False
# 判断验证码超时 # 判断验证码超时
@ -247,7 +254,7 @@ async def loginPassword(chromium_path, workList, uid, headless):
return True return True
return False return False
except Exception as e: except Exception as e:
print("isStillInSMSCodeSentPage " + str(e)) logger.info("isStillInSMSCodeSentPage " + str(e))
return False return False
# 判断验证码错误 # 判断验证码错误
@ -263,24 +270,24 @@ async def loginPassword(chromium_path, workList, uid, headless):
return True return True
return False return False
except Exception as e: except Exception as e:
print("needResendSMSCode " + str(e)) logger.info("needResendSMSCode " + str(e))
return False return False
async def isSendSMSDirectly(page): async def isSendSMSDirectly(page):
try: try:
title = await page.title() title = await page.title()
if title in ['手机语音验证', '手机短信验证']: if title in ['手机语音验证', '手机短信验证']:
print('需要' + title) logger.info('需要' + title)
return True return True
return False return False
except Exception as e: except Exception as e:
print("isSendSMSDirectly " + str(e)) logger.info("isSendSMSDirectly " + str(e))
return False return False
usernum = workList[uid].account usernum = workList[uid].account
passwd = workList[uid].password passwd = workList[uid].password
sms_sent = False sms_sent = False
print(f"正在登录 {usernum} 的账号") logger.info(f"正在登录 {usernum} 的账号")
browser = await launch( browser = await launch(
{ {
@ -309,59 +316,130 @@ async def loginPassword(chromium_path, workList, uid, headless):
while True: while True:
now_time = datetime.datetime.now() now_time = datetime.datetime.now()
print("循环检测中...") logger.info("循环检测中...")
if (now_time - start_time).total_seconds() > 70: if (now_time - start_time).total_seconds() > 70:
print("进入超时分支") logger.info("进入超时分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "登录超时" workList[uid].msg = "登录超时"
logger.info("超时了，正在保存当前页面信息......")
dateTime = datetime.datetime.now()
logger.info(f"页面截图保存到： {usernum}-screenshot-{dateTime}.png")
await page.screenshot({'path': f"{usernum}-screenshot-{dateTime}.png"})
logger.info(f"页面HTML保存到： {usernum}-html-{dateTime}.html")
content = await page.content()
with open(f"{usernum}-html-{dateTime}.html", 'w', encoding='utf-8') as f:
f.write(content)
break break
elif await page.J("#searchWrapper"): elif await page.J("#searchWrapper"):
print("进入成功获取cookie分支") logger.info("进入成功获取cookie分支")
workList[uid].cookie = await getCookie(page) workList[uid].cookie = await getCookie(page)
workList[uid].status = "pass" workList[uid].status = "pass"
break break
elif await isWrongAccountOrPassword(page): elif await isWrongAccountOrPassword(page):
print("进入账号密码不正确分支") logger.info("进入账号密码不正确分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "账号或密码不正确" workList[uid].msg = "账号或密码不正确"
break break
elif await page.xpath('//*[@id="captcha_modal"]'): elif await page.xpath('//*[@id="captcha_modal"]'):
print("进入安全验证分支") logger.info("进入安全验证分支")
if await page.xpath('//*[@id="small_img"]'): if await page.xpath('//*[@id="small_img"]'):
print("进入过滑块分支") logger.info("进入过滑块分支")
workList[uid].status = "pending" workList[uid].status = "pending"
workList[uid].msg = "正在过滑块检测" workList[uid].msg = "正在过滑块检测"
await verification(page) await verification(page)
await page.waitFor(3000) await page.waitFor(3000)
elif await page.xpath('//*[@id="captcha_modal"]/div/div[3]/button'): elif await page.xpath('//*[@id="captcha_modal"]/div/div[3]/button'):
print("进入点形状、颜色验证分支") logger.info("进入点形状、颜色验证分支")
workList[uid].status = "pending" workList[uid].status = "pending"
workList[uid].msg = "正在过形状、颜色检测" workList[uid].msg = "正在过形状、颜色检测"
if await verification_shape(page) == "notSupport": if await verification_shape(page) == "notSupport":
print("即将重启浏览器重试") logger.info("即将重启浏览器重试")
await browser.close() await browser.close()
return "notSupport" return "notSupport"
await page.waitFor(3000) await page.waitFor(3000)
elif await page.querySelector('.dialog'):
print("进入弹出对话框分支")
workList[uid].status = "error"
workList[uid].msg = "账号异常，自行检查"
break
if not sms_sent:
if await page.J(".sub-title"): elif await page.J('.drag-content'):
print("进入选择短信验证分支") logger.info("进入旋转图片分支")
logger.info("正在保存当前页面信息......")
dateTime = datetime.datetime.now()
logger.info(f"页面截图保存到： drag_{usernum}-screenshot-{dateTime}.png")
await page.screenshot({'path': f"drag_{usernum}-screenshot-{dateTime}.png"})
logger.info("即将重启浏览器重试")
await browser.close()
return "notSupport"
elif await page.J('.alert-body #alertMsg'):
logger.info("进入弹框分支")
element = await page.J('.alert-body #alertMsg')
if element:
alertMsg = await page.evaluate(
"(element) => element.textContent", element
)
logger.info(f"弹框内容为: {alertMsg}")
if alertMsg == "发送短信验证码过于频繁，请稍后再试":
workList[uid].status = "error"
workList[uid].msg = "发送短信验证码过于频繁，请稍后再试"
break
elif alertMsg == "身份证号输入错误,若包含字母X,请输入大写字母":
logger.info("进入身份证号错误分支")
workList[uid].ID_CARD = None
workList[uid].status = "wrongIDCard"
workList[uid].msg = "身份证号输入错误,若包含字母X,请输入大写字母"
await page.click('.alert-sure')
await page.waitFor(random.randint(100, 2000))
input_elements = await page.JJ('.input-container.id-wrap > div')
await input_elements[5].click()
for i in range(6):
await page.keyboard.press('Backspace')
await page.waitFor(random.randint(100, 2000))
await page.waitFor(3000)
await typeIDCard(page, workList, uid)
if workList[uid].status == "error":
logger.info("输入身份证超时")
break
elif alertMsg == "您已超过当日请求上限，请明天再试":
workList[uid].status = "error"
workList[uid].msg = "您已超过当日请求上限，请明天再试"
break
elif alertMsg == "验证码错误多次，请重新获取":
workList[uid].status = "error"
workList[uid].msg = "验证码错误多次，请重新获取"
break
# 需要身份证验证
if await page.J(".sub-title") and await page.J('.icon-default.icon-userid'):
logger.info("进入身份证号验证分支")
if not workList[uid].isAuto:
await page.click(".icon-default.icon-userid")
workList[uid].status = "IDCard"
workList[uid].msg = "存在安全风险，请输入身份证号码的前两位和后四位验证身份"
await page.waitFor(3000)
await typeIDCard(page, workList, uid)
if workList[uid].status == "error":
logger.info("输入身份证超时")
break
else:
workList[uid].status = "error"
workList[uid].msg = "自动续期时不能使用身份证号验证"
logger.info("自动续期时不能使用身份证号验证")
break
if not sms_sent:
if await page.J(".sub-title") and not await page.J('.icon-default.icon-userid'):
logger.info("进入选择短信验证分支")
if not workList[uid].isAuto: if not workList[uid].isAuto:
workList[uid].status = "SMS" workList[uid].status = "SMS"
workList[uid].msg = "需要短信验证" workList[uid].msg = "需要短信验证"
if await sendSMS(page) == "notSupport": if await sendSMS(page) == "notSupport":
print("即将重启浏览器重试") logger.info("即将重启浏览器重试")
await browser.close() await browser.close()
return "notSupport" return "notSupport"
await page.waitFor(3000) await page.waitFor(3000)
@ -371,16 +449,16 @@ async def loginPassword(chromium_path, workList, uid, headless):
else: else:
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "自动续期时不能使用短信验证" workList[uid].msg = "自动续期时不能使用短信验证"
print("自动续期时不能使用短信验证") logger.info("自动续期时不能使用短信验证")
break break
elif await isSendSMSDirectly(page): elif await isSendSMSDirectly(page):
print("进入直接发短信分支") logger.info("进入直接发短信分支")
if not workList[uid].isAuto: if not workList[uid].isAuto:
workList[uid].status = "SMS" workList[uid].status = "SMS"
workList[uid].msg = "需要短信验证" workList[uid].msg = "需要短信验证"
if await sendSMSDirectly(page) == "notSupport": if await sendSMSDirectly(page) == "notSupport":
print("即将重启浏览器重试") logger.info("即将重启浏览器重试")
await browser.close() await browser.close()
return "notSupport" return "notSupport"
await page.waitFor(3000) await page.waitFor(3000)
@ -390,11 +468,11 @@ async def loginPassword(chromium_path, workList, uid, headless):
else: else:
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "自动续期时不能使用短信验证" workList[uid].msg = "自动续期时不能使用短信验证"
print("自动续期时不能使用短信验证") logger.info("自动续期时不能使用短信验证")
break break
else: else:
if await isStillInSMSCodeSentPage(page): if await isStillInSMSCodeSentPage(page):
print("进入验证码错误分支") logger.info("进入验证码错误分支")
IN_SMS_TIMES += 1 IN_SMS_TIMES += 1
if IN_SMS_TIMES % 3 == 0: if IN_SMS_TIMES % 3 == 0:
workList[uid].SMS_CODE = None workList[uid].SMS_CODE = None
@ -403,24 +481,69 @@ async def loginPassword(chromium_path, workList, uid, headless):
await typeSMScode(page, workList, uid) await typeSMScode(page, workList, uid)
elif await needResendSMSCode(page): elif await needResendSMSCode(page):
print("进入验证码超时分支") logger.info("进入验证码超时分支")
workList[uid].status = "error" workList[uid].status = "error"
workList[uid].msg = "验证码超时，请重新开始" workList[uid].msg = "验证码超时，请重新开始"
break break
await asyncio.sleep(1) await asyncio.sleep(1)
except Exception as e: except Exception as e:
print("异常退出") logger.info("异常退出")
print(e) logger.info(e)
await browser.close() await browser.close()
await deleteSessionDelay(workList, uid) await deleteSessionDelay(workList, uid)
raise e raise e
print("任务完成退出") logger.info("任务完成退出")
logger.info("开始删除缓存文件......")
if os.path.exists("image.png"):
os.remove("image.png")
if os.path.exists("template.png"):
os.remove("template.png")
if os.path.exists("shape_image.png"):
os.remove("shape_image.png")
if os.path.exists("rgba_word_img.png"):
os.remove("rgba_word_img.png")
if os.path.exists("rgb_word_img.png"):
os.remove("rgb_word_img.png")
logger.info("缓存文件已删除！")
logger.info("开始关闭浏览器....")
await browser.close() await browser.close()
logger.info("浏览器已关闭！")
return return
async def typeIDCard(page, workList, uid):
logger.info("开始输入身份证")
async def get_verification_IDCard(workList, uid):
logger.info("开始从全局变量获取身份证")
retry = 60
while not workList[uid].ID_CARD and not retry < 0:
await asyncio.sleep(1)
retry -= 1
if retry < 0:
workList[uid].status = "error"
workList[uid].msg = "输入身份证超时"
return
workList[uid].status = "pending"
return workList[uid].ID_CARD
ID_CARD = await get_verification_IDCard(workList, uid)
if not ID_CARD:
return
workList[uid].status = "pending"
workList[uid].msg = "正在通过身份证验证"
logger.info("正在输入身份证。。。。。")
input_elements = await page.JJ('.input-container.id-wrap > div')
await input_elements[0].click()
for ID in ID_CARD:
await page.keyboard.type(str(ID))
await page.waitFor(random.randint(100, 2000))
await page.click(".btn.J_ping")
await page.waitFor(3000)
async def typephoneuser(page, usernum): async def typephoneuser(page, usernum):
await page.waitFor(random.randint(200, 500)) await page.waitFor(random.randint(200, 500))
tel_input = await page.waitForSelector('input[type="tel"]') tel_input = await page.waitForSelector('input[type="tel"]')
@ -435,7 +558,7 @@ async def typephoneuser(page, usernum):
await page.click(".getMsg-btn.text-btn.timer") await page.click(".getMsg-btn.text-btn.timer")
await page.waitFor(random.randint(500, 1000)) await page.waitFor(random.randint(500, 1000))
async def typeuser(page, usernum, passwd): async def typeuser(page, usernum, passwd):
print("开始输入账号密码") logger.info("开始输入账号密码")
await page.waitForSelector(".J_ping.planBLogin") await page.waitForSelector(".J_ping.planBLogin")
await page.click(".J_ping.planBLogin") await page.click(".J_ping.planBLogin")
await page.type( await page.type(
@ -464,7 +587,7 @@ async def sendSMSDirectly(page):
await page.waitFor(3000) await page.waitFor(3000)
await preSendSMS(page) await preSendSMS(page)
print("开始发送验证码") logger.info("开始发送验证码")
try: try:
while True: while True:
@ -486,7 +609,7 @@ async def sendSMSDirectly(page):
async def sendSMS(page): async def sendSMS(page):
async def preSendSMS(page): async def preSendSMS(page):
print("进行发送验证码前置操作") logger.info("进行发送验证码前置操作")
await page.waitForXPath( await page.waitForXPath(
'//*[@id="app"]/div/div[2]/div[2]/span/a' '//*[@id="app"]/div/div[2]/div[2]/span/a'
) )
@ -506,7 +629,7 @@ async def sendSMS(page):
await page.waitFor(3000) await page.waitFor(3000)
await preSendSMS(page) await preSendSMS(page)
print("开始发送验证码") logger.info("开始发送验证码")
try: try:
while True: while True:
@ -526,10 +649,10 @@ async def sendSMS(page):
raise e raise e
async def typePhoneSMScode(page, workList, uid): async def typePhoneSMScode(page, workList, uid):
print("开始输入验证码") logger.info("开始输入验证码")
async def get_verification_code(workList, uid): async def get_verification_code(workList, uid):
print("开始从全局变量获取验证码") logger.info("开始从全局变量获取验证码")
retry = 60 retry = 60
while not workList[uid].SMS_CODE and not retry < 0: while not workList[uid].SMS_CODE and not retry < 0:
await asyncio.sleep(1) await asyncio.sleep(1)
@ -554,13 +677,11 @@ async def typePhoneSMScode(page, workList, uid):
await button.click() await button.click()
await page.waitFor(random.randint(2, 3) * 1000) await page.waitFor(random.randint(2, 3) * 1000)
async def typeSMScode(page, workList, uid): async def typeSMScode(page, workList, uid):
print("开始输入验证码") logger.info("开始输入验证码")
async def get_verification_code(workList, uid): async def get_verification_code(workList, uid):
print("开始从全局变量获取验证码") logger.info("开始从全局变量获取验证码")
retry = 60 retry = 60
while not workList[uid].SMS_CODE and not retry < 0: while not workList[uid].SMS_CODE and not retry < 0:
await asyncio.sleep(1) await asyncio.sleep(1)
@ -572,6 +693,7 @@ async def typeSMScode(page, workList, uid):
workList[uid].status = "pending" workList[uid].status = "pending"
return workList[uid].SMS_CODE return workList[uid].SMS_CODE
await page.waitForXPath('//*[@id="app"]/div/div[2]/div[2]/div/input') await page.waitForXPath('//*[@id="app"]/div/div[2]/div[2]/div/input')
code = await get_verification_code(workList, uid) code = await get_verification_code(workList, uid)
if not code: if not code:
@ -585,13 +707,13 @@ async def typeSMScode(page, workList, uid):
if input_elements: if input_elements:
input_value = await input_elements[0].getProperty("value") input_value = await input_elements[0].getProperty("value")
if input_value: if input_value:
print("清除验证码输入框中已有的验证码") logger.info("清除验证码输入框中已有的验证码")
await page.evaluate( await page.evaluate(
'(element) => element.value = ""', input_elements[0] '(element) => element.value = ""', input_elements[0]
) )
except Exception as e: except Exception as e:
print("typeSMScode" + str(e)) logger.info("typeSMScode" + str(e))
await input_elements[0].type(code) await input_elements[0].type(code)
await page.waitForXPath('//*[@id="app"]/div/div[2]/a[1]') await page.waitForXPath('//*[@id="app"]/div/div[2]/a[1]')
@ -602,7 +724,7 @@ async def typeSMScode(page, workList, uid):
async def verification(page): async def verification(page):
print("开始过滑块") logger.info("开始过滑块")
async def get_distance(): async def get_distance():
img = cv2.imread("image.png", 0) img = cv2.imread("image.png", 0)
@ -667,11 +789,11 @@ async def verification(page):
box["x"] + distance, box["y"], {"steps": 10} box["x"] + distance, box["y"], {"steps": 10}
) )
await page.mouse.up() await page.mouse.up()
print("过滑块结束") logger.info("过滑块结束")
async def verification_shape(page): async def verification_shape(page):
print("开始过颜色、形状验证") logger.info("开始过颜色、形状验证")
def get_shape_location_by_type(img_path, type: str): def get_shape_location_by_type(img_path, type: str):
def sort_rectangle_vertices(vertices): def sort_rectangle_vertices(vertices):
@ -810,7 +932,7 @@ async def verification_shape(page):
if button is None: if button is None:
button = await page.querySelector(".sure_btn") button = await page.querySelector(".sure_btn")
if button is None: if button is None:
print("未找到提交按钮") logger.info("未找到提交按钮")
raise "未找到提交按钮" raise "未找到提交按钮"
refresh_button = await page.querySelector("div.captcha_header img.jcap_refresh") refresh_button = await page.querySelector("div.captcha_header img.jcap_refresh")
if refresh_button is None: if refresh_button is None:
@ -822,14 +944,14 @@ async def verification_shape(page):
if word.find("色") > 0: if word.find("色") > 0:
target_color = word.split("请选出图中")[1].split("的图形")[0] target_color = word.split("请选出图中")[1].split("的图形")[0]
if target_color in supported_colors: if target_color in supported_colors:
print(f"正在找{target_color}") logger.info(f"正在找{target_color}")
center_x, center_y = get_shape_location_by_color( center_x, center_y = get_shape_location_by_color(
"shape_image.png", target_color "shape_image.png", target_color
) )
if center_x is None and center_y is None: if center_x is None and center_y is None:
print("识别失败，刷新") logger.info("识别失败，刷新")
if refresh_button is None: if refresh_button is None:
print("未找到刷新按钮") logger.info("未找到刷新按钮")
raise "未找到刷新按钮" raise "未找到刷新按钮"
await refresh_button.click() await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
@ -841,23 +963,23 @@ async def verification_shape(page):
await asyncio.sleep(random.uniform(0.3, 1)) await asyncio.sleep(random.uniform(0.3, 1))
break break
else: else:
print(f"不支持{target_color}，重试") logger.info(f"不支持{target_color}，重试")
if refresh_button is None: if refresh_button is None:
print("未找到刷新按钮") logger.info("未找到刷新按钮")
raise "未找到刷新按钮" raise "未找到刷新按钮"
await refresh_button.click() await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
break break
elif word.find("依次") > 0: elif word.find("依次") > 0:
if retry_count < 1: if retry_count < 1:
print("文字点选重试失败") logger.info("文字点选重试失败")
return "notSupport" return "notSupport"
i = 3 i = 3
print("进入文字点选") logger.info("进入文字点选")
print(f"文字点选第{11 - retry_count}次尝试") logger.info(f"文字点选第{11 - retry_count}次尝试")
retry_count -= 1 retry_count -= 1
target_word = word.replace("\"", "")[-4:] target_word = word.replace("\"", "")[-4:]
print(f"点选字为： {target_word}") logger.info(f"点选字为： {target_word}")
gray_img = get_gray_img("shape_image.png") gray_img = get_gray_img("shape_image.png")
xy_list = ocrDet.detection(gray_img) xy_list = ocrDet.detection(gray_img)
src_img = Image.open("shape_image.png") src_img = Image.open("shape_image.png")
@ -869,7 +991,7 @@ async def verification_shape(page):
result_word = ocr.classification(corp, png_fix=True) result_word = ocr.classification(corp, png_fix=True)
words.append(result_word) words.append(result_word)
result = dict(zip(words, xy_list)) result = dict(zip(words, xy_list))
print(f"result: {result}") logger.info(f"result: {result}")
img_xy = {} img_xy = {}
for key, xy in result.items(): for key, xy in result.items():
img_xy[key] = (int((xy[0] + xy[2]) / 2), int((xy[1] + xy[3]) / 2)) img_xy[key] = (int((xy[0] + xy[2]) / 2), int((xy[1] + xy[3]) / 2))
@ -877,9 +999,9 @@ async def verification_shape(page):
click_points = {} click_points = {}
for wd in target_word: for wd in target_word:
if wd not in img_xy: if wd not in img_xy:
print(f"\"{wd}\"未找到，识别失败,刷新") logger.info(f"\"{wd}\"未找到，识别失败,刷新")
if refresh_button is None: if refresh_button is None:
print("未找到刷新按钮") logger.info("未找到刷新按钮")
raise "未找到刷新按钮" raise "未找到刷新按钮"
await refresh_button.click() await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
@ -888,14 +1010,14 @@ async def verification_shape(page):
center_x, center_y = img_xy[wd] center_x, center_y = img_xy[wd]
click_x, click_y = image_top_left_x + center_x, image_top_left_y + center_y click_x, click_y = image_top_left_x + center_x, image_top_left_y + center_y
click_points[wd] = [click_x, click_y] click_points[wd] = [click_x, click_y]
print(click_points) logger.info(click_points)
if os.path.exists("gray.png"): if os.path.exists("gray.png"):
os.remove("gray.png") os.remove("gray.png")
if not_found: if not_found:
continue continue
print("文字点选识别正常") logger.info("文字点选识别正常")
for wd, point in click_points.items(): for wd, point in click_points.items():
print(f"点击\"{wd}\",坐标{point[0]}:{point[1]}") logger.info(f"点击\"{wd}\",坐标{point[0]}:{point[1]}")
await page.mouse.click(point[0], point[1]) await page.mouse.click(point[0], point[1])
await asyncio.sleep(random.uniform(0.5, 2)) await asyncio.sleep(random.uniform(0.5, 2))
await button.click() await button.click()
@ -904,16 +1026,16 @@ async def verification_shape(page):
else: else:
shape_type = word.split("请选出图中的")[1] shape_type = word.split("请选出图中的")[1]
if shape_type in supported_types: if shape_type in supported_types:
print(f"正在找{shape_type}") logger.info(f"正在找{shape_type}")
if shape_type == "圆环": if shape_type == "圆环":
shape_type = shape_type.replace("圆环", "圆形") shape_type = shape_type.replace("圆环", "圆形")
center_x, center_y = get_shape_location_by_type( center_x, center_y = get_shape_location_by_type(
"shape_image.png", shape_type "shape_image.png", shape_type
) )
if center_x is None and center_y is None: if center_x is None and center_y is None:
print(f"识别失败,刷新") logger.info(f"识别失败,刷新")
if refresh_button is None: if refresh_button is None:
print("未找到刷新按钮") logger.info("未找到刷新按钮")
raise "未找到刷新按钮" raise "未找到刷新按钮"
await refresh_button.click() await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
@ -925,14 +1047,14 @@ async def verification_shape(page):
await asyncio.sleep(random.uniform(0.3, 1)) await asyncio.sleep(random.uniform(0.3, 1))
break break
else: else:
print(f"不支持{shape_type},刷新中......") logger.info(f"不支持{shape_type},刷新中......")
if refresh_button is None: if refresh_button is None:
print("未找到刷新按钮") logger.info("未找到刷新按钮")
raise "未找到刷新按钮" raise "未找到刷新按钮"
await refresh_button.click() await refresh_button.click()
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
continue continue
print("过图形结束") logger.info("过图形结束")
async def getCookie(page): async def getCookie(page):
@ -945,7 +1067,7 @@ async def getCookie(page):
elif cookie["name"] == "pt_pin": elif cookie["name"] == "pt_pin":
pt_pin = cookie["value"] pt_pin = cookie["value"]
ck = f"pt_key={pt_key};pt_pin={pt_pin};" ck = f"pt_key={pt_key};pt_pin={pt_pin};"
print(f"登录成功 {ck}") logger.info(f"登录成功 {ck}")
return ck return ck
@ -964,8 +1086,8 @@ async def download_file(url, file_path):
file.write(chunk) file.write(chunk)
downloaded_size += len(chunk) downloaded_size += len(chunk)
progress = (downloaded_size / file_size) * 100 progress = (downloaded_size / file_size) * 100
print(f"已下载{progress:.2f}%...", end="\r") logger.info(f"已下载{progress:.2f}%...", end="\r")
print("下载完成，进行解压安装....") logger.info("下载完成，进行解压安装....")
async def main(workList, uid, oocr, oocrDet): async def main(workList, uid, oocr, oocrDet):
@ -1000,7 +1122,7 @@ async def main(workList, uid, oocr, oocrDet):
if os.path.exists(chrome_exe): if os.path.exists(chrome_exe):
return chrome_exe return chrome_exe
else: else:
print("貌似第一次使用，未找到chrome，正在下载chrome浏览器....") logger.info("貌似第一次使用，未找到chrome，正在下载chrome浏览器....")
chromeurl = "https://mirrors.huaweicloud.com/chromium-browser-snapshots/Win_x64/588429/chrome-win32.zip" chromeurl = "https://mirrors.huaweicloud.com/chromium-browser-snapshots/Win_x64/588429/chrome-win32.zip"
target_file = "chrome-win.zip" target_file = "chrome-win.zip"
@ -1012,7 +1134,7 @@ async def main(workList, uid, oocr, oocrDet):
source_item = os.path.join(chmod_dir, item) source_item = os.path.join(chmod_dir, item)
destination_item = os.path.join(chrome_dir, item) destination_item = os.path.join(chrome_dir, item)
os.rename(source_item, destination_item) os.rename(source_item, destination_item)
print("解压安装完成") logger.info("解压安装完成")
await asyncio.sleep(1) await asyncio.sleep(1)
return chrome_exe return chrome_exe
@ -1026,11 +1148,11 @@ async def main(workList, uid, oocr, oocrDet):
if os.path.isfile(chrome_path): if os.path.isfile(chrome_path):
return chrome_path return chrome_path
else: else:
print("貌似第一次使用，未找到chrome，正在下载chrome浏览器....") logger.info("貌似第一次使用，未找到chrome，正在下载chrome浏览器....")
print("文件位于github，请耐心等待，如遇到网络问题可到项目地址手动下载") logger.info("文件位于github，请耐心等待，如遇到网络问题可到项目地址手动下载")
download_url = "https://mirrors.huaweicloud.com/chromium-browser-snapshots/Linux_x64/884014/chrome-linux.zip" download_url = "https://mirrors.huaweicloud.com/chromium-browser-snapshots/Linux_x64/884014/chrome-linux.zip"
if 'arm' in platform.machine(): if 'arm' in platform.machine():
download_url = "https://playwright.azureedge.net/builds/chromium/1088/chromium-linux-arm64.zip"; download_url = "https://playwright.azureedge.net/builds/chromium/1088/chromium-linux-arm64.zip"
if not os.path.exists(download_path): if not os.path.exists(download_path):
os.makedirs(download_path, exist_ok=True) os.makedirs(download_path, exist_ok=True)
target_file = os.path.join( target_file = os.path.join(
@ -1046,24 +1168,24 @@ async def main(workList, uid, oocr, oocrDet):
return "mac" return "mac"
else: else:
return "unknown" return "unknown"
print("判断初始化浏览器") logger.info("初始化浏览器")
chromium_path = await init_chrome() chromium_path = await init_chrome()
headless = 'new' headless = 'new'
#headless = False #headless = False
print("选择登录") logger.info("选择登录")
try_time = 1 try_time = 1
while True: while True:
if workList[uid].type == "phone": if workList[uid].type == "phone":
print("选择手机号登录") logger.info("选择手机号登录")
result = await loginPhone(chromium_path, workList, uid, headless) result = await loginPhone(chromium_path, workList, uid, headless)
elif workList[uid].type == "password": elif workList[uid].type == "password":
print("选择密码登录") logger.info("选择密码登录")
result = await loginPassword(chromium_path, workList, uid, headless) result = await loginPassword(chromium_path, workList, uid, headless)
if result != "notSupport" or try_time > 5: if result != "notSupport" or try_time > 5:
break break
await asyncio.sleep(random.uniform(2, 4)) await asyncio.sleep(random.uniform(2, 4))
print(f"进行第{try_time}次重试") logger.info(f"进行第{try_time}次重试")
try_time += 1 try_time += 1
if os.path.exists("image.png"): if os.path.exists("image.png"):
os.remove("image.png") os.remove("image.png")
@ -1076,5 +1198,5 @@ async def main(workList, uid, oocr, oocrDet):
if os.path.exists("rgb_word_img.png"): if os.path.exists("rgb_word_img.png"):
os.remove("rgb_word_img.png") os.remove("rgb_word_img.png")
await deleteSessionDelay(workList, uid) await deleteSessionDelay(workList, uid)
print("登录完成") logger.info("登录完成")
await asyncio.sleep(10) await asyncio.sleep(10)