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

modified: amazon/__pycache__/approve.cpython-39.pyc

Browse Source 
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
This commit is contained in:
铭坤
2026-04-08 16:03:48 +08:00
parent 3149d80456
commit 7151f6b5e0
9 changed files with 231 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
app/amazon/del_brand.py
View File

@@ -10,6 +10,7 @@ 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"]):
@@ -33,7 +34,7 @@ class ZiniaoDriver:
self.socket_port = socket_port
self.client_path = None
self.browser = None
self.tab = None
self.tab: ChromiumTab = None
self.store_id = None
def get_zinaio_exe(self, protocol_name: str = "superbrowser"):
@@ -136,11 +137,11 @@ class ZiniaoDriver:
print(r)
return r.get("browserList")
elif str(r.get("statusCode")) == "-10003":
print(f"login Err {json.dumps(r, ensure_ascii=False)}")
exit()
print(f"【get_browser_list】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
exit()
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=""):
@@ -188,11 +189,11 @@ class ZiniaoDriver:
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()
raise RuntimeError(f"【open_store】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
exit()
raise RuntimeError(f"【open_store】失败 {json.dumps(r, ensure_ascii=False)} ")
# exit()
def get_browser(self, port) -> Chromium:
"""
@@ -209,6 +210,17 @@ class ZiniaoDriver:
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',
@@ -261,7 +273,8 @@ class ZiniaoDriver:
print("ip检测页地址为空，请升级紫鸟浏览器到最新版")
print(f"=====关闭店铺：{shop_name}=====")
self.close_store(self.store_id)
exit()
# exit()
raise RuntimeError("没有IP检测地址，为了店铺安全不打开店铺")
ip_usable = self.open_ip_check(self.browser, ip_check_url)
if ip_usable:
print("ip检测通过，打开店铺平台主页")
@@ -298,11 +311,11 @@ class ZiniaoDriver:
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()
raise RuntimeError(f"【close_store】登录失败 {json.dumps(r, ensure_ascii=False)}")
# exit()
else:
print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
exit()
raise RuntimeError(f"【close_store】失败： {json.dumps(r, ensure_ascii=False)} ")
# exit()
def open_launcher_page(self, launcher_page: str, browser: Chromium = None):
"""
@@ -436,8 +449,38 @@ class AmamzonBase(ZiniaoDriver):
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"""