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

modified: app/ali_oss.py

Browse Source 
new file:   app/amazon/del_brand.py
	new file:   app/amazon/main.py
	modified:   app/blueprints/__pycache__/brand.cpython-39.pyc
	modified:   app/blueprints/__pycache__/communication.cpython-39.pyc
	modified:   app/blueprints/__pycache__/main.cpython-39.pyc
	modified:   app/blueprints/brand.py
	new file:   "app/blueprints/brand_\345\244\207\344\273\275.py"
	modified:   app/blueprints/communication.py
	modified:   app/blueprints/main.py
	modified:   app/config.py
	modified:   app/main.py
This commit is contained in:
铭坤
2026-04-01 11:53:12 +08:00
parent 6eb0f33421
commit 4df3945131
12 changed files with 1779 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

519
app/amazon/del_brand.py Normal file
View File

@@ -0,0 +1,519 @@
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
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 = 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):
exe_path = command.split()[0]
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):
exe_path = command.split()[0]
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'
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"login Err {json.dumps(r, ensure_ascii=False)}")
exit()
else:
print(f"Fail {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":
print(f"login Err {json.dumps(r, ensure_ascii=False)}")
exit()
else:
print(f"Fail {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):
"""启动紫鸟客户端"""
self.client_path = self.get_zinaio_exe("superbrowser").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()
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":
print(f"login Err {json.dumps(r, ensure_ascii=False)}")
exit()
else:
print(f"Fail {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 AmazoneDriver(ZiniaoDriver):
"""亚马逊专用驱动类，继承自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 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)
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=20)
time.sleep(1)
sku_ls = self.tab.eles("//div[@data-sku]")
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.ele("xpath:.//button[@role='menuitem' and text()='删除商品信息']")
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)