zayac/tools-pyqt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

设置了休眠时间,防止浏览器过早关闭

Browse Source
This commit is contained in:
zayac 2024-05-31 18:02:44 +08:00
parent c79c832128
commit 63588e4f27
1 changed files with 92 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

225
src/core/login.py
View File

@ -1,88 +1,69 @@
import base64 import base64
import time
import re import re
import asyncio from playwright.sync_api import sync_playwright, TimeoutError
from playwright.async_api import async_playwright, TimeoutError
from loguru import logger from loguru import logger
from src.core.message_client import send_message
from src.entity.account import Account, AccountType from src.entity.account import Account, AccountType
import aiohttp import requests
from src.entity.database import db from src.entity.database import db
from src.ui import global_signals
# 提取重复的请求和响应处理逻辑 def handle_request(request, result_container, response_container):
async def handle_event(page, event_type, callback, *args): if 'loginFlow' in request.url:
page.on(event_type, lambda event: asyncio.create_task(callback(event, *args))) response = request.response()
if response:
data = response.json()
async def handle_request(request, result_container, request_event): response_container['data'] = data
if 'perInfo' in request.url: if 'banner' in request.url:
headers = request.headers headers = request.headers
result_container['headers'] = headers result_container['headers'] = headers
logger.info(f'Successfully got headers: {headers}') logger.info(f'Successfully got headers: {headers}')
request_event.set()
async def handle_response(response, response_container, response_event): def playwright_login(account: Account) -> dict:
if 'loginFlow' in response.url:
data = await response.json()
response_container['data'] = data
response_event.set()
async def playwright_login(account: Account) -> dict:
result_container = {'headers': {}} result_container = {'headers': {}}
response_container = {'data': {}} response_container = {'data': {}}
logger.info(f'Starting login for username: {account.username}') logger.info(f'Starting login for username: {account.username}')
try: try:
async with async_playwright() as playwright: with sync_playwright() as playwright:
await perform_login(playwright, account, result_container, response_container) perform_login(playwright, account, result_container, response_container)
return result_container['headers']
except Exception as e: except Exception as e:
logger.error(f'Error during login for account {account.username}: {e}', exc_info=True) logger.error(f'Error during login for account {account.username}: {e}', exc_info=True)
await handle_login_failure(account) handle_login_failure(account)
return result_container['headers']
async def perform_login(playwright, account: Account, request_container: dict, def perform_login(playwright, account: Account, request_container: dict, response_container: dict):
response_container: dict): browser = playwright.chromium.launch(headless=False)
browser = await playwright.chromium.launch(headless=False) context = browser.new_context()
context = await browser.new_context() page = context.new_page()
page = await context.new_page()
request_event = asyncio.Event() page.on("request", lambda request: handle_request(request, request_container, response_container))
response_event = asyncio.Event()
await page.goto(account.url) page.goto(account.url, timeout=60000)
await handle_event(page, 'request', handle_request, request_container, request_event)
await handle_event(page, 'response', handle_response, response_container, response_event)
# 填充基本的元素 # 填充基本的元素
await fill_form_common(page, account) fill_form_common(page, account)
# 如果是九游 验证码滑动,并且滑动之后自动登录成功
if account.type == AccountType.jy: if account.type == AccountType.jy:
await handle_jy_captcha(page, account, 0) handle_jy_captcha(page, account)
logger.info("验证码处理成功") logger.info("验证码处理成功")
try: time.sleep(3)
# 等待请求发送 获取header返回值
await request_event.wait()
except TimeoutError:
# 超时 即验证码处理失败
logger.warning("登录处理失败")
else: else:
# 验证码输入成功
max_retries = 3 max_retries = 3
for attempt in range(max_retries): for attempt in range(max_retries):
if await handle_captcha(page): if handle_captcha(page):
response_event.clear() # 重置事件 page.locator("div").filter(has_text=re.compile(r"^登录$")).get_by_role("button").click()
await page.locator("div").filter(has_text=re.compile(r"^登录$")).get_by_role("button").click() time.sleep(2)
try: try:
await response_event.wait() if response_container['data'] and response_container['data'].get('status_code') != 6000:
if response_container['data'].get('status_code') != 6000:
logger.warning(f"登录发生错误, {response_container['data'].get('message', '未知错误')}") logger.warning(f"登录发生错误, {response_container['data'].get('message', '未知错误')}")
close_button = page.locator("button.ant-modal-close") close_button = page.locator("button.ant-modal-close")
await close_button.wait_for() close_button.wait_for()
await close_button.click() close_button.click()
else: else:
break break
except TimeoutError: except TimeoutError:
@ -91,97 +72,79 @@ async def perform_login(playwright, account: Account, request_container: dict,
logger.warning(f"验证码输入失败，重试 {attempt + 1}/{max_retries}") logger.warning(f"验证码输入失败，重试 {attempt + 1}/{max_retries}")
else: else:
logger.warning("登录失败，达到最大重试次数") logger.warning("登录失败，达到最大重试次数")
try: persistence(account, response_container['data'])
# 等待请求发送 获取header返回值 # 通知app数据更新了
await request_event.wait() global_signals.user_data_updated.emit()
except TimeoutError: page.wait_for_url(f'{account.url}/app/home?showWelcome=false')
# 超时 即验证码处理失败 close_resources(page, context, browser)
logger.warning("登录处理失败")
await close_resources(page, context, browser)
async def fill_form_common(page, account: Account): def fill_form_common(page, account: Account):
username_input = page.get_by_placeholder('用户名') username_input = page.get_by_placeholder('用户名')
password_input = page.get_by_placeholder('密码') password_input = page.get_by_placeholder('密码')
await username_input.click() username_input.click()
await username_input.fill(account.username) username_input.fill(account.username)
await password_input.click() password_input.click()
await password_input.fill(account.password) password_input.fill(account.password)
async def handle_jy_captcha(page, account, retry_count=0): def handle_jy_captcha(page, account, retry_count=0):
await page.locator("div").filter(has_text=re.compile(r"^登录$")).get_by_role("button").click() if retry_count < 3:
validate_code = page.locator('.geetest_box') page.locator("div").filter(has_text=re.compile(r"^登录$")).get_by_role("button").click()
await validate_code.wait_for(state='visible') validate_code = page.locator('.geetest_box')
await asyncio.sleep(1) validate_code.wait_for(state='visible')
validate_code_buffer = await validate_code.screenshot() time.sleep(1)
img = base64.b64encode(validate_code_buffer).decode('utf-8') validate_code_buffer = validate_code.screenshot()
res = await base64_api(img=img, typeid=33) img = base64.b64encode(validate_code_buffer).decode('utf-8')
drag_btn = validate_code.locator(".geetest_btn") res = base64_api(img=img, typeid=33)
drag_btn_box = await drag_btn.bounding_box() drag_btn = validate_code.locator(".geetest_btn")
try: drag_btn_box = drag_btn.bounding_box()
distance = int(res) + 3 try:
target_x = drag_btn_box['x'] + distance distance = int(res) + 3
await drag_btn.hover() target_x = drag_btn_box['x'] + distance
await page.mouse.down() drag_btn.hover()
await page.mouse.move(target_x, drag_btn_box["y"], steps=25) page.mouse.down()
await page.mouse.up() page.mouse.move(target_x, drag_btn_box["y"], steps=25)
await validate_code.wait_for(state='hidden') page.mouse.up()
return True validate_code.wait_for(state='hidden')
except (ValueError, TimeoutError): return True
retry_count += 1 except (ValueError, TimeoutError):
logger.error(f"验证码处理失败，重试: {res}") retry_count += 1
return await handle_jy_captcha(page, account, retry_count) logger.error(f"验证码处理失败，重试: {retry_count}")
return handle_jy_captcha(page, account, retry_count)
else:
logger.error("验证码处理失败")
async def handle_captcha(page): def handle_captcha(page):
validate_code_input = page.get_by_placeholder('请输入验证码') validate_code_input = page.get_by_placeholder('请输入验证码')
validate_code_img = page.locator('#boxModle') validate_code_img = page.locator('#boxModle')
await validate_code_img.wait_for() validate_code_img.wait_for()
validate_code_buffer = await validate_code_img.screenshot() validate_code_buffer = validate_code_img.screenshot()
img_base64 = base64.b64encode(validate_code_buffer).decode('utf-8') img_base64 = base64.b64encode(validate_code_buffer).decode('utf-8')
res = await base64_api(img=img_base64, typeid=1003) res = base64_api(img=img_base64, typeid=1003)
await validate_code_input.click() validate_code_input.click()
await validate_code_input.fill(res) validate_code_input.fill(res)
return True return True
async def handle_standard_captcha(page, validate_code, img): def close_resources(page, context, browser):
res = await base64_api(img=img)
if '|' in res:
await click_captcha_positions(validate_code, res)
await validate_code.locator('.geetest_submit').click()
await validate_code.wait_for(state='hidden')
logger.debug('验证码点击成功')
return True
else:
logger.error(res)
return False
async def click_captcha_positions(validate_code, positions_str):
for part in positions_str.split('|'):
x, y = part.split(',')
await validate_code.click(position={"x": int(x), "y": int(y)})
await asyncio.sleep(.5)
async def close_resources(page, context, browser):
try: try:
await page.close() if page and not page.is_closed():
await context.close() page.close()
await browser.close() if context:
context.close()
if browser:
browser.close()
except Exception as e: except Exception as e:
logger.error(f'Error closing browser resources: {e}', exc_info=True) logger.error(f'Error closing browser resources: {e}', exc_info=True)
async def handle_login_failure(account: Account): def handle_login_failure(account: Account):
send_message(account.user.bot_token, account.user.group_id, pass
f'{account.url}:加载超时，请检查是否后台更换了链接')
return
async def base64_api(uname='luffy230505', pwd='qwer12345', img='', typeid=20): def base64_api(uname='luffy230505', pwd='qwer12345', img='', typeid=20):
logger.info('Calling third part interfaces') logger.info('Calling third part interfaces')
data = { data = {
"username": uname, "username": uname,
@ -190,25 +153,21 @@ async def base64_api(uname='luffy230505', pwd='qwer12345', img='', typeid=20):
"image": img, "image": img,
'softid': '8d13df0efe074035b54ee9c2bef85106' 'softid': '8d13df0efe074035b54ee9c2bef85106'
} }
async with aiohttp.ClientSession() as session: response = requests.post("http://api.ttshitu.com/predict", json=data)
async with session.post("http://api.ttshitu.com/predict", json=data) as response: result = response.json()
result = await response.json() if result['success']:
if result['success']: return result["data"]["result"]
return result["data"]["result"] else:
else: return result["message"]
return result["message"]
def login(account: Account): def login(account: Account):
headers = asyncio.run(playwright_login(account)) headers = playwright_login(account)
if headers: if headers:
account.headers = headers account.headers = headers
persistence(account, headers)
return account return account
else: else:
send_message(account.user.bot_token, account.user.group_id, pass
f'{account.url}:加载超时，请检查是否后台更换了链接')
return
def persistence(account: Account, headers: dict): def persistence(account: Account, headers: dict):