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

new file: amazon/amazon_base.py

Browse Source 
modified:   amazon/approve.py
	modified:   amazon/asin_status.py
	new file:   amazon/chrome_base.py
	modified:   amazon/del_brand.py
	modified:   amazon/detail_spider.py
	new file:   "amazon/detail_spider_\345\244\207\344\273\275.py"
	modified:   amazon/main.py
	modified:   amazon/match_action.py
	modified:   amazon/price_match.py
	new file:   amazon/similar_asin.py
	modified:   app.py
	new file:   assets/appearance-patent-Br1dtmol.css
	new file:   assets/appearance-patent-D1DgeYhe.css
	new file:   assets/delete-brand-BfMLVSQU.css
	new file:   assets/patrol-delete-BAzbYtgc.css
	new file:   assets/price-track-Hozgt_em.css
	new file:   assets/product-risk-CbX7bwZi.css
	new file:   assets/query-asin-B4RsOiza.css
	new file:   assets/shop-match-B2HWAgBY.css
	modified:   main.py
This commit is contained in:
铭坤
2026-05-04 19:15:50 +08:00
parent 147b324658
commit 3ca569ed1e
13 changed files with 11073 additions and 2970 deletions
Download Patch File Download Diff File Expand all files Collapse all files

847
app/amazon/amazon_base.py Normal file
View File

@@ -0,0 +1,847 @@
import traceback
import winreg
import subprocess
import time
import uuid
import requests
import json
import os
import re
from datetime import datetime
from DrissionPage import Chromium,ChromiumOptions
from DrissionPage.common import By
from DrissionPage._pages.chromium_tab import ChromiumTab
from typing import Literal,Dict,Any,TypeVar,Type
from amazon.tool import get_shop_info,show_notification
from config import runing_shop
# 定义类型变量，用于泛型类型注解
T = TypeVar('T', bound='AmamzonBase')
def kill_process(version: Literal["v5", "v6"]):
"""杀紫鸟客户端进程（独立函数版本）"""
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
self.url = None #用于记录当前链接，实现断点续传
def get_zinaio_exe(self, protocol_name: str = "superbrowser"):
"""
获取紫鸟安装目录
Args:
protocol_name: 协议名称，默认 "superbrowser"
Returns:
exe_path: 紫鸟浏览器可执行文件路径
"""
try:
key_path = rf"SOFTWARE\Classes\{protocol_name}\shell\open\command"
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, 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]
return exe_path
except FileNotFoundError:
try:
key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, 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]
return exe_path
except FileNotFoundError:
return None
def update_core(self):
"""
下载所有内核，打开店铺前调用，需客户端版本5.285.7以上
因为http有超时时间，所以这个action适合循环调用，直到返回成功
"""
data = {
"action": "updateCore",
"requestId": str(uuid.uuid4()),
}
data.update(self.user_info)
while True:
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()
print(result)
if result is None:
print("等待客户端启动...")
time.sleep(2)
continue
if result.get("statusCode") is None or result.get("statusCode") == -10003:
print("当前版本不支持此接口，请升级客户端")
return
elif result.get("statusCode") == 0:
print("更新内核完成")
return
else:
print(f"等待更新内核: {json.dumps(result)}")
time.sleep(2)
def kill_process(self, version: Literal["v5", "v6"]):
"""
杀紫鸟客户端进程
Args:
version: 客户端版本
"""
if version == "v5":
process_name = 'SuperBrowser.exe'
os.system('taskkill /f /t /im ' + "starter.exe")
else:
process_name = 'ziniao.exe'
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)
url = f'http://127.0.0.1:{self.socket_port}'
response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
r = response.json()
if str(r.get("statusCode")) == "0":
print(r)
return r.get("browserList")
elif str(r.get("statusCode")) == "-10003":
print(f"【get_browser_list】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
print(f"【get_browser_list】失败 {json.dumps(r, ensure_ascii=False)} ")
# exit()
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)
url = f'http://127.0.0.1:{self.socket_port}'
response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
r = response.json()
if str(r.get("statusCode")) == "0":
return r
elif str(r.get("statusCode")) == "-10003":
raise RuntimeError(f"【open_store】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
raise RuntimeError(f"【open_store】失败 {json.dumps(r, ensure_ascii=False)} ")
# exit()
def get_browser(self, port) -> Chromium:
"""
获取浏览器实例
Args:
port: 调试端口
Returns:
Chromium: DrissionPage浏览器实例
"""
co = ChromiumOptions()
# 设置不加载图片、静音
co.set_local_port(port)
co.no_imgs(True).mute(True)
browser = Chromium(co)
return browser
def start_client(self):
"""启动紫鸟客户端"""
# 检查端口是否已经启动
try:
url = f'http://127.0.0.1:{self.socket_port}'
response = requests.get(url, timeout=2)
print(f"端口 {self.socket_port} 已经启动，跳过启动客户端操作")
return
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
# 端口未启动，继续执行启动操作
print(f"端口 {self.socket_port} 未启动，开始启动客户端")
self.kill_process('v6')
time.sleep(5)
self.client_path = self.get_zinaio_exe("superbrowserv6").strip('"')
print(self.client_path)
cmd = [self.client_path, '--run_type=web_driver', '--ipc_type=http',
'--port=' + str(self.socket_port)]
print(" ".join(cmd))
# 最大重试次数
max_retries = 3
for retry_count in range(max_retries):
print(f"第 {retry_count + 1} 次尝试启动客户端...")
# 启动进程
subprocess.Popen(cmd)
# 循环检测10秒，每0.5秒检测一次
start_check_time = time.time()
client_started = False
while time.time() - start_check_time < 10:
try:
response = requests.get(url, timeout=2)
print(f"客户端启动成功！(第 {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
else:
print(f"第 {retry_count + 1} 次尝试启动失败，10秒内未检测到客户端启动")
# 超过最大重试次数，抛出异常
raise RuntimeError(f"客户端启动失败：重试 {max_retries} 次后仍未成功启动")
def open_shop(self, shop_name: str):
"""
打开指定店铺
Args:
shop_name: 店铺名称
Returns:
Chromium or str: 成功返回浏览器实例，失败返回错误信息
"""
print("=============打开指定店铺================")
# 启动客户端
self.start_client()
# 获取店铺列表
shop_ls = self.get_browser_list()
self.store_id = None
print(shop_ls)
for shop in shop_ls:
if shop.get("browserName") == shop_name:
self.store_id = shop.get('browserOauth')
break
if not self.store_id:
print("店铺不存在")
return "店铺不存在"
# 打开店铺
ret_json = self.open_store(self.store_id)
print(ret_json)
self.store_id = ret_json.get("browserOauth")
if self.store_id is None:
self.store_id = ret_json.get("browserId")
# 获取drissionpage浏览器会话
self.browser = self.get_browser(ret_json.get('debuggingPort'))
ip_check_url = ret_json.get("ipDetectionPage")
if not ip_check_url:
print("ip检测页地址为空，请升级紫鸟浏览器到最新版")
print(f"=====关闭店铺：{shop_name}=====")
self.close_store(self.store_id)
# exit()
raise RuntimeError("没有IP检测地址，为了店铺安全不打开店铺")
ip_usable = self.open_ip_check(self.browser, ip_check_url)
if ip_usable:
print("ip检测通过，打开店铺平台主页")
self.open_launcher_page(ret_json.get("launcherPage"), self.browser)
else:
print("IP检测不通过")
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)
url = f'http://127.0.0.1:{self.socket_port}'
response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
r = response.json()
if str(r.get("statusCode")) == "0":
return r
elif str(r.get("statusCode")) == "-10003":
raise RuntimeError(f"【close_store】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
raise RuntimeError(f"【close_store】失败： {json.dumps(r, ensure_ascii=False)} ")
# exit()
def open_launcher_page(self, launcher_page: str, browser: Chromium = None):
"""
打开启动页面
Args:
launcher_page: 要打开的页面URL
browser: 浏览器实例，如果不提供则使用当前浏览器实例
"""
if browser is None:
browser = self.browser
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:
tab = browser.latest_tab
tab.get(ip_check_url)
success_button = tab.ele((By.XPATH, '//button[contains(@class, "styles_btn--success")]'),
timeout=60) # 等待查找元素60秒
if success_button:
print("ip检测成功")
return True
else:
print("ip检测超时")
return False
except Exception as e:
print("ip检测异常:" + traceback.format_exc())
return False
class AmamzonBase(ZiniaoDriver):
"""亚马逊操作基类，包含一些通用方法"""
mark_name = "操作基类"
def log(self, message: str, level: str = "INFO"):
"""日志输出
Args:
message: 日志消息
level: 日志级别
"""
try:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if level == "ERROR":
show_notification(message, "error")
print(f"[{timestamp}] [{self.mark_name}] [{level}] {message}")
except Exception as e:
print(f"输出出错,{e}")
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:
print("=============切换国家================")
if self.browser is None:
print("浏览器实例不存在，请先打开店铺")
return False
# 获取当前标签页
tab = self.tab
tab.wait.doc_loaded(timeout=120, raise_err=False)
# 步骤1：获取当前国家名称
self.log(f"正在检查当前国家...")
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()
self.log(f"当前国家：{current_country}")
# 判断是否与目标国家相同
if current_country == country_name:
self.log(f"当前已经是目标国家 {country_name}，无需切换")
return True
else:
self.log("无法获取当前国家信息")
return False
# 步骤2：点击打开下拉框
self.log(f"正在打开国家切换下拉框...")
dropdown_header = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]', timeout=10)
if not dropdown_header:
self.log("找不到国家切换下拉框")
return False
dropdown_header.click()
time.sleep(1) # 等待下拉框展开
# 步骤3：点击第一个展开国家列表
self.log(f"正在展开国家列表...")
first_item = tab.ele('xpath://div[@class="dropdown-account-switcher-list-item"]', timeout=10)
if not first_item:
self.log("找不到国家列表项")
return False
first_item.click()
time.sleep(1) # 等待国家列表展开
# 步骤4：点击目标国家
self.log(f"正在切换到国家：{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:
self.log(f"找不到目标国家：{country_name}")
return False
target_country.click()
# 步骤5：等待页面加载完成并验证切换结果
self.log(f"等待页面加载...")
# time.sleep(3) # 等待页面加载
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:
self.log(f"国家切换成功：{new_country}")
return True
else:
self.log(f"国家切换失败，当前国家：{new_country}，目标国家：{country_name}")
return False
else:
self.log("无法验证切换结果")
return False
except Exception as e:
self.log(f"切换国家时发生异常：{traceback.format_exc()}")
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:
self.log("检测到需要登录元素")
return True
return False
def login(self, password, username=""):
try:
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
for _ in range(4):
switch_account = self.tab.eles(
'xpath://div[@data-test-id="switchableAccounts"]//div[@class="a-fixed-left-grid"]', timeout=5)
if len(switch_account) > 0:
switch_account[0].click()
time.sleep(1)
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
pwd_input = self.tab.eles('xpath://input[@type="password"]', timeout=5)
if len(pwd_input) > 0:
pwd_input[0].input(password, clear=True)
submit_btn = self.tab.eles('xpath://input[@id="signInSubmit"]', timeout=5)
if len(submit_btn) > 0:
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)
self.log(f"发送一次性密码:{len(send_code)}")
if len(send_code) > 0:
self.log("检测到 发送一次性密码")
send_code[0].click()
time.sleep(1)
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
else:
for j in range(5):
opt_code_input = self.tab.eles('xpath://input[@name="otpCode"]', timeout=30)
self.log(f"验证码输入框:{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() != "":
print("检测到验证码输入完成")
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)
# return True
error_mes = self.tab.eles('xpath://div[@id="auth-error-message-box"]',
timeout=10)
if len(error_mes) > 0:
self.log("验证码输入错误")
self.log(f"验证码输入错误提示:{error_mes[0].text}")
self.tab.refresh()
else:
self.log("登入成功")
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:
submit_btn[0].click()
except Exception as e:
self.log(f"登录过程中发生异常:{traceback.format_exc()}")
return False
def SwitchPage(self):
"""
切换至 管理所有库存页面
1、等待 //navigation-favorites-bar[@class="hydrated"] 出现
"""
navigation = self.tab.ele('xpath://navigation-favorites-bar[@class="hydrated"]')
navigation.wait.displayed(raise_err=False)
page_btn = navigation.sr('xpath://internal-fav-bar-links[@data-internal="navigation"]').sr(
'xpath://a[@data-page-id="ezdpc-gui-inventory-mons"]')
page_btn.wait.displayed(raise_err=False)
page_btn.click(timeout=5)
self.tab.wait.doc_loaded()
# 等待搜索框出现
search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group')
search_region.wait.displayed(raise_err=False, timeout=60)
def search(self,filter_type="ApprovalRequired"):
sku_ls = []
load_ele = self.tab.eles("xpath://div[contains(@class,'Loader-module__loader')]",timeout=5)
if len(load_ele) > 0:
load_ele[0].wait.deleted(timeout=3, raise_err=False)
time.sleep(0.5)
drop_down = self.tab.ele('xpath://div[contains(@class,"VolusListingStatusDropDown-module__verticalContainer")]//kat-dropdown')
drop_down.wait.displayed(raise_err=False)
drop_down.wait.enabled(raise_err=False)
time.sleep(0.6)
drop_down.click()
# //kat-option[@value="SearchSuppressed"]
xp = f'xpath://kat-option[@value="{filter_type}"]'
self.log(f"正在寻找筛选条件 {filter_type}，xpath: {xp}")
approval_required = self.tab.eles(xp,timeout=5)
if len(approval_required) == 0:
self.log(f"没有需要{filter_type}选项】没有需要{filter_type}的商品了")
return sku_ls # "没有需要审批的商品了"
else:
approval_required = approval_required[0]
approval_required.wait.displayed(raise_err=False)
approval_required.click()
approval_required_text = approval_required.text
self.log(f"已选择筛选条件: {approval_required_text}")
count = re.findall(r'\d+', approval_required_text)
if count:
count = int(count[0])
self.log(f"待审批的商品数量: {count}")
if count <= 0:
self.log(f"没有需要{filter_type}的商品了")
return sku_ls #"没有需要审批的商品了"
for _ in range(3):
# 等待加载完成
load_ele = self.tab.eles("xpath://div[contains(@class,'Loader-module__loader')]")
if len(load_ele) > 0:
load_ele[0].wait.deleted(timeout=3, raise_err=False)
time.sleep(0.5)
sku_ls = self.tab.eles("xpath://div[@data-sku]",timeout=3)
if len(sku_ls) > 0:
break
approval_required.click()
return sku_ls
class TaskBase:
task_name = "任务基类"
country_info = {
"DE": "德国",
"FR": "法国",
"ES": "西班牙",
"IT": "意大利",
"UK": "英国"
}
def __init__(self, user_info: dict = None):
"""初始化审批任务处理器
Args:
user_info: 用户信息字典，包含 company, username, password
"""
self.user_info = user_info or {}
self.running = True
def log(self, message: str, level: str = "INFO"):
"""日志输出
Args:
message: 日志消息
level: 日志级别
"""
try:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if level == "ERROR":
show_notification(message, "error")
print(f"[{timestamp}] [{self.task_name}] [{level}] {message}")
except Exception as e:
print(f"输出出错,{e}")
def process_task(self, task_data: dict):
pass
def open_shop(self, cls: Type[T], max_retries: int, company_name: str, shop_name: str, iskill: bool = False) -> T:
error_info = ""
driver = None
for retry in range(max_retries):
try:
self.log(f"尝试打开店铺 {shop_name} (第 {retry + 1}/{max_retries} 次)")
if iskill:
self.log("重试前先杀掉浏览器进程...")
kill_process("v6")
kill_process("v5")
time.sleep(2)
# 组装用户信息并创建驱动
user_info = {
**self.user_info,
"company": company_name
}
driver = cls(user_info)
browser = driver.open_shop(shop_name)
if browser and browser != "店铺不存在":
self.log(f"成功打开店铺 {shop_name}")
else:
self.log(f"打开店铺失败: {browser}", "WARNING")
driver = None
continue
# 判断是否需要登录
need_login = driver.need_login()
print("【是否需要登录】：", need_login)
if need_login:
self.log(f"店铺 {shop_name} 需要登录，正在登录...")
# 获取店铺凭证
response = get_shop_info(shop_name)
print("【获取店铺凭证返回】：", response.text)
shop_data = response.json()
if not shop_data:
mes = f"获取店铺凭证失败，响应数据: {shop_data.get('message', '未知错误')}"
self.log(mes, "ERROR")
show_notification(mes, "ERROR")
continue
password = shop_data["data"]["password"]
login_success = driver.login(password)
if login_success:
self.log(f"店铺 {shop_name} 登录成功，正在重新打开店铺...")
browser = driver.open_shop(shop_name)
if browser and browser != "店铺不存在":
self.log(f"成功打开店铺 {shop_name} 登录后")
break
else:
self.log(f"登录后打开店铺失败: {browser}", "WARNING")
driver = None
else:
self.log(f"店铺 {shop_name} 登录失败", "WARNING")
driver = None
else:
break
except Exception as e:
import traceback
self.log(f"打开店铺异常: {traceback.format_exc()}", "INFO")
driver = None
error_info = str(e)
time.sleep(10)
# 如果还有重试机会，等待后继续
if retry < max_retries - 1:
time.sleep(3)
# 检查是否成功打开
if not driver or not browser or browser == "店铺不存在":
error_msg = f"店铺 {shop_name} 打开失败，已重试 {max_retries} 次，跳过该店铺,{error_info}"
self.log(error_msg, "ERROR")
# 从执行列表中移除
if shop_name in runing_shop:
del runing_shop[shop_name]
return driver
return driver
def process_shop(self, shop_data: Dict[str, Any], task_id: int):
pass
def action_init(self, driver, country_name, shop_name, risk_listing_filter=None):
"""切换到指定国际 -> 页面 -》 筛选好"""
max_retries = 5
switch_success = False
switch_success_pg = False
search_success = False
sku_ls = []
for retry in range(max_retries):
try:
self.log(f"尝试切换到国家 {country_name} (第 {retry + 1}/{max_retries} 次)")
if retry > 2:
# 刷新不行就重新打开店铺
self.log("重试前重新打开店铺...")
try:
driver.close_store()
time.sleep(3)
driver.open_shop(shop_name)
except Exception as e:
self.log(f"关闭重新打开店铺: {str(e)}", "WARNING")
# 如果不是第一次尝试，先刷新页面
if retry > 0:
self.log("重试前刷新页面...")
try:
driver.tab.refresh()
time.sleep(3)
except Exception as e:
self.log(f"刷新页面失败: {str(e)}", "WARNING")
switch_success = driver.SwitchingCountries(country_name)
if switch_success:
self.log(f"成功切换到国家 {country_name}")
# 切换到库存管理页面
driver.SwitchPage()
self.log(f"已切换到库存管理页面")
time.sleep(3)
switch_success_pg = True
if risk_listing_filter is None:
self.log(f"risk_listing_filter 为空，不筛选, {risk_listing_filter}")
search_success = True
break
sku_ls = driver.search(filter_type=risk_listing_filter)
search_success = True
self.log(f"国家 {country_name} 搜索出 {len(sku_ls)} 商品，开始处理...")
break
else:
self.log(f"切换到国家 {country_name} 失败", "WARNING")
except Exception as e:
import traceback
self.log(f"切换国家 {country_name} 异常: {str(e)}", "ERROR")
self.log(traceback.format_exc(), "ERROR")
# 如果还有重试机会，等待后继续
if retry < max_retries - 1:
time.sleep(2)
return switch_success, switch_success_pg, search_success, sku_ls