feat: update driver BaseClass
This commit is contained in:
koko
580
app/amazon/base.py
Normal file
580
app/amazon/base.py
Normal file
@@ -0,0 +1,580 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
import traceback
|
||||
import uuid
|
||||
from typing import Literal
|
||||
|
||||
import requests
|
||||
from DrissionPage import Chromium
|
||||
from DrissionPage._pages.chromium_tab import ChromiumTab
|
||||
from DrissionPage.common import By
|
||||
|
||||
try:
|
||||
import winreg
|
||||
except ImportError:
|
||||
winreg = None
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def kill_process(version: Literal["v5", "v6"]):
|
||||
"""杀紫鸟客户端进程（独立函数版本）"""
|
||||
logger.info("准备杀紫鸟客户端进程，version=%s", version)
|
||||
driver = ZiniaoDriver({})
|
||||
driver.kill_process(version)
|
||||
|
||||
|
||||
class ZiniaoDriver:
|
||||
"""紫鸟浏览器自动化驱动类"""
|
||||
|
||||
def __init__(self, user_info: dict, socket_port: int = 19890):
|
||||
"""
|
||||
初始化紫鸟浏览器驱动
|
||||
|
||||
Args:
|
||||
user_info: 用户信息字典，包含 company, username, password
|
||||
socket_port: 客户端通信端口，默认 19890
|
||||
"""
|
||||
self.user_info = user_info
|
||||
self.socket_port = socket_port
|
||||
self.client_path = None
|
||||
self.browser = None
|
||||
self.tab: ChromiumTab = None
|
||||
self.store_id = None
|
||||
|
||||
def get_zinaio_exe(self, protocol_name: str = "superbrowser"):
|
||||
"""
|
||||
获取紫鸟安装目录
|
||||
|
||||
Args:
|
||||
protocol_name: 协议名称，默认 "superbrowser"
|
||||
|
||||
Returns:
|
||||
exe_path: 紫鸟浏览器可执行文件路径
|
||||
"""
|
||||
if winreg is None:
|
||||
logger.error("当前系统不支持 winreg，无法从注册表获取紫鸟客户端路径")
|
||||
return None
|
||||
|
||||
key_path = rf"SOFTWARE\Classes\{protocol_name}\shell\open\command"
|
||||
for root_key, root_name in (
|
||||
(winreg.HKEY_CURRENT_USER, "HKEY_CURRENT_USER"),
|
||||
(winreg.HKEY_LOCAL_MACHINE, "HKEY_LOCAL_MACHINE"),
|
||||
):
|
||||
try:
|
||||
logger.info("正在从注册表读取紫鸟客户端路径：%s\\%s", root_name, key_path)
|
||||
key = winreg.OpenKey(root_key, key_path)
|
||||
command, _ = winreg.QueryValueEx(key, "")
|
||||
winreg.CloseKey(key)
|
||||
if isinstance(command, str):
|
||||
sub = "ziniao.exe"
|
||||
exe_path = command[0: command.find(sub) + len(sub) + 1]
|
||||
else:
|
||||
exe_path = command[0]
|
||||
logger.info("已获取紫鸟客户端路径：%s", exe_path)
|
||||
return exe_path
|
||||
except FileNotFoundError:
|
||||
logger.warning("注册表中未找到紫鸟客户端路径：%s\\%s", root_name, key_path)
|
||||
|
||||
logger.error("未能获取紫鸟客户端路径，protocol_name=%s", protocol_name)
|
||||
return None
|
||||
|
||||
def update_core(self):
|
||||
"""
|
||||
下载所有内核，打开店铺前调用，需客户端版本5.285.7以上
|
||||
因为http有超时时间，所以这个action适合循环调用，直到返回成功
|
||||
"""
|
||||
data = {
|
||||
"action": "updateCore",
|
||||
"requestId": str(uuid.uuid4()),
|
||||
}
|
||||
data.update(self.user_info)
|
||||
url = f"http://127.0.0.1:{self.socket_port}"
|
||||
|
||||
logger.info("开始更新紫鸟浏览器内核")
|
||||
while True:
|
||||
response = requests.post(url, json.dumps(data).encode("utf-8"), timeout=120)
|
||||
result = response.json()
|
||||
logger.info("更新内核返回：%s", result)
|
||||
if result is None:
|
||||
logger.info("等待客户端启动后继续更新内核")
|
||||
time.sleep(2)
|
||||
continue
|
||||
if result.get("statusCode") is None or result.get("statusCode") == -10003:
|
||||
logger.error("当前紫鸟客户端版本不支持更新内核接口，请升级客户端")
|
||||
return
|
||||
if result.get("statusCode") == 0:
|
||||
logger.info("紫鸟浏览器内核更新完成")
|
||||
return
|
||||
|
||||
logger.info("等待更新内核完成：%s", json.dumps(result, ensure_ascii=False))
|
||||
time.sleep(2)
|
||||
|
||||
def kill_process(self, version: Literal["v5", "v6"]):
|
||||
"""
|
||||
杀紫鸟客户端进程
|
||||
|
||||
Args:
|
||||
version: 客户端版本
|
||||
"""
|
||||
if version == "v5":
|
||||
process_name = "SuperBrowser.exe"
|
||||
logger.info("准备结束紫鸟 v5 starter.exe")
|
||||
os.system("taskkill /f /t /im starter.exe")
|
||||
else:
|
||||
process_name = "ziniao.exe"
|
||||
|
||||
logger.info("准备结束紫鸟客户端进程：%s", process_name)
|
||||
os.system("taskkill /f /t /im " + process_name)
|
||||
time.sleep(3)
|
||||
|
||||
def get_browser_list(self) -> list:
|
||||
"""
|
||||
获取浏览器列表
|
||||
|
||||
Returns:
|
||||
list: 浏览器列表
|
||||
"""
|
||||
request_id = str(uuid.uuid4())
|
||||
data = {
|
||||
"action": "getBrowserList",
|
||||
"requestId": request_id,
|
||||
}
|
||||
data.update(self.user_info)
|
||||
|
||||
logger.info("开始获取紫鸟浏览器列表，requestId=%s", request_id)
|
||||
url = f"http://127.0.0.1:{self.socket_port}"
|
||||
response = requests.post(url, json.dumps(data).encode("utf-8"), timeout=120)
|
||||
result = response.json()
|
||||
if str(result.get("statusCode")) == "0":
|
||||
browser_list = result.get("browserList")
|
||||
logger.info("获取紫鸟浏览器列表成功，数量=%s", len(browser_list or []))
|
||||
return browser_list
|
||||
if str(result.get("statusCode")) == "-10003":
|
||||
logger.error("获取紫鸟浏览器列表登录失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
return None
|
||||
|
||||
logger.error("获取紫鸟浏览器列表失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
return None
|
||||
|
||||
def open_store(
|
||||
self,
|
||||
store_info,
|
||||
isWebDriverReadOnlyMode=0,
|
||||
isprivacy=0,
|
||||
isHeadless=0,
|
||||
cookieTypeSave=0,
|
||||
jsInfo="",
|
||||
):
|
||||
"""
|
||||
打开店铺
|
||||
|
||||
Args:
|
||||
store_info: 店铺信息（browserId 或 browserOauth）
|
||||
isWebDriverReadOnlyMode: 是否只读模式，默认 0
|
||||
isprivacy: 隐私模式，默认 0
|
||||
isHeadless: 无头模式，默认 0
|
||||
cookieTypeSave: cookie保存类型，默认 0
|
||||
jsInfo: 注入的JS信息，默认 ""
|
||||
|
||||
Returns:
|
||||
dict: 返回结果
|
||||
"""
|
||||
request_id = str(uuid.uuid4())
|
||||
data = {
|
||||
"action": "startBrowser",
|
||||
"isWaitPluginUpdate": 0,
|
||||
"isHeadless": isHeadless,
|
||||
"requestId": request_id,
|
||||
"isWebDriverReadOnlyMode": isWebDriverReadOnlyMode,
|
||||
"cookieTypeLoad": 0,
|
||||
"cookieTypeSave": cookieTypeSave,
|
||||
"runMode": "1",
|
||||
"isLoadUserPlugin": True,
|
||||
"pluginIdType": 1,
|
||||
"privacyMode": isprivacy,
|
||||
}
|
||||
data.update(self.user_info)
|
||||
|
||||
if store_info.isdigit():
|
||||
data["browserId"] = store_info
|
||||
else:
|
||||
data["browserOauth"] = store_info
|
||||
|
||||
if len(str(jsInfo)) > 2:
|
||||
data["injectJsInfo"] = json.dumps(jsInfo)
|
||||
|
||||
logger.info("开始打开紫鸟店铺，store_info=%s，requestId=%s", store_info, request_id)
|
||||
url = f"http://127.0.0.1:{self.socket_port}"
|
||||
response = requests.post(url, json.dumps(data).encode("utf-8"), timeout=120)
|
||||
result = response.json()
|
||||
if str(result.get("statusCode")) == "0":
|
||||
logger.info("紫鸟店铺打开成功，debuggingPort=%s", result.get("debuggingPort"))
|
||||
return result
|
||||
if str(result.get("statusCode")) == "-10003":
|
||||
logger.error("打开紫鸟店铺登录失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
raise RuntimeError(f"【open_store】登录失败 {json.dumps(result, ensure_ascii=False)}")
|
||||
|
||||
logger.error("打开紫鸟店铺失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
raise RuntimeError(f"【open_store】失败 {json.dumps(result, ensure_ascii=False)} ")
|
||||
|
||||
def get_browser(self, port) -> Chromium:
|
||||
"""
|
||||
获取浏览器实例
|
||||
|
||||
Args:
|
||||
port: 调试端口
|
||||
|
||||
Returns:
|
||||
Chromium: DrissionPage浏览器实例
|
||||
"""
|
||||
logger.info("开始连接 DrissionPage 浏览器，port=%s", port)
|
||||
browser = Chromium(port)
|
||||
logger.info("DrissionPage 浏览器连接完成，port=%s", port)
|
||||
return browser
|
||||
|
||||
def start_client(self):
|
||||
"""启动紫鸟客户端"""
|
||||
url = f"http://127.0.0.1:{self.socket_port}"
|
||||
|
||||
try:
|
||||
requests.get(url, timeout=2)
|
||||
logger.info("端口 %s 已启动，跳过启动紫鸟客户端", self.socket_port)
|
||||
return
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
logger.info("端口 %s 未启动，开始启动紫鸟客户端", self.socket_port)
|
||||
|
||||
self.kill_process("v6")
|
||||
time.sleep(5)
|
||||
client_path = self.get_zinaio_exe("superbrowserv6")
|
||||
if not client_path:
|
||||
raise RuntimeError("未找到紫鸟客户端路径，无法启动客户端")
|
||||
|
||||
self.client_path = client_path.strip('"')
|
||||
cmd = [
|
||||
self.client_path,
|
||||
"--run_type=web_driver",
|
||||
"--ipc_type=http",
|
||||
"--port=" + str(self.socket_port),
|
||||
]
|
||||
logger.info("紫鸟客户端启动命令：%s", " ".join(cmd))
|
||||
|
||||
max_retries = 3
|
||||
for retry_count in range(max_retries):
|
||||
logger.info("第 %s/%s 次尝试启动紫鸟客户端", retry_count + 1, max_retries)
|
||||
subprocess.Popen(cmd)
|
||||
|
||||
start_check_time = time.time()
|
||||
client_started = False
|
||||
while time.time() - start_check_time < 10:
|
||||
try:
|
||||
requests.get(url, timeout=2)
|
||||
logger.info("紫鸟客户端启动成功，第 %s 次尝试", retry_count + 1)
|
||||
client_started = True
|
||||
break
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
|
||||
time.sleep(0.5)
|
||||
|
||||
if client_started:
|
||||
time.sleep(5)
|
||||
self.update_core()
|
||||
return
|
||||
|
||||
logger.warning("第 %s 次尝试启动失败，10秒内未检测到客户端启动", retry_count + 1)
|
||||
|
||||
logger.error("紫鸟客户端启动失败，已重试 %s 次", max_retries)
|
||||
raise RuntimeError(f"客户端启动失败：重试 {max_retries} 次后仍未成功启动")
|
||||
|
||||
def open_shop(self, shop_name: str):
|
||||
"""
|
||||
打开指定店铺
|
||||
|
||||
Args:
|
||||
shop_name: 店铺名称
|
||||
|
||||
Returns:
|
||||
Chromium or str: 成功返回浏览器实例，失败返回错误信息
|
||||
"""
|
||||
logger.info("开始打开店铺：%s", shop_name)
|
||||
self.start_client()
|
||||
|
||||
shop_ls = self.get_browser_list()
|
||||
self.store_id = None
|
||||
for shop in shop_ls or []:
|
||||
if shop.get("browserName") == shop_name:
|
||||
self.store_id = shop.get("browserOauth")
|
||||
break
|
||||
|
||||
if not self.store_id:
|
||||
logger.warning("店铺不存在：%s", shop_name)
|
||||
return "店铺不存在"
|
||||
|
||||
ret_json = self.open_store(self.store_id)
|
||||
self.store_id = ret_json.get("browserOauth")
|
||||
if self.store_id is None:
|
||||
self.store_id = ret_json.get("browserId")
|
||||
|
||||
self.browser = self.get_browser(ret_json.get("debuggingPort"))
|
||||
|
||||
ip_check_url = ret_json.get("ipDetectionPage")
|
||||
if not ip_check_url:
|
||||
logger.error("ip检测页地址为空，准备关闭店铺：%s", shop_name)
|
||||
self.close_store(self.store_id)
|
||||
raise RuntimeError("没有IP检测地址，为了店铺安全不打开店铺")
|
||||
|
||||
ip_usable = self.open_ip_check(self.browser, ip_check_url)
|
||||
if ip_usable:
|
||||
logger.info("IP检测通过，打开店铺平台主页：%s", shop_name)
|
||||
self.open_launcher_page(ret_json.get("launcherPage"), self.browser)
|
||||
else:
|
||||
logger.error("IP检测不通过，停止打开店铺：%s", shop_name)
|
||||
raise RuntimeError("IP检测不通过，可能是因为网络环境变化导致的，为了店铺安全不打开店铺")
|
||||
return self.browser
|
||||
|
||||
def close_store(self, browser_oauth=None):
|
||||
"""
|
||||
关闭店铺
|
||||
|
||||
Args:
|
||||
browser_oauth: 店铺OAuth标识，如果不提供则使用当前打开的店铺
|
||||
|
||||
Returns:
|
||||
dict: 返回结果
|
||||
"""
|
||||
if browser_oauth is None:
|
||||
browser_oauth = self.store_id
|
||||
|
||||
request_id = str(uuid.uuid4())
|
||||
data = {
|
||||
"action": "stopBrowser",
|
||||
"requestId": request_id,
|
||||
"duplicate": 0,
|
||||
"browserOauth": browser_oauth,
|
||||
}
|
||||
data.update(self.user_info)
|
||||
|
||||
logger.info("开始关闭紫鸟店铺，browserOauth=%s，requestId=%s", browser_oauth, request_id)
|
||||
url = f"http://127.0.0.1:{self.socket_port}"
|
||||
response = requests.post(url, json.dumps(data).encode("utf-8"), timeout=120)
|
||||
result = response.json()
|
||||
if str(result.get("statusCode")) == "0":
|
||||
logger.info("紫鸟店铺关闭成功，browserOauth=%s", browser_oauth)
|
||||
return result
|
||||
if str(result.get("statusCode")) == "-10003":
|
||||
logger.error("关闭紫鸟店铺登录失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
raise RuntimeError(f"【close_store】登录失败 {json.dumps(result, ensure_ascii=False)}")
|
||||
|
||||
logger.error("关闭紫鸟店铺失败：%s", json.dumps(result, ensure_ascii=False))
|
||||
raise RuntimeError(f"【close_store】失败： {json.dumps(result, ensure_ascii=False)} ")
|
||||
|
||||
def open_launcher_page(self, launcher_page: str, browser: Chromium = None):
|
||||
"""
|
||||
打开启动页面
|
||||
|
||||
Args:
|
||||
launcher_page: 要打开的页面URL
|
||||
browser: 浏览器实例，如果不提供则使用当前浏览器实例
|
||||
"""
|
||||
if browser is None:
|
||||
browser = self.browser
|
||||
|
||||
logger.info("打开店铺平台启动页：%s", launcher_page)
|
||||
tab = browser.new_tab(url=launcher_page)
|
||||
self.tab = tab
|
||||
return tab
|
||||
|
||||
def open_ip_check(self, browser: Chromium, ip_check_url: str):
|
||||
"""
|
||||
打开ip检测页检测ip是否正常
|
||||
:param browser: drissionpage浏览器会话
|
||||
:param ip_check_url ip检测页地址
|
||||
:return 检测结果
|
||||
"""
|
||||
try:
|
||||
logger.info("开始打开IP检测页：%s", ip_check_url)
|
||||
tab = browser.latest_tab
|
||||
tab.get(ip_check_url)
|
||||
success_button = tab.ele(
|
||||
(By.XPATH, '//button[contains(@class, "styles_btn--success")]'),
|
||||
timeout=60,
|
||||
)
|
||||
if success_button:
|
||||
logger.info("IP检测成功")
|
||||
return True
|
||||
|
||||
logger.warning("IP检测超时或未找到成功按钮")
|
||||
return False
|
||||
except Exception:
|
||||
logger.exception("IP检测异常")
|
||||
return False
|
||||
|
||||
|
||||
class AmamzonBase(ZiniaoDriver):
|
||||
"""亚马逊操作基类，包含一些通用方法"""
|
||||
|
||||
def SwitchingCountries(self, country_name: str):
|
||||
"""
|
||||
切换国家
|
||||
操作：
|
||||
1、//div[@class="dropdown-account-switcher-header-label"]/span[last()] 获取此元素文本，判断当前国家，如果与目标国家相同则不操作，否则执行下一步
|
||||
2、点击 //div[@class="dropdown-account-switcher-header-label"] 打开下拉框
|
||||
3、点击 //div[@class="dropdown-account-switcher-list-item"] 第一个展开国家列表
|
||||
4、点击 //div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="国家名"] 切换到目标国家
|
||||
5、等待页面加载完成，判断国家是否切换成功，成功则返回True，否则返回False
|
||||
|
||||
Args:
|
||||
country_name: 目标国家名称
|
||||
|
||||
Returns:
|
||||
bool: 切换成功返回True，失败返回False
|
||||
"""
|
||||
try:
|
||||
if self.browser is None:
|
||||
logger.warning("浏览器实例不存在，请先打开店铺")
|
||||
return False
|
||||
|
||||
tab = self.tab
|
||||
tab.wait.doc_loaded(timeout=120, raise_err=False)
|
||||
|
||||
logger.info("正在检查当前国家")
|
||||
current_country_ele = tab.ele(
|
||||
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]',
|
||||
timeout=20,
|
||||
)
|
||||
if current_country_ele:
|
||||
current_country = current_country_ele.text.strip()
|
||||
logger.info("当前国家：%s", current_country)
|
||||
|
||||
if current_country == country_name:
|
||||
logger.info("当前已经是目标国家 %s，无需切换", country_name)
|
||||
return True
|
||||
else:
|
||||
logger.warning("无法获取当前国家信息")
|
||||
return False
|
||||
|
||||
logger.info("正在打开国家切换下拉框")
|
||||
dropdown_header = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]', timeout=10)
|
||||
if not dropdown_header:
|
||||
logger.warning("找不到国家切换下拉框")
|
||||
return False
|
||||
dropdown_header.click()
|
||||
time.sleep(1)
|
||||
|
||||
logger.info("正在展开国家列表")
|
||||
first_item = tab.ele('xpath://div[@class="dropdown-account-switcher-list-item"]', timeout=10)
|
||||
if not first_item:
|
||||
logger.warning("找不到国家列表项")
|
||||
return False
|
||||
first_item.click()
|
||||
time.sleep(1)
|
||||
|
||||
logger.info("正在切换到国家：%s", country_name)
|
||||
target_country_xpath = f'//div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="{country_name}"]'
|
||||
target_country = tab.ele(f"xpath:{target_country_xpath}", timeout=10)
|
||||
if not target_country:
|
||||
logger.warning("找不到目标国家：%s", country_name)
|
||||
return False
|
||||
target_country.click()
|
||||
|
||||
logger.info("等待国家切换后页面加载")
|
||||
self.tab.wait.doc_loaded()
|
||||
|
||||
new_country_ele = tab.ele(
|
||||
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]',
|
||||
timeout=10,
|
||||
)
|
||||
if new_country_ele:
|
||||
new_country = new_country_ele.text.strip()
|
||||
if new_country == country_name:
|
||||
logger.info("国家切换成功：%s", new_country)
|
||||
return True
|
||||
|
||||
logger.warning("国家切换失败，当前国家：%s，目标国家：%s", new_country, country_name)
|
||||
return False
|
||||
|
||||
logger.warning("无法验证国家切换结果")
|
||||
return False
|
||||
|
||||
except Exception:
|
||||
logger.exception("切换国家时发生异常")
|
||||
return False
|
||||
|
||||
def need_login(self):
|
||||
"""
|
||||
判断是否需要登录，部分国家可能需要登录后才能切换国家
|
||||
"""
|
||||
time.sleep(3)
|
||||
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
|
||||
need_login_ele = self.tab.eles(
|
||||
'xpath://h1[@class="a-spacing-small"]|//span[contains(text(),"登录")]',
|
||||
timeout=5,
|
||||
)
|
||||
if len(need_login_ele) > 0:
|
||||
logger.info("检测到需要登录元素")
|
||||
return True
|
||||
logger.info("未检测到登录元素")
|
||||
return False
|
||||
|
||||
def login(self, password, username=""):
|
||||
try:
|
||||
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
|
||||
for _ in range(4):
|
||||
pwd_input = self.tab.eles('xpath://input[@type="password"]', timeout=5)
|
||||
if len(pwd_input) > 0:
|
||||
logger.info("检测到密码输入框，准备输入密码")
|
||||
pwd_input[0].input(password, clear=True)
|
||||
submit_btn = self.tab.eles('xpath://input[@id="signInSubmit"]', timeout=5)
|
||||
if len(submit_btn) > 0:
|
||||
logger.info("点击登录提交按钮")
|
||||
submit_btn[0].click()
|
||||
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
|
||||
|
||||
send_code = self.tab.eles(
|
||||
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]',
|
||||
timeout=5,
|
||||
)
|
||||
logger.info("发送一次性密码元素数量：%s", len(send_code))
|
||||
if len(send_code) > 0:
|
||||
logger.info("检测到发送一次性密码按钮，准备点击")
|
||||
send_code[0].click()
|
||||
time.sleep(1)
|
||||
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
|
||||
else:
|
||||
for _ in range(5):
|
||||
opt_code_input = self.tab.eles('xpath://input[@name="otpCode"]', timeout=30)
|
||||
logger.info("验证码输入框数量：%s", len(opt_code_input))
|
||||
if len(opt_code_input) > 0:
|
||||
opt_code_input[0].wait.displayed(timeout=10, raise_err=False)
|
||||
for _ in range(30):
|
||||
if opt_code_input[0].value is not None and opt_code_input[0].value.strip() != "":
|
||||
logger.info("检测到验证码输入完成")
|
||||
submit_btn = self.tab.eles('xpath://input[@id="auth-signin-button"]', timeout=10)
|
||||
if len(submit_btn) > 0:
|
||||
submit_btn[0].click()
|
||||
self.tab.wait.doc_loaded(timeout=20, raise_err=False)
|
||||
error_mes = self.tab.eles(
|
||||
'xpath://div[@id="auth-error-message-box"]',
|
||||
timeout=10,
|
||||
)
|
||||
if len(error_mes) > 0:
|
||||
logger.warning("验证码输入错误：%s", error_mes[0].text)
|
||||
self.tab.refresh()
|
||||
else:
|
||||
logger.info("登录成功")
|
||||
return True
|
||||
|
||||
time.sleep(1)
|
||||
else:
|
||||
break
|
||||
|
||||
submit_btn = self.tab.eles('xpath://input[@id="auth-signin-button"]', timeout=10)
|
||||
if len(submit_btn) > 0:
|
||||
logger.info("点击二次登录提交按钮")
|
||||
submit_btn[0].click()
|
||||
except Exception:
|
||||
logger.exception("登录过程中发生异常：%s", traceback.format_exc())
|
||||
return False
|
||||
Reference in New Issue
Block a user