7151f6b5e0
modified: amazon/__pycache__/del_brand.cpython-39.pyc modified: amazon/__pycache__/main.cpython-39.pyc modified: amazon/approve.py new file: amazon/backend_api.py modified: amazon/del_brand.py modified: amazon/main.py modified: blueprints/main.py modified: main.py
576 lines
21 KiB
Python
576 lines
21 KiB
Python
import traceback
|
||
import winreg
|
||
import subprocess
|
||
import time
|
||
import uuid
|
||
import requests
|
||
import json
|
||
import os
|
||
|
||
from typing import Literal
|
||
from DrissionPage import Chromium
|
||
from DrissionPage.common import By
|
||
from DrissionPage._pages.chromium_tab import ChromiumTab
|
||
|
||
|
||
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
|
||
|
||
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": False,
|
||
"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浏览器实例
|
||
"""
|
||
browser = Chromium(port)
|
||
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.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))
|
||
subprocess.Popen(cmd)
|
||
time.sleep(5)
|
||
|
||
def open_shop(self, shop_name: str):
|
||
"""
|
||
打开指定店铺
|
||
|
||
Args:
|
||
shop_name: 店铺名称
|
||
|
||
Returns:
|
||
Chromium or str: 成功返回浏览器实例，失败返回错误信息
|
||
"""
|
||
# 启动客户端
|
||
self.start_client()
|
||
|
||
# 更新内核
|
||
self.update_core()
|
||
|
||
# 获取店铺列表
|
||
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检测不通过")
|
||
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):
|
||
"""亚马逊操作基类，包含一些通用方法"""
|
||
|
||
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:
|
||
print("浏览器实例不存在，请先打开店铺")
|
||
return False
|
||
|
||
# 获取当前标签页
|
||
tab = self.tab
|
||
|
||
# 步骤1：获取当前国家名称
|
||
print(f"正在检查当前国家...")
|
||
current_country_ele = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]',
|
||
timeout=10)
|
||
if current_country_ele:
|
||
current_country = current_country_ele.text.strip()
|
||
print(f"当前国家：{current_country}")
|
||
|
||
# 判断是否与目标国家相同
|
||
if current_country == country_name:
|
||
print(f"当前已经是目标国家 {country_name}，无需切换")
|
||
return True
|
||
else:
|
||
print("无法获取当前国家信息")
|
||
return False
|
||
|
||
# 步骤2：点击打开下拉框
|
||
print(f"正在打开国家切换下拉框...")
|
||
dropdown_header = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]', timeout=10)
|
||
if not dropdown_header:
|
||
print("找不到国家切换下拉框")
|
||
return False
|
||
dropdown_header.click()
|
||
time.sleep(1) # 等待下拉框展开
|
||
|
||
# 步骤3：点击第一个展开国家列表
|
||
print(f"正在展开国家列表...")
|
||
first_item = tab.ele('xpath://div[@class="dropdown-account-switcher-list-item"]', timeout=10)
|
||
if not first_item:
|
||
print("找不到国家列表项")
|
||
return False
|
||
first_item.click()
|
||
time.sleep(1) # 等待国家列表展开
|
||
|
||
# 步骤4：点击目标国家
|
||
print(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:
|
||
print(f"找不到目标国家：{country_name}")
|
||
return False
|
||
target_country.click()
|
||
|
||
# 步骤5：等待页面加载完成并验证切换结果
|
||
print(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:
|
||
print(f"国家切换成功：{new_country}")
|
||
return True
|
||
else:
|
||
print(f"国家切换失败，当前国家：{new_country}，目标国家：{country_name}")
|
||
return False
|
||
else:
|
||
print("无法验证切换结果")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"切换国家时发生异常：{traceback.format_exc()}")
|
||
return False
|
||
|
||
def need_login(self):
|
||
"""
|
||
判断是否需要登录，部分国家可能需要登录后才能切换国家
|
||
处理流程：
|
||
|
||
"""
|
||
need_login_ele = self.tab.eles('xpath://h1[@class="a-spacing-small"]',timeout=5)
|
||
if len(need_login_ele) >0:
|
||
print("检测到需要登录元素")
|
||
return True
|
||
return False
|
||
|
||
def login(self,password,username=""):
|
||
try:
|
||
pwd_input = self.tab.ele('xpath://input[@type="password"]',timeout=10)
|
||
pwd_input.input(password)
|
||
submit_btn = self.tab.ele('xpath://input[@id="signInSubmit"]',timeout=10)
|
||
submit_btn.click()
|
||
self.tab.wait.doc_loaded()
|
||
|
||
opt_code_input = self.tab.ele('xpath://input[@name="otpCode"]',timeout=10)
|
||
opt_code_input.wait.displayed(timeout=10,raise_err=False)
|
||
for _ in range(30):
|
||
if opt_code_input.value is not None and opt_code_input.value.strip() != "":
|
||
print("检测到验证码输入完成")
|
||
submit_btn = self.tab.ele('xpath://input[@id="auth-signin-button"]',timeout=10)
|
||
submit_btn.click()
|
||
self.tab.wait.doc_loaded()
|
||
return True
|
||
except Exception as e:
|
||
print("登录过程中发生异常:" + traceback.format_exc())
|
||
return False
|
||
|
||
class AmazoneDriver(AmamzonBase):
|
||
"""亚马逊专用驱动类，继承自ZiniaoDriver"""
|
||
|
||
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)
|
||
|
||
def search(self, asin):
|
||
search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group')
|
||
search_region.wait.displayed(raise_err=False)
|
||
time.sleep(0.6)
|
||
search_input = self.tab.ele("xpath://kat-input[contains(@class,'SearchBox-module__searchInput')]").sr(
|
||
'xpath://span[@class="container"]//input[@part="input"]')
|
||
search_input.input(asin,clear=True)
|
||
sku_ls = []
|
||
for _ in range(3):
|
||
search_btn = self.tab.ele("xpath://kat-icon[@name='search']")
|
||
search_btn.click()
|
||
|
||
load_ele = self.tab.ele("xpath://div[contains(@class,'Loader-module__loader')]")
|
||
# load_ele.wait.hidden(timeout=3, raise_err=False)
|
||
load_ele.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
|
||
return sku_ls
|
||
|
||
def del_action(self, sku_ele):
|
||
dropdown = sku_ele.ele('xpath:.//kat-dropdown-button[@variant="secondary"]')
|
||
dropdown.click()
|
||
time.sleep(1)
|
||
|
||
del_btn = dropdown.sr("xpath:.//button[@role='menuitem' and @data-action='DeleteListing']")
|
||
del_btn.wait.displayed(raise_err=False)
|
||
del_btn.click()
|
||
|
||
confirm_btn = self.tab.ele('xpath://kat-modal[@data-testid="action-modal"]//kat-button[@variant="primary"]')
|
||
confirm_btn.wait.displayed(raise_err=False)
|
||
confirm_btn.wait.enabled()
|
||
confirm_btn.click()
|
||
|
||
suc_alert = self.tab.ele("xpath://kat-alert[@variant='success']")
|
||
suc = suc_alert.wait.displayed(raise_err=False, timeout=5)
|
||
return suc
|
||
|
||
|
||
if __name__ == '__main__':
|
||
# 使用示例
|
||
user_info = {
|
||
"company": "rongchuang123",
|
||
"username": "自动化_Robot",
|
||
"password": "#20zsg25"
|
||
}
|
||
shop_name = "郭亚芳"
|
||
kill_process("v6")
|
||
|
||
# 创建驱动实例
|
||
driver = AmazoneDriver(user_info)
|
||
|
||
# 打开店铺并获取浏览器实例
|
||
browser = driver.open_shop(shop_name)
|
||
|
||
country = "西班牙"
|
||
asin = "Voanos"
|
||
|
||
sw_suc = driver.SwitchingCountries(country)
|
||
driver.SwitchPage()
|
||
sku_ls = driver.search(asin=asin)
|
||
print("查询结果有：", len(sku_ls))
|
||
|
||
for sku in sku_ls:
|
||
suc = driver.del_action(sku)
|
||
print(sku, "删除结果", suc)
|
||
|
||
|
||
|