super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update driver BaseClass

Browse Source
This commit is contained in:
koko
2026-04-25 17:14:23 +08:00
parent 26a920a1cc
commit f9b2d2da25
8 changed files with 2595 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
app/.env
View File

@@ -1,4 +1,3 @@
base_url=http://8.136.19.173:15124
workflow_id=7608812635877900322
mysql_host=8.136.19.173
mysql_user=aiimage
@@ -12,8 +11,8 @@ zn_username=%E8%87%AA%E5%8A%A8%E5%8C%96_Robot
client_name=ShuFuAI
# java_api_base=http://47.111.163.154:18080
java_api_base=http://127.0.0.1:18080
java_api_base=http://47.111.163.154:18080
# java_api_base=http://127.0.0.1:18080
# java_api_base=http://8.136.19.173:18080

18
app/.env.example Normal file
View File

@@ -0,0 +1,18 @@
workflow_id=7608812635877900322
mysql_host=8.136.19.173
mysql_user=aiimage
proxy_url=https://api.jikip.com/ip-get?num=1&minute=1&format=json&area=all&protocol=1&mode=2&key=t24g6gi44ubufd8
proxy_mode=2
zn_company=rongchuang123
zn_username=%E8%87%AA%E5%8A%A8%E5%8C%96_Robot
client_name=ShuFuAI
java_api_base=http://47.111.163.154:18080
# java_api_base=http://127.0.0.1:18080
# java_api_base=http://8.136.19.173:18080

580
app/amazon/base.py Normal file
View 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

98
app/pyproject.toml Normal file
View File

@@ -0,0 +1,98 @@
[project]
name = "crawler-desktop-app"
version = "0.1.0"
description = "Desktop crawler app backend and pywebview shell."
requires-python = ">=3.9,<3.10"
dependencies = [
"alibabacloud-oss-v2==1.2.4",
"altgraph==0.17.5",
"annotated-types==0.7.0",
"anyio==4.12.1",
"Authlib==1.6.8",
"blinker==1.9.0",
"bottle==0.13.4",
"certifi==2026.1.4",
"cffi==2.0.0",
"charset-normalizer==3.4.4",
"click==8.1.8",
"clr_loader==0.2.10",
"colorama==0.4.6",
"cozepy==0.20.0",
"crcmod-plus==2.3.1",
"cryptography==41.0.0",
"cssselect==1.3.0",
"DataRecorder==3.6.2",
"distro==1.9.0",
"DownloadKit==2.0.7",
"DrissionPage==4.1.1.2",
"et_xmlfile==2.0.0",
"exceptiongroup==1.3.1",
"filelock==3.19.1",
"Flask==3.1.3",
"flask-cors==6.0.2",
"h11==0.16.0",
"httpcore==1.0.9",
"httpx==0.28.1",
"idna==3.11",
"importlib_metadata==8.7.1",
"itsdangerous==2.2.0",
"Jinja2==3.1.6",
"lxml==6.0.2",
"MarkupSafe==3.0.3",
"MouseInfo==0.1.3",
"Nuitka==2.8.6",
"numpy==1.25.2",
"openpyxl==3.1.5",
"ordered-set==4.1.0",
"packaging==26.0",
"pandas==2.3.3",
"pefile==2024.8.26",
"pillow==11.3.0",
"proxy_tools==0.1.0",
"psutil==7.2.2",
"PyAutoGUI==0.9.54",
"pycparser==2.23",
"pycryptodome==3.23.0",
"pydantic==2.12.5",
"pydantic_core==2.41.5",
"PyGetWindow==0.0.9",
"pyinstaller==6.19.0",
"pyinstaller-hooks-contrib==2026.3",
"PyMsgBox==2.0.1",
"PyMySQL==1.1.2",
"pyperclip==1.11.0",
"PyQt5==5.15.11",
"PyQt5-Qt5==5.15.2",
"PyQt5_sip==12.17.1",
"PyRect==0.2.0",
"PyScreeze==1.0.1",
"python-dateutil==2.9.0.post0",
"python-dotenv==1.2.1",
"pythonnet==3.0.5",
"pytweening==1.2.0",
"pytz==2026.1.post1",
"pywebview==6.1",
"pywin32-ctypes==0.2.3",
"requests==2.32.5",
"requests-file==3.0.1",
"six==1.17.0",
"tldextract==5.3.0",
"typing-inspection==0.4.2",
"typing_extensions==4.15.0",
"tzdata==2025.3",
"urllib3==2.6.3",
"websocket-client==1.9.0",
"websockets==14.2",
"Werkzeug==3.1.6",
"zipp==3.23.0",
"zstandard==0.25.0",
]
[tool.uv]
package = false
[dependency-groups]
dev = [
"pytest>=8.4.2",
"ruff>=0.14.0",
]

0
app/tests/conftest.py Normal file
View File

337
app/tests/test_amazon_base.py Normal file
View File

@@ -0,0 +1,337 @@
import json
import pytest
from amazon import base
class FakeResponse:
def __init__(self, payload):
self.payload = payload
def json(self):
return self.payload
class FakeWait:
def __init__(self, *, raise_on_doc_loaded=False):
self.doc_loaded_calls = []
self.displayed_calls = []
self.raise_on_doc_loaded = raise_on_doc_loaded
def doc_loaded(self, **kwargs):
if self.raise_on_doc_loaded:
raise RuntimeError("doc load failed")
self.doc_loaded_calls.append(kwargs)
def displayed(self, **kwargs):
self.displayed_calls.append(kwargs)
return True
class FakeElement:
def __init__(self, *, text="", value=None):
self.text = text
self.value = value
self.clicks = 0
self.inputs = []
self.wait = FakeWait()
def click(self, *args, **kwargs):
self.clicks += 1
def input(self, value, clear=False):
self.inputs.append((value, clear))
class FakeTab:
def __init__(self, *, ele_returns=None, eles_returns=None, raise_on_doc_loaded=False):
self.ele_returns = {key: list(values) for key, values in (ele_returns or {}).items()}
self.eles_returns = {key: list(values) for key, values in (eles_returns or {}).items()}
self.wait = FakeWait(raise_on_doc_loaded=raise_on_doc_loaded)
self.refreshes = 0
def ele(self, locator, timeout=None):
values = self.ele_returns.get(locator)
if not values:
return None
return values.pop(0)
def eles(self, locator, timeout=None):
values = self.eles_returns.get(locator)
if not values:
return []
return values.pop(0)
def refresh(self):
self.refreshes += 1
def test_ziniao_driver_initializes_browser_state():
user_info = {
"company": "test-company",
"username": "test-user",
"password": "test-password",
}
driver = base.ZiniaoDriver(user_info, socket_port=19999)
assert driver.user_info is user_info
assert driver.socket_port == 19999
assert driver.client_path is None
assert driver.browser is None
assert driver.tab is None
assert driver.store_id is None
def test_amazon_base_inherits_ziniao_driver():
driver = base.AmamzonBase({})
assert isinstance(driver, base.ZiniaoDriver)
def test_kill_process_uses_ziniao_driver(monkeypatch):
calls = []
class FakeDriver:
def __init__(self, user_info):
calls.append(("init", user_info))
def kill_process(self, version):
calls.append(("kill_process", version))
monkeypatch.setattr(base, "ZiniaoDriver", FakeDriver)
base.kill_process("v6")
assert calls == [("init", {}), ("kill_process", "v6")]
def test_get_zinaio_exe_returns_none_when_winreg_unavailable(monkeypatch, caplog):
driver = base.ZiniaoDriver({})
monkeypatch.setattr(base, "winreg", None)
with caplog.at_level("ERROR", logger=base.logger.name):
result = driver.get_zinaio_exe()
assert result is None
assert any("winreg" in message for message in caplog.messages)
@pytest.mark.parametrize(
("store_info", "expected_key"),
[
("12345", "browserId"),
("oauth-token", "browserOauth"),
],
)
def test_open_store_payload_uses_id_or_oauth(monkeypatch, store_info, expected_key):
payloads = []
def fake_post(url, data, timeout):
payloads.append(json.loads(data.decode("utf-8")))
return FakeResponse({"statusCode": 0, "debuggingPort": 9222})
monkeypatch.setattr(base.requests, "post", fake_post)
result = base.ZiniaoDriver({"company": "c"}).open_store(store_info)
assert result["debuggingPort"] == 9222
assert payloads[0]["action"] == "startBrowser"
assert payloads[0][expected_key] == store_info
@pytest.mark.parametrize(
("payload", "expected"),
[
({"statusCode": 0, "browserList": [{"browserName": "shop"}]}, [{"browserName": "shop"}]),
({"statusCode": -10003}, None),
({"statusCode": 500, "message": "failed"}, None),
],
)
def test_get_browser_list_status_handling(monkeypatch, payload, expected):
monkeypatch.setattr(base.requests, "post", lambda *args, **kwargs: FakeResponse(payload))
assert base.ZiniaoDriver({}).get_browser_list() == expected
def test_close_store_success_uses_current_store(monkeypatch):
payloads = []
def fake_post(url, data, timeout):
payloads.append(json.loads(data.decode("utf-8")))
return FakeResponse({"statusCode": 0, "ok": True})
monkeypatch.setattr(base.requests, "post", fake_post)
driver = base.ZiniaoDriver({})
driver.store_id = "oauth-1"
result = driver.close_store()
assert result["ok"] is True
assert payloads[0]["action"] == "stopBrowser"
assert payloads[0]["browserOauth"] == "oauth-1"
def test_close_store_login_failure_raises(monkeypatch):
monkeypatch.setattr(base.requests, "post", lambda *args, **kwargs: FakeResponse({"statusCode": -10003}))
with pytest.raises(RuntimeError, match="【close_store】登录失败"):
base.ZiniaoDriver({}).close_store("oauth-1")
def test_open_shop_returns_message_when_shop_missing(monkeypatch):
driver = base.ZiniaoDriver({})
monkeypatch.setattr(driver, "start_client", lambda: None)
monkeypatch.setattr(driver, "get_browser_list", lambda: [{"browserName": "other"}])
assert driver.open_shop("target") == "店铺不存在"
assert driver.store_id is None
def test_open_shop_opens_matching_store(monkeypatch):
opened_pages = []
fake_browser = object()
driver = base.ZiniaoDriver({})
monkeypatch.setattr(driver, "start_client", lambda: None)
monkeypatch.setattr(driver, "get_browser_list", lambda: [{"browserName": "target", "browserOauth": "oauth-1"}])
monkeypatch.setattr(
driver,
"open_store",
lambda store_id: {
"browserOauth": "oauth-2",
"debuggingPort": 9222,
"ipDetectionPage": "https://ip-check",
"launcherPage": "https://launcher",
},
)
monkeypatch.setattr(driver, "get_browser", lambda port: fake_browser)
monkeypatch.setattr(driver, "open_ip_check", lambda browser, url: True)
monkeypatch.setattr(driver, "open_launcher_page", lambda page, browser: opened_pages.append((page, browser)))
assert driver.open_shop("target") is fake_browser
assert driver.store_id == "oauth-2"
assert opened_pages == [("https://launcher", fake_browser)]
def test_switching_countries_returns_true_when_already_current():
current_country = FakeElement(text="西班牙")
driver = base.AmamzonBase({})
driver.browser = object()
driver.tab = FakeTab(
ele_returns={
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [current_country],
}
)
assert driver.SwitchingCountries("西班牙") is True
def test_switching_countries_returns_false_when_target_missing(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
driver = base.AmamzonBase({})
driver.browser = object()
driver.tab = FakeTab(
ele_returns={
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [FakeElement(text="英国")],
'xpath://div[@class="dropdown-account-switcher-header-label"]': [FakeElement()],
'xpath://div[@class="dropdown-account-switcher-list-item"]': [FakeElement()],
'xpath://div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="西班牙"]': [
None
],
}
)
assert driver.SwitchingCountries("西班牙") is False
def test_switching_countries_clicks_and_verifies_target(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
dropdown = FakeElement()
first_item = FakeElement()
target = FakeElement()
driver = base.AmamzonBase({})
driver.browser = object()
driver.tab = FakeTab(
ele_returns={
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [
FakeElement(text="英国"),
FakeElement(text="西班牙"),
],
'xpath://div[@class="dropdown-account-switcher-header-label"]': [dropdown],
'xpath://div[@class="dropdown-account-switcher-list-item"]': [first_item],
'xpath://div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="西班牙"]': [
target
],
}
)
assert driver.SwitchingCountries("西班牙") is True
assert dropdown.clicks == 1
assert first_item.clicks == 1
assert target.clicks == 1
def test_login_returns_true_when_otp_submit_succeeds(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
password_input = FakeElement()
password_submit = FakeElement()
otp_input = FakeElement(value="123456")
otp_submit = FakeElement()
driver = base.AmamzonBase({})
driver.tab = FakeTab(
eles_returns={
'xpath://input[@type="password"]': [[password_input]],
'xpath://input[@id="signInSubmit"]': [[password_submit]],
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
'xpath://input[@name="otpCode"]': [[otp_input]],
'xpath://input[@id="auth-signin-button"]': [[otp_submit]],
'xpath://div[@id="auth-error-message-box"]': [[]],
}
)
assert driver.login("secret") is True
assert password_input.inputs == [("secret", True)]
assert password_submit.clicks == 1
assert otp_submit.clicks == 1
def test_login_refreshes_on_otp_error(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
otp_input = FakeElement(value="123456")
driver = base.AmamzonBase({})
driver.tab = FakeTab(
eles_returns={
'xpath://input[@type="password"]': [[]],
'xpath://input[@id="signInSubmit"]': [[]],
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
'xpath://input[@name="otpCode"]': [[otp_input]],
'xpath://input[@id="auth-signin-button"]': [[FakeElement()]],
'xpath://div[@id="auth-error-message-box"]': [[FakeElement(text="验证码错误")]],
}
)
assert driver.login("secret") is False
assert driver.tab.refreshes == 1
def test_login_returns_false_when_no_otp_input(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
driver = base.AmamzonBase({})
driver.tab = FakeTab(
eles_returns={
'xpath://input[@type="password"]': [[]],
'xpath://input[@id="signInSubmit"]': [[]],
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
'xpath://input[@name="otpCode"]': [[]],
'xpath://input[@id="auth-signin-button"]': [[]],
}
)
assert driver.login("secret") is False
def test_login_returns_false_on_exception():
driver = base.AmamzonBase({})
driver.tab = FakeTab(raise_on_doc_loaded=True)
assert driver.login("secret") is False

1477
app/uv.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff