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

feat: patrol delete

Browse Source
This commit is contained in:
koko
2026-04-27 00:23:44 +08:00
parent a04c2f9a19
commit e1543416e3
44 changed files with 4489 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
app/amazon/base.py
View File

@@ -1,11 +1,11 @@
import json
import logging
import os
import subprocess
import time
import uuid
from typing import Any, Dict, List, Literal, Optional, TypedDict
from loguru import logger
import requests
from DrissionPage import Chromium
from DrissionPage._pages.chromium_tab import ChromiumTab
@@ -17,8 +17,6 @@ except ImportError:
winreg = None
logger = logging.getLogger(__name__)
STATUS_OK = "0"
STATUS_LOGIN_FAILED = "-10003"
@@ -76,7 +74,7 @@ class UserInfo(TypedDict):
def kill_process(version: Literal["v5", "v6"]):
"""结束指定版本的紫鸟客户端进程."""
logger.info("准备杀紫鸟客户端进程,version=%s", version)
logger.info("准备杀紫鸟客户端进程,version={}", version)
driver = ZiniaoDriver({})
driver.kill_process(version)
@@ -171,7 +169,7 @@ class ZiniaoDriver:
(winreg.HKEY_LOCAL_MACHINE, "HKEY_LOCAL_MACHINE"),
):
try:
logger.info("正在从注册表读取紫鸟客户端路径:%s\\%s", root_name, key_path)
logger.info("正在从注册表读取紫鸟客户端路径:{}\\{}", root_name, key_path)
key = winreg.OpenKey(root_key, key_path)
command, _ = winreg.QueryValueEx(key, "")
winreg.CloseKey(key)
@@ -180,12 +178,12 @@ class ZiniaoDriver:
exe_path = command[0 : command.find(sub) + len(sub) + 1]
else:
exe_path = command[0]
logger.info("已获取紫鸟客户端路径:%s", exe_path)
logger.info("已获取紫鸟客户端路径:{}", exe_path)
return exe_path
except FileNotFoundError:
logger.warning("注册表中未找到紫鸟客户端路径:%s\\%s", root_name, key_path)
logger.warning("注册表中未找到紫鸟客户端路径:{}\\{}", root_name, key_path)
logger.error("未能获取紫鸟客户端路径,protocol_name=%s", protocol_name)
logger.error("未能获取紫鸟客户端路径,protocol_name={}", protocol_name)
return None
def update_core(self):
@@ -199,13 +197,13 @@ class ZiniaoDriver:
logger.info("开始更新紫鸟浏览器内核")
while True:
result = self._post_client(payload)
logger.info("更新内核返回:%s", result)
logger.info("更新内核返回:{}", result)
if self._handle_update_core_result(result):
return
if result is None:
continue
logger.info("等待更新内核完成:%s", self._result_text(result))
logger.info("等待更新内核完成:{}", self._result_text(result))
time.sleep(UPDATE_CORE_RETRY_DELAY)
def _handle_update_core_result(self, result: Optional[Dict[str, Any]]) -> bool:
@@ -232,14 +230,35 @@ class ZiniaoDriver:
if version == "v5":
process_name = "SuperBrowser.exe"
logger.info("准备结束紫鸟 v5 starter.exe")
os.system("taskkill /f /t /im starter.exe")
self._kill_process_by_name("starter.exe")
else:
process_name = "ziniao.exe"
logger.info("准备结束紫鸟客户端进程:%s", process_name)
os.system("taskkill /f /t /im " + process_name)
logger.info("准备结束紫鸟客户端进程:{}", process_name)
self._kill_process_by_name(process_name)
time.sleep(PROCESS_KILL_DELAY)
def _kill_process_by_name(self, process_name: str) -> None:
result = subprocess.run(
["taskkill", "/f", "/t", "/im", process_name],
capture_output=True,
text=True,
encoding="utf-8",
errors="ignore",
check=False,
)
output = (result.stdout or result.stderr or "").strip()
if result.returncode == 0:
if output:
logger.info("结束进程返回:{}", output)
return
if "没有找到进程" in output or "not found" in output.lower():
logger.info("进程未运行,跳过结束:{}", process_name)
return
logger.warning("结束进程失败:{} | {}", process_name, output or f"returncode={result.returncode}")
def get_browser_list(self) -> Optional[list[Dict[str, Any]]]:
"""获取紫鸟浏览器店铺列表.
@@ -249,18 +268,18 @@ class ZiniaoDriver:
payload = self._build_payload("getBrowserList")
request_id = payload["requestId"]
logger.info("开始获取紫鸟浏览器列表,requestId=%s", request_id)
logger.info("开始获取紫鸟浏览器列表,requestId={}", 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("获取紫鸟浏览器列表成功,数量={}", len(browser_list or []))
return browser_list
if status_code == STATUS_LOGIN_FAILED:
logger.error("获取紫鸟浏览器列表登录失败:%s", self._result_text(result))
logger.error("获取紫鸟浏览器列表登录失败:{}", self._result_text(result))
return None
logger.error("获取紫鸟浏览器列表失败:%s", self._result_text(result))
logger.error("获取紫鸟浏览器列表失败:{}", self._result_text(result))
return None
def open_store(
@@ -295,17 +314,17 @@ class ZiniaoDriver:
)
request_id = payload["requestId"]
logger.info("开始打开紫鸟店铺,store_info=%s,requestId=%s", store_info, request_id)
logger.info("开始打开紫鸟店铺,store_info={},requestId={}", 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"))
logger.info("紫鸟店铺打开成功,debuggingPort={}", result.get("debuggingPort"))
return result
if status_code == STATUS_LOGIN_FAILED:
logger.error("打开紫鸟店铺登录失败:%s", self._result_text(result))
logger.error("打开紫鸟店铺登录失败:{}", self._result_text(result))
raise RuntimeError(f"[open_store]登录失败 {self._result_text(result)}")
logger.error("打开紫鸟店铺失败:%s", self._result_text(result))
logger.error("打开紫鸟店铺失败:{}", self._result_text(result))
raise RuntimeError(f"[open_store]失败 {self._result_text(result)} ")
def _build_start_browser_payload(
@@ -352,18 +371,18 @@ class ZiniaoDriver:
Returns:
DrissionPage Chromium 实例.
"""
logger.info("开始连接 DrissionPage 浏览器,port=%s", port)
logger.info("开始连接 DrissionPage 浏览器,port={}", port)
browser = Chromium(port)
logger.info("DrissionPage 浏览器连接完成,port=%s", port)
logger.info("DrissionPage 浏览器连接完成,port={}", port)
return browser
def start_client(self):
"""启动紫鸟客户端并更新浏览器内核."""
if self._is_client_ready():
logger.info("端口 %s 已启动,跳过启动紫鸟客户端", self.socket_port)
logger.info("端口 {} 已启动,跳过启动紫鸟客户端", self.socket_port)
return
logger.info("端口 %s 未启动,开始启动紫鸟客户端", self.socket_port)
logger.info("端口 {} 未启动,开始启动紫鸟客户端", self.socket_port)
self.kill_process("v6")
time.sleep(CLIENT_RESTART_DELAY)
client_path = self.get_zinaio_exe("superbrowserv6")
@@ -377,21 +396,21 @@ class ZiniaoDriver:
"--ipc_type=http",
"--port=" + str(self.socket_port),
]
logger.info("紫鸟客户端启动命令:%s", " ".join(cmd))
logger.info("紫鸟客户端启动命令:{}", " ".join(cmd))
for retry_count in range(CLIENT_START_RETRIES):
logger.info("第 %s/%s 次尝试启动紫鸟客户端", retry_count + 1, CLIENT_START_RETRIES)
logger.info("第 {}/{} 次尝试启动紫鸟客户端", retry_count + 1, CLIENT_START_RETRIES)
subprocess.Popen(cmd)
if self._wait_until_client_ready(CLIENT_READY_TIMEOUT):
logger.info("紫鸟客户端启动成功,第 %s 次尝试", retry_count + 1)
logger.info("紫鸟客户端启动成功,第 {} 次尝试", retry_count + 1)
time.sleep(CLIENT_POST_START_DELAY)
self.update_core()
return
logger.warning("第 %s 次尝试启动失败,10秒内未检测到客户端启动", retry_count + 1)
logger.warning("第 {} 次尝试启动失败,10秒内未检测到客户端启动", retry_count + 1)
logger.error("紫鸟客户端启动失败,已重试 %s 次", CLIENT_START_RETRIES)
logger.error("紫鸟客户端启动失败,已重试 {} 次", CLIENT_START_RETRIES)
raise RuntimeError(f"客户端启动失败:重试 {CLIENT_START_RETRIES} 次后仍未成功启动")
def open_shop(self, shop_name: str):
@@ -403,12 +422,12 @@ class ZiniaoDriver:
Returns:
成功时返回浏览器实例.店铺不存在时返回错误信息.
"""
logger.info("开始打开店铺:%s", shop_name)
logger.info("开始打开店铺:{}", shop_name)
self.start_client()
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("店铺不存在:{}", shop_name)
return "店铺不存在"
ret_json = self.open_store(self.store_id)
@@ -419,16 +438,16 @@ class ZiniaoDriver:
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检测页地址为空,准备关闭店铺:{}", shop_name)
self.close_store(self.store_id)
raise RuntimeError("没有IP检测地址,为了店铺安全不打开店铺")
if self.open_ip_check(self.browser, ip_check_url):
logger.info("IP检测通过,打开店铺平台主页:%s", shop_name)
logger.info("IP检测通过,打开店铺平台主页:{}", shop_name)
self.open_launcher_page(ret_json.get("launcherPage"), self.browser)
return self.browser
logger.error("IP检测不通过,停止打开店铺:%s", shop_name)
logger.error("IP检测不通过,停止打开店铺:{}", shop_name)
raise RuntimeError("IP检测不通过,可能是因为网络环境变化导致的,为了店铺安全不打开店铺")
def close_store(self, browser_oauth=None):
@@ -452,17 +471,17 @@ class ZiniaoDriver:
)
request_id = payload["requestId"]
logger.info("开始关闭紫鸟店铺,browserOauth=%s,requestId=%s", browser_oauth, request_id)
logger.info("开始关闭紫鸟店铺,browserOauth={},requestId={}", 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)
logger.info("紫鸟店铺关闭成功,browserOauth={}", browser_oauth)
return result
if status_code == STATUS_LOGIN_FAILED:
logger.error("关闭紫鸟店铺登录失败:%s", self._result_text(result))
raise RuntimeError(f"[close_store]登录失败 {self._result_text(result)}")
logger.error("关闭紫鸟店铺登录失败:{}", self._result_text(result))
raise RuntimeError(f"【close_store】登录失败 {self._result_text(result)}")
logger.error("关闭紫鸟店铺失败:%s", self._result_text(result))
logger.error("关闭紫鸟店铺失败:{}", self._result_text(result))
raise RuntimeError(f"[close_store]失败: {self._result_text(result)} ")
def open_launcher_page(self, launcher_page: str, browser: Chromium = None):
@@ -475,7 +494,7 @@ class ZiniaoDriver:
if browser is None:
browser = self.browser
logger.info("打开店铺平台启动页:%s", launcher_page)
logger.info("打开店铺平台启动页:{}", launcher_page)
tab = browser.new_tab(url=launcher_page)
self.tab = tab
return tab
@@ -491,7 +510,7 @@ class ZiniaoDriver:
IP 检测通过时返回 True,否则返回 False.
"""
try:
logger.info("开始打开IP检测页:%s", ip_check_url)
logger.info("开始打开IP检测页:{}", ip_check_url)
tab = browser.latest_tab
tab.get(ip_check_url)
success_button = tab.ele(
@@ -512,7 +531,7 @@ class ZiniaoDriver:
class AmamzonBase(ZiniaoDriver):
"""亚马逊页面操作基类."""
def SwitchingCountries(self, country_name: str):
def switch_to_country(self, country_name: str):
"""切换亚马逊账号国家并验证结果.
流程包括读取当前国家,打开国家下拉框,选择目标国家并等待页面加载.
@@ -534,7 +553,7 @@ class AmamzonBase(ZiniaoDriver):
logger.warning("无法获取当前国家信息")
return False
if current_country == country_name:
logger.info("当前已经是目标国家 %s,无需切换", country_name)
logger.info("当前已经是目标国家 {},无需切换", country_name)
return True
if not self._open_country_dropdown():
@@ -557,7 +576,7 @@ class AmamzonBase(ZiniaoDriver):
return None
current_country = current_country_ele.text.strip()
logger.info("当前国家:%s", current_country)
logger.info("当前国家:{}", current_country)
return current_country
def _open_country_dropdown(self) -> bool:
@@ -579,10 +598,10 @@ class AmamzonBase(ZiniaoDriver):
return True
def _select_country(self, country_name: str) -> bool:
logger.info("正在切换到国家:%s", country_name)
logger.info("正在切换到国家:{}", 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)
logger.warning("找不到目标国家:{}", country_name)
return False
target_country.click()
return True
@@ -601,10 +620,10 @@ class AmamzonBase(ZiniaoDriver):
logger.warning("无法验证国家切换结果")
return False
if new_country == country_name:
logger.info("国家切换成功:%s", new_country)
logger.info("国家切换成功:{}", new_country)
return True
logger.warning("国家切换失败,当前国家:%s,目标国家:%s", new_country, country_name)
logger.warning("国家切换失败,当前国家:{},目标国家:{}", new_country, country_name)
return False
def need_login(self):
@@ -657,7 +676,7 @@ class AmamzonBase(ZiniaoDriver):
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))
logger.info("发送一次性密码元素数量:{}", len(send_code))
if len(send_code) == 0:
return False
@@ -678,7 +697,7 @@ class AmamzonBase(ZiniaoDriver):
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))
logger.info("验证码输入框数量:{}", len(otp_code_input))
if len(otp_code_input) == 0:
return None
@@ -706,7 +725,7 @@ class AmamzonBase(ZiniaoDriver):
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)
logger.warning("验证码输入错误:{}", error_mes[0].text)
self.tab.refresh()
return False
@@ -718,3 +737,4 @@ class AmamzonBase(ZiniaoDriver):
if len(submit_btn) > 0:
logger.info("点击二次登录提交按钮")
submit_btn[0].click()