diff --git a/app/.rtk/filters.toml b/app/.rtk/filters.toml new file mode 100644 index 0000000..d9bd43f --- /dev/null +++ b/app/.rtk/filters.toml @@ -0,0 +1,13 @@ +# Project-local RTK filters — commit this file with your repo. +# Filters here override user-global and built-in filters. +# Docs: https://github.com/rtk-ai/rtk#custom-filters +schema_version = 1 + +# Example: suppress build noise from a custom tool +# [filters.my-tool] +# description = "Compact my-tool output" +# match_command = "^my-tool\\s+build" +# strip_ansi = true +# strip_lines_matching = ["^\\s*$", "^Downloading", "^Installing"] +# max_lines = 30 +# on_empty = "my-tool: ok" diff --git a/app/amazon/base.py b/app/amazon/base.py index 6dddbe2..60327db 100644 --- a/app/amazon/base.py +++ b/app/amazon/base.py @@ -3,9 +3,8 @@ import logging import os import subprocess import time -import traceback import uuid -from typing import Literal +from typing import Any, Dict, List, Literal, Optional, TypedDict import requests from DrissionPage import Chromium @@ -20,44 +19,150 @@ except ImportError: logger = logging.getLogger(__name__) +STATUS_OK = "0" +STATUS_LOGIN_FAILED = "-10003" + +DEFAULT_SOCKET_PORT = 19890 +CLIENT_API_TIMEOUT = 120 +PORT_CHECK_TIMEOUT = 2 +UPDATE_CORE_RETRY_DELAY = 2 +CLIENT_RESTART_DELAY = 5 +CLIENT_START_RETRIES = 3 +CLIENT_READY_TIMEOUT = 10 +CLIENT_READY_INTERVAL = 0.5 +CLIENT_POST_START_DELAY = 5 +PROCESS_KILL_DELAY = 3 + +DOC_LOAD_TIMEOUT = 30 +COUNTRY_INITIAL_LOAD_TIMEOUT = 120 +COUNTRY_LOOKUP_TIMEOUT = 20 +COUNTRY_CLICK_TIMEOUT = 10 +COUNTRY_DROPDOWN_DELAY = 1 +IP_CHECK_TIMEOUT = 60 + +LOGIN_ATTEMPTS = 4 +PASSWORD_INPUT_TIMEOUT = 5 +PASSWORD_SUBMIT_TIMEOUT = 5 +OTP_SEND_TIMEOUT = 5 +OTP_INPUT_LOOKUP_ATTEMPTS = 5 +OTP_INPUT_TIMEOUT = 30 +OTP_INPUT_DISPLAY_TIMEOUT = 10 +OTP_CODE_WAIT_SECONDS = 30 +OTP_SUBMIT_TIMEOUT = 10 +OTP_RESULT_LOAD_TIMEOUT = 20 +OTP_ERROR_TIMEOUT = 10 +OTP_SEND_DELAY = 1 +LOGIN_FINAL_SUBMIT_TIMEOUT = 10 +NEED_LOGIN_DELAY = 3 +NEED_LOGIN_TIMEOUT = 5 + +COUNTRY_LABEL_XPATH = 'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]' +COUNTRY_DROPDOWN_XPATH = 'xpath://div[@class="dropdown-account-switcher-header-label"]' +COUNTRY_LIST_ITEM_XPATH = 'xpath://div[@class="dropdown-account-switcher-list-item"]' +NEED_LOGIN_XPATH = 'xpath://h1[@class="a-spacing-small"]|//span[contains(text(),"登录")]' +PASSWORD_INPUT_XPATH = 'xpath://input[@type="password"]' +PASSWORD_SUBMIT_XPATH = 'xpath://input[@id="signInSubmit"]' +OTP_SEND_XPATH = 'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]' +OTP_INPUT_XPATH = 'xpath://input[@name="otpCode"]' +OTP_SUBMIT_XPATH = 'xpath://input[@id="auth-signin-button"]' +OTP_ERROR_XPATH = 'xpath://div[@id="auth-error-message-box"]' + + +class UserInfo(TypedDict): + username: str + password: str + company: str + def kill_process(version: Literal["v5", "v6"]): - """杀紫鸟客户端进程（独立函数版本）""" - logger.info("准备杀紫鸟客户端进程，version=%s", version) + """结束指定版本的紫鸟客户端进程.""" + logger.info("准备杀紫鸟客户端进程,version=%s", version) driver = ZiniaoDriver({}) driver.kill_process(version) class ZiniaoDriver: - """紫鸟浏览器自动化驱动类""" + """封装紫鸟客户端启动,店铺浏览器生命周期和客户端 API 调用.""" - def __init__(self, user_info: dict, socket_port: int = 19890): - """ - 初始化紫鸟浏览器驱动 + def __init__(self, user_info: UserInfo, socket_port: int = DEFAULT_SOCKET_PORT): + """初始化紫鸟浏览器驱动. Args: - user_info: 用户信息字典，包含 company, username, password - socket_port: 客户端通信端口，默认 19890 + user_info: 紫鸟账号信息,包含 company, username, password. + socket_port: 客户端 HTTP 通信端口. """ self.user_info = user_info self.socket_port = socket_port self.client_path = None self.browser = None - self.tab: ChromiumTab = None + self.tab: Optional[ChromiumTab] = None self.store_id = None + @property + def client_url(self) -> str: + return f"http://127.0.0.1:{self.socket_port}" + + def _build_payload(self, action: str, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + payload = { + "action": action, + "requestId": str(uuid.uuid4()), + } + if extra: + payload.update(extra) + payload.update(self.user_info) + return payload + + def _post_client(self, payload: Dict[str, Any]) -> Dict[str, Any]: + response = requests.post( + self.client_url, + json.dumps(payload).encode("utf-8"), + timeout=CLIENT_API_TIMEOUT, + ) + return response.json() + + @staticmethod + def _status_code(result: Optional[Dict[str, Any]]) -> Optional[str]: + if result is None: + return None + return str(result.get("statusCode")) + + @staticmethod + def _result_text(result: Dict[str, Any]) -> str: + return json.dumps(result, ensure_ascii=False) + + @staticmethod + def _find_store_oauth(shop_list: Optional[List[Dict[str, Any]]], shop_name: str): + for shop in shop_list or []: + if shop.get("browserName") == shop_name: + return shop.get("browserOauth") + return None + + def _is_client_ready(self) -> bool: + try: + requests.get(self.client_url, timeout=PORT_CHECK_TIMEOUT) + return True + except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): + return False + + def _wait_until_client_ready(self, timeout: float) -> bool: + start_check_time = time.time() + while time.time() - start_check_time < timeout: + if self._is_client_ready(): + return True + time.sleep(CLIENT_READY_INTERVAL) + return False + def get_zinaio_exe(self, protocol_name: str = "superbrowser"): - """ - 获取紫鸟安装目录 + """从 Windows 注册表读取紫鸟客户端可执行文件路径. Args: - protocol_name: 协议名称，默认 "superbrowser" + protocol_name: 注册表协议名称. Returns: - exe_path: 紫鸟浏览器可执行文件路径 + 紫鸟浏览器可执行文件路径. 未找到时返回 None. """ if winreg is None: - logger.error("当前系统不支持 winreg，无法从注册表获取紫鸟客户端路径") + logger.error("当前系统不支持 winreg,无法从注册表获取紫鸟客户端路径") return None key_path = rf"SOFTWARE\Classes\{protocol_name}\shell\open\command" @@ -66,60 +171,63 @@ class ZiniaoDriver: (winreg.HKEY_LOCAL_MACHINE, "HKEY_LOCAL_MACHINE"), ): try: - logger.info("正在从注册表读取紫鸟客户端路径：%s\\%s", root_name, key_path) + 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] + exe_path = command[0 : command.find(sub) + len(sub) + 1] else: exe_path = command[0] - logger.info("已获取紫鸟客户端路径：%s", exe_path) + logger.info("已获取紫鸟客户端路径:%s", exe_path) return exe_path except FileNotFoundError: - logger.warning("注册表中未找到紫鸟客户端路径：%s\\%s", root_name, key_path) + logger.warning("注册表中未找到紫鸟客户端路径:%s\\%s", root_name, key_path) - logger.error("未能获取紫鸟客户端路径，protocol_name=%s", protocol_name) + logger.error("未能获取紫鸟客户端路径,protocol_name=%s", protocol_name) return None def update_core(self): + """更新紫鸟浏览器内核. + + 打开店铺前调用,要求客户端版本 5.285.7 以上. 接口可能因 HTTP 超时 + 返回未完成状态,因此会循环调用直到成功或判定客户端不支持. """ - 下载所有内核，打开店铺前调用，需客户端版本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}" + payload = self._build_payload("updateCore") logger.info("开始更新紫鸟浏览器内核") while True: - response = requests.post(url, json.dumps(data).encode("utf-8"), timeout=120) - result = response.json() - logger.info("更新内核返回：%s", result) + result = self._post_client(payload) + logger.info("更新内核返回:%s", result) + if self._handle_update_core_result(result): + return 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) + logger.info("等待更新内核完成:%s", self._result_text(result)) + time.sleep(UPDATE_CORE_RETRY_DELAY) + + def _handle_update_core_result(self, result: Optional[Dict[str, Any]]) -> bool: + if result is None: + logger.info("等待客户端启动后继续更新内核") + time.sleep(UPDATE_CORE_RETRY_DELAY) + return False + + status_code = self._status_code(result) + if status_code is None or status_code == STATUS_LOGIN_FAILED: + logger.error("当前紫鸟客户端版本不支持更新内核接口,请升级客户端") + return True + if status_code == STATUS_OK: + logger.info("紫鸟浏览器内核更新完成") + return True + return False def kill_process(self, version: Literal["v5", "v6"]): - """ - 杀紫鸟客户端进程 + """结束指定版本的紫鸟客户端进程. Args: - version: 客户端版本 + version: 客户端主版本. """ if version == "v5": process_name = "SuperBrowser.exe" @@ -128,37 +236,31 @@ class ZiniaoDriver: else: process_name = "ziniao.exe" - logger.info("准备结束紫鸟客户端进程：%s", process_name) + logger.info("准备结束紫鸟客户端进程:%s", process_name) os.system("taskkill /f /t /im " + process_name) - time.sleep(3) + time.sleep(PROCESS_KILL_DELAY) - def get_browser_list(self) -> list: - """ - 获取浏览器列表 + def get_browser_list(self) -> Optional[list[Dict[str, Any]]]: + """获取紫鸟浏览器店铺列表. Returns: - list: 浏览器列表 + 浏览器店铺列表.登录失败或请求失败时返回 None. """ - request_id = str(uuid.uuid4()) - data = { - "action": "getBrowserList", - "requestId": request_id, - } - data.update(self.user_info) + payload = self._build_payload("getBrowserList") + request_id = payload["requestId"] - 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": + logger.info("开始获取紫鸟浏览器列表,requestId=%s", request_id) + result = self._post_client(payload) + status_code = self._status_code(result) + if status_code == STATUS_OK: browser_list = result.get("browserList") - logger.info("获取紫鸟浏览器列表成功，数量=%s", len(browser_list or [])) + 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)) + if status_code == STATUS_LOGIN_FAILED: + logger.error("获取紫鸟浏览器列表登录失败:%s", self._result_text(result)) return None - logger.error("获取紫鸟浏览器列表失败：%s", json.dumps(result, ensure_ascii=False)) + logger.error("获取紫鸟浏览器列表失败:%s", self._result_text(result)) return None def open_store( @@ -170,89 +272,103 @@ class ZiniaoDriver: cookieTypeSave=0, jsInfo="", ): - """ - 打开店铺 + """启动紫鸟店铺浏览器. Args: - store_info: 店铺信息（browserId 或 browserOauth） - isWebDriverReadOnlyMode: 是否只读模式，默认 0 - isprivacy: 隐私模式，默认 0 - isHeadless: 无头模式，默认 0 - cookieTypeSave: cookie保存类型，默认 0 - jsInfo: 注入的JS信息，默认 "" + store_info: 店铺 browserId 或 browserOauth. + isWebDriverReadOnlyMode: WebDriver 只读模式开关. + isprivacy: 隐私模式开关. + isHeadless: 无头模式开关. + cookieTypeSave: Cookie 保存类型. + 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) + payload = self._build_start_browser_payload( + store_info=store_info, + is_web_driver_read_only_mode=isWebDriverReadOnlyMode, + is_privacy=isprivacy, + is_headless=isHeadless, + cookie_type_save=cookieTypeSave, + js_info=jsInfo, + ) + request_id = payload["requestId"] + + logger.info("开始打开紫鸟店铺,store_info=%s,requestId=%s", store_info, request_id) + result = self._post_client(payload) + status_code = self._status_code(result) + if status_code == STATUS_OK: + logger.info("紫鸟店铺打开成功,debuggingPort=%s", result.get("debuggingPort")) + return result + if status_code == STATUS_LOGIN_FAILED: + logger.error("打开紫鸟店铺登录失败:%s", self._result_text(result)) + raise RuntimeError(f"[open_store]登录失败 {self._result_text(result)}") + + logger.error("打开紫鸟店铺失败:%s", self._result_text(result)) + raise RuntimeError(f"[open_store]失败 {self._result_text(result)} ") + + def _build_start_browser_payload( + self, + *, + store_info, + is_web_driver_read_only_mode=0, + is_privacy=0, + is_headless=0, + cookie_type_save=0, + js_info="", + ) -> Dict[str, Any]: + payload = self._build_payload( + "startBrowser", + { + "isWaitPluginUpdate": 0, + "isHeadless": is_headless, + "isWebDriverReadOnlyMode": is_web_driver_read_only_mode, + "cookieTypeLoad": 0, + "cookieTypeSave": cookie_type_save, + "runMode": "1", + "isLoadUserPlugin": True, + "pluginIdType": 1, + "privacyMode": is_privacy, + }, + ) if store_info.isdigit(): - data["browserId"] = store_info + payload["browserId"] = store_info else: - data["browserOauth"] = store_info + payload["browserOauth"] = store_info - if len(str(jsInfo)) > 2: - data["injectJsInfo"] = json.dumps(jsInfo) + if len(str(js_info)) > 2: + payload["injectJsInfo"] = json.dumps(js_info) - 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)} ") + return payload def get_browser(self, port) -> Chromium: - """ - 获取浏览器实例 + """连接 DrissionPage 浏览器实例. Args: - port: 调试端口 + port: Chromium 调试端口. Returns: - Chromium: DrissionPage浏览器实例 + DrissionPage Chromium 实例. """ - logger.info("开始连接 DrissionPage 浏览器，port=%s", port) + logger.info("开始连接 DrissionPage 浏览器,port=%s", port) browser = Chromium(port) - logger.info("DrissionPage 浏览器连接完成，port=%s", 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) + """启动紫鸟客户端并更新浏览器内核.""" + if self._is_client_ready(): + logger.info("端口 %s 已启动,跳过启动紫鸟客户端", self.socket_port) return - except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): - logger.info("端口 %s 未启动，开始启动紫鸟客户端", self.socket_port) + logger.info("端口 %s 未启动,开始启动紫鸟客户端", self.socket_port) self.kill_process("v6") - time.sleep(5) + time.sleep(CLIENT_RESTART_DELAY) client_path = self.get_zinaio_exe("superbrowserv6") if not client_path: - raise RuntimeError("未找到紫鸟客户端路径，无法启动客户端") + raise RuntimeError("未找到紫鸟客户端路径,无法启动客户端") self.client_path = client_path.strip('"') cmd = [ @@ -261,56 +377,38 @@ class ZiniaoDriver: "--ipc_type=http", "--port=" + str(self.socket_port), ] - logger.info("紫鸟客户端启动命令：%s", " ".join(cmd)) + logger.info("紫鸟客户端启动命令:%s", " ".join(cmd)) - max_retries = 3 - for retry_count in range(max_retries): - logger.info("第 %s/%s 次尝试启动紫鸟客户端", retry_count + 1, max_retries) + for retry_count in range(CLIENT_START_RETRIES): + logger.info("第 %s/%s 次尝试启动紫鸟客户端", retry_count + 1, CLIENT_START_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) + if self._wait_until_client_ready(CLIENT_READY_TIMEOUT): + logger.info("紫鸟客户端启动成功,第 %s 次尝试", retry_count + 1) + time.sleep(CLIENT_POST_START_DELAY) self.update_core() return - logger.warning("第 %s 次尝试启动失败，10秒内未检测到客户端启动", retry_count + 1) + logger.warning("第 %s 次尝试启动失败,10秒内未检测到客户端启动", retry_count + 1) - logger.error("紫鸟客户端启动失败，已重试 %s 次", max_retries) - raise RuntimeError(f"客户端启动失败：重试 {max_retries} 次后仍未成功启动") + logger.error("紫鸟客户端启动失败,已重试 %s 次", CLIENT_START_RETRIES) + raise RuntimeError(f"客户端启动失败:重试 {CLIENT_START_RETRIES} 次后仍未成功启动") def open_shop(self, shop_name: str): - """ - 打开指定店铺 + """打开指定店铺并完成 IP 检测. Args: - shop_name: 店铺名称 + shop_name: 店铺名称. Returns: - Chromium or str: 成功返回浏览器实例，失败返回错误信息 + 成功时返回浏览器实例.店铺不存在时返回错误信息. """ - logger.info("开始打开店铺：%s", shop_name) + 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 - + self.store_id = self._find_store_oauth(self.get_browser_list(), shop_name) if not self.store_id: - logger.warning("店铺不存在：%s", shop_name) + logger.warning("店铺不存在:%s", shop_name) return "店铺不存在" ret_json = self.open_store(self.store_id) @@ -319,88 +417,86 @@ class ZiniaoDriver: 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) + logger.error("ip检测页地址为空,准备关闭店铺:%s", shop_name) self.close_store(self.store_id) - raise RuntimeError("没有IP检测地址，为了店铺安全不打开店铺") + raise RuntimeError("没有IP检测地址,为了店铺安全不打开店铺") - ip_usable = self.open_ip_check(self.browser, ip_check_url) - if ip_usable: - logger.info("IP检测通过，打开店铺平台主页：%s", shop_name) + if self.open_ip_check(self.browser, ip_check_url): + 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 + return self.browser + + logger.error("IP检测不通过,停止打开店铺:%s", shop_name) + raise RuntimeError("IP检测不通过,可能是因为网络环境变化导致的,为了店铺安全不打开店铺") def close_store(self, browser_oauth=None): - """ - 关闭店铺 + """关闭紫鸟店铺浏览器. Args: - browser_oauth: 店铺OAuth标识，如果不提供则使用当前打开的店铺 + 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) + payload = self._build_payload( + "stopBrowser", + { + "duplicate": 0, + "browserOauth": browser_oauth, + }, + ) + request_id = payload["requestId"] - 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) + logger.info("开始关闭紫鸟店铺,browserOauth=%s,requestId=%s", browser_oauth, request_id) + result = self._post_client(payload) + status_code = self._status_code(result) + if status_code == STATUS_OK: + 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)}") + if status_code == STATUS_LOGIN_FAILED: + logger.error("关闭紫鸟店铺登录失败:%s", self._result_text(result)) + raise RuntimeError(f"[close_store]登录失败 {self._result_text(result)}") - logger.error("关闭紫鸟店铺失败：%s", json.dumps(result, ensure_ascii=False)) - raise RuntimeError(f"【close_store】失败： {json.dumps(result, ensure_ascii=False)} ") + logger.error("关闭紫鸟店铺失败:%s", self._result_text(result)) + raise RuntimeError(f"[close_store]失败: {self._result_text(result)} ") def open_launcher_page(self, launcher_page: str, browser: Chromium = None): - """ - 打开启动页面 + """打开店铺平台启动页. Args: - launcher_page: 要打开的页面URL - browser: 浏览器实例，如果不提供则使用当前浏览器实例 + launcher_page: 要打开的启动页 URL. + browser: 浏览器实例. 未提供时使用当前浏览器实例. """ if browser is None: browser = self.browser - logger.info("打开店铺平台启动页：%s", launcher_page) + 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 检测结果 + """打开 IP 检测页并判断网络环境是否通过. + + Args: + browser: DrissionPage 浏览器会话. + ip_check_url: IP 检测页地址. + + Returns: + IP 检测通过时返回 True,否则返回 False. """ try: - logger.info("开始打开IP检测页：%s", ip_check_url) + 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, + timeout=IP_CHECK_TIMEOUT, ) if success_button: logger.info("IP检测成功") @@ -414,105 +510,108 @@ class ZiniaoDriver: 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: 目标国家名称 + country_name: 目标国家名称. Returns: - bool: 切换成功返回True，失败返回False + 切换成功返回 True,否则返回 False. """ try: if self.browser is None: - logger.warning("浏览器实例不存在，请先打开店铺") + 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: + self.tab.wait.doc_loaded(timeout=COUNTRY_INITIAL_LOAD_TIMEOUT, raise_err=False) + current_country = self._get_current_country(timeout=COUNTRY_LOOKUP_TIMEOUT) + if current_country is None: logger.warning("无法获取当前国家信息") return False + if current_country == country_name: + logger.info("当前已经是目标国家 %s,无需切换", country_name) + return True - logger.info("正在打开国家切换下拉框") - dropdown_header = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]', timeout=10) - if not dropdown_header: - logger.warning("找不到国家切换下拉框") + if not self._open_country_dropdown(): 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("找不到国家列表项") + if not self._select_country(country_name): 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 + return self._verify_country(country_name) 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, + def _get_current_country(self, *, timeout: int) -> Optional[str]: + logger.info("正在检查当前国家") + current_country_ele = self.tab.ele(COUNTRY_LABEL_XPATH, timeout=timeout) + if not current_country_ele: + return None + + current_country = current_country_ele.text.strip() + logger.info("当前国家:%s", current_country) + return current_country + + def _open_country_dropdown(self) -> bool: + logger.info("正在打开国家切换下拉框") + dropdown_header = self.tab.ele(COUNTRY_DROPDOWN_XPATH, timeout=COUNTRY_CLICK_TIMEOUT) + if not dropdown_header: + logger.warning("找不到国家切换下拉框") + return False + dropdown_header.click() + time.sleep(COUNTRY_DROPDOWN_DELAY) + + logger.info("正在展开国家列表") + first_item = self.tab.ele(COUNTRY_LIST_ITEM_XPATH, timeout=COUNTRY_CLICK_TIMEOUT) + if not first_item: + logger.warning("找不到国家列表项") + return False + first_item.click() + time.sleep(COUNTRY_DROPDOWN_DELAY) + return True + + def _select_country(self, country_name: str) -> bool: + logger.info("正在切换到国家:%s", country_name) + target_country = self.tab.ele(self._country_option_xpath(country_name), timeout=COUNTRY_CLICK_TIMEOUT) + if not target_country: + logger.warning("找不到目标国家:%s", country_name) + return False + target_country.click() + return True + + @staticmethod + def _country_option_xpath(country_name: str) -> str: + return ( + "xpath://div[@class=" + '"dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" ' + f'and @title="{country_name}"]' ) + + def _verify_country(self, country_name: str) -> bool: + new_country = self._get_current_country(timeout=COUNTRY_CLICK_TIMEOUT) + if new_country is None: + logger.warning("无法验证国家切换结果") + return False + if new_country == country_name: + logger.info("国家切换成功:%s", new_country) + return True + + logger.warning("国家切换失败,当前国家:%s,目标国家:%s", new_country, country_name) + return False + + def need_login(self): + """判断当前页面是否需要登录.""" + time.sleep(NEED_LOGIN_DELAY) + self.tab.wait.doc_loaded(timeout=DOC_LOAD_TIMEOUT, raise_err=False) + need_login_ele = self.tab.eles(NEED_LOGIN_XPATH, timeout=NEED_LOGIN_TIMEOUT) if len(need_login_ele) > 0: logger.info("检测到需要登录元素") return True @@ -520,61 +619,102 @@ class AmamzonBase(ZiniaoDriver): return False def login(self, password, username=""): + """登录当前页面,必要时等待并提交一次性验证码. + + Args: + password: 登录密码. + username: 兼容保留参数,当前未使用. + + Returns: + 登录成功返回 True,失败返回 False. + """ 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) + self.tab.wait.doc_loaded(timeout=DOC_LOAD_TIMEOUT, raise_err=False) + for _ in range(LOGIN_ATTEMPTS): + self._submit_password_if_present(password) + self.tab.wait.doc_loaded(timeout=DOC_LOAD_TIMEOUT, 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 + if self._send_otp_if_present(): + continue + if self._submit_ready_otp(): + 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() + self._click_final_login_submit_if_present() except Exception: - logger.exception("登录过程中发生异常：%s", traceback.format_exc()) + logger.exception("登录过程中发生异常") return False + + def _submit_password_if_present(self, password) -> None: + pwd_input = self.tab.eles(PASSWORD_INPUT_XPATH, timeout=PASSWORD_INPUT_TIMEOUT) + if len(pwd_input) > 0: + logger.info("检测到密码输入框,准备输入密码") + pwd_input[0].input(password, clear=True) + + submit_btn = self.tab.eles(PASSWORD_SUBMIT_XPATH, timeout=PASSWORD_SUBMIT_TIMEOUT) + if len(submit_btn) > 0: + logger.info("点击登录提交按钮") + submit_btn[0].click() + + def _send_otp_if_present(self) -> bool: + send_code = self.tab.eles(OTP_SEND_XPATH, timeout=OTP_SEND_TIMEOUT) + logger.info("发送一次性密码元素数量:%s", len(send_code)) + if len(send_code) == 0: + return False + + logger.info("检测到发送一次性密码按钮,准备点击") + send_code[0].click() + time.sleep(OTP_SEND_DELAY) + self.tab.wait.doc_loaded(timeout=DOC_LOAD_TIMEOUT, raise_err=False) + return True + + def _submit_ready_otp(self) -> bool: + for _ in range(OTP_INPUT_LOOKUP_ATTEMPTS): + otp_input = self._find_otp_input() + if otp_input is None: + return False + if self._wait_for_otp_value(otp_input) and self._submit_otp_if_ready(): + return True + return False + + def _find_otp_input(self): + otp_code_input = self.tab.eles(OTP_INPUT_XPATH, timeout=OTP_INPUT_TIMEOUT) + logger.info("验证码输入框数量:%s", len(otp_code_input)) + if len(otp_code_input) == 0: + return None + + otp_code_input[0].wait.displayed(timeout=OTP_INPUT_DISPLAY_TIMEOUT, raise_err=False) + return otp_code_input[0] + + @staticmethod + def _has_otp_value(otp_input) -> bool: + return otp_input.value is not None and otp_input.value.strip() != "" + + def _wait_for_otp_value(self, otp_input) -> bool: + for _ in range(OTP_CODE_WAIT_SECONDS): + if self._has_otp_value(otp_input): + logger.info("检测到验证码输入完成") + return True + time.sleep(1) + return False + + def _submit_otp_if_ready(self) -> bool: + submit_btn = self.tab.eles(OTP_SUBMIT_XPATH, timeout=OTP_SUBMIT_TIMEOUT) + if len(submit_btn) == 0: + return False + + submit_btn[0].click() + self.tab.wait.doc_loaded(timeout=OTP_RESULT_LOAD_TIMEOUT, raise_err=False) + error_mes = self.tab.eles(OTP_ERROR_XPATH, timeout=OTP_ERROR_TIMEOUT) + if len(error_mes) > 0: + logger.warning("验证码输入错误:%s", error_mes[0].text) + self.tab.refresh() + return False + + logger.info("登录成功") + return True + + def _click_final_login_submit_if_present(self) -> None: + submit_btn = self.tab.eles(OTP_SUBMIT_XPATH, timeout=LOGIN_FINAL_SUBMIT_TIMEOUT) + if len(submit_btn) > 0: + logger.info("点击二次登录提交按钮") + submit_btn[0].click() diff --git a/app/amazon/product.py b/app/amazon/product.py new file mode 100644 index 0000000..21c0d5d --- /dev/null +++ b/app/amazon/product.py @@ -0,0 +1,11 @@ +from .base import AmamzonBase, UserInfo + +if __name__ == "__main__": + user_info: UserInfo = {"company": "rongchuang123", "username": "自动化_Robot", "password": "#20zsg25"} + driver = AmamzonBase(user_info) + + shop_name = "郭亚芳" + country = "德国" + + driver.open_shop(shop_name) + driver.SwitchingCountries(country) diff --git a/app/pyproject.toml b/app/pyproject.toml index 54fdfb8..b6fd8dc 100644 --- a/app/pyproject.toml +++ b/app/pyproject.toml @@ -91,6 +91,9 @@ dependencies = [ [tool.uv] package = false +[tool.pytest.ini_options] +pythonpath = ["."] + [dependency-groups] dev = [ "pytest>=8.4.2", diff --git a/app/tests/test_amazon_base.py b/app/tests/test_amazon_base.py index 45dec55..ce2383f 100644 --- a/app/tests/test_amazon_base.py +++ b/app/tests/test_amazon_base.py @@ -154,6 +154,33 @@ def test_get_browser_list_status_handling(monkeypatch, payload, expected): assert base.ZiniaoDriver({}).get_browser_list() == expected +def test_update_core_retries_none_then_returns_on_success(monkeypatch): + calls = [] + responses = [FakeResponse(None), FakeResponse({"statusCode": 0})] + + def fake_post(url, data, timeout): + calls.append(json.loads(data.decode("utf-8"))) + return responses.pop(0) + + monkeypatch.setattr(base.requests, "post", fake_post) + monkeypatch.setattr(base.time, "sleep", lambda seconds: None) + + base.ZiniaoDriver({}).update_core() + + assert [payload["action"] for payload in calls] == ["updateCore", "updateCore"] + + +def test_wait_until_client_ready_polls_until_available(monkeypatch): + readiness = [False, False, True] + sleeps = [] + driver = base.ZiniaoDriver({}) + monkeypatch.setattr(driver, "_is_client_ready", lambda: readiness.pop(0)) + monkeypatch.setattr(base.time, "sleep", lambda seconds: sleeps.append(seconds)) + + assert driver._wait_until_client_ready(timeout=10) is True + assert sleeps == [base.CLIENT_READY_INTERVAL, base.CLIENT_READY_INTERVAL] + + def test_close_store_success_uses_current_store(monkeypatch): payloads = []