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 336338cff5
commit e02ddab599
9 changed files with 231 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
app/amazon/approve.py
View File

@@ -7,6 +7,12 @@ from config import runing_task, runing_shop
from datetime import datetime
from amazon.del_brand import AmamzonBase, kill_process
# 导入 webview 用于前端通知
try:
import webview
except ImportError:
webview = None
class AmzoneApprove(AmamzonBase):
def SwitchPage(self):
@@ -190,8 +196,8 @@ class AmzoneApprove(AmamzonBase):
num = 0
while 1:
if num > 3: #测试
return
# if num > 3: #测试
# return
# 等待加载完成
load_ele = self.tab.eles("xpath://div[contains(@class,'Loader-module__loader')]",timeout=5)
if len(load_ele) > 0:
@@ -202,14 +208,17 @@ class AmzoneApprove(AmamzonBase):
page_pamel = self.tab.eles('xpath://kat-pagination',timeout=5)
if len(page_pamel) > 0:
current_page = page_pamel[0].sr('xpath:.//ul[@class="pages"]//li[@aria-current="true"]').text
print(f"开始处理，当前页码: {current_page}")
print(f"【current_page】开始处理，当前页码: {current_page}")
sku_ls = self.tab.eles("xpath://div[@data-sku]",timeout=3)
sku_ls = self.tab.eles("xpath://div[@data-sku]",timeout=10)
print(f"获取到 {len(sku_ls)}")
for sku_ele in sku_ls[0:2]:
# for sku_ele in sku_ls[0:2]:
for sku_ele in sku_ls:
# solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]')
solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]|.//kat-link[@label="修复被禁止显示的商品"]|//kat-link[@label="解决商品移除风险"]')
solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]|.//kat-link[@label="修复被禁止显示的商品"]|.//kat-link[@label="解决商品移除风险"]')
if len(solve_problem) == 0:
print(f"{asin},没有处理入口，不处理")
yield (asin,"没有处理入口，不处理")
continue
asin = sku_ele.ele('xpath:.//div[contains(@class,"JanusSplitBox-module__container")]//div[contains(@class,"JanusSplitBox-module__panel--") and contains(string(.),"ASIN")]/..//div[last()]').text
print(f"ASIN {asin} 存在问题，正在点击解决...")
@@ -235,9 +244,9 @@ class AmzoneApprove(AmamzonBase):
# 解决商品信息违规问题 按钮
kat_box_ls = action_panel.eles('xpath:.//kat-box[@variant="white"]',timeout=5)
if len(kat_box_ls) == 0:
print(f"ASIN {asin} 的操作面板中未找到选项，跳过...")
continue
# if len(kat_box_ls) == 0:
# print(f"ASIN {asin} 的操作面板中未找到选项，跳过...")
# continue
print(f"ASIN {asin}需要处理的有：{len(kat_box_ls)}个问题.")
# 解决商品信息违规问题 按钮，有多少个都要处理
for i in range(len(kat_box_ls)):
@@ -284,7 +293,7 @@ class AmzoneApprove(AmamzonBase):
kat_box_ls = action_panel.eles('xpath:.//kat-box[@variant="white"]', timeout=5)
# 修复商品信息
repair_product = action_panel.eles('xpath://kat-button[@class="action-button"]',timeout=5)
repair_product = action_panel.eles('xpath:.//kat-button[@class="action-button"]',timeout=5)
if len(repair_product) > 0:
print(f"ASIN {asin} 存在修复商品信息按钮，正在点击...")
self.clear_tab()
@@ -316,7 +325,7 @@ class AmzoneApprove(AmamzonBase):
break
next_page_btn.click()
num += 1
print(f"正在翻页，已翻 {num} 页...")
# print(f"正在翻页，已翻 {num} 页...")
class ApproveTask:
@@ -330,6 +339,75 @@ class ApproveTask:
"UK": "英国"
}
@staticmethod
def show_notification(message: str, message_type: str = "error"):
"""显示 pywebview 顶层通知（5秒后自动消失）
Args:
message: 通知消息
message_type: 消息类型 (success/warning/error/info)
"""
if webview and webview.windows:
try:
# 转义单引号，防止 JavaScript 语法错误
safe_message = message.replace("'", "\\'").replace('"', '\\"').replace('\n', '\\n')
# 设置通知样式颜色
color_map = {
'success': '#67C23A',
'warning': '#E6A23C',
'error': '#F56C6C',
'info': '#909399'
}
bg_color = color_map.get(message_type, '#F56C6C')
# 构建前端通知的 JavaScript 代码 - 创建原生 HTML 通知
js_code = f"""
(function() {{
// 移除已存在的通知
var existingNotif = document.getElementById('pywebview-notification');
if (existingNotif) {{
existingNotif.remove();
}}
// 创建通知容器
var notif = document.createElement('div');
notif.id = 'pywebview-notification';
notif.style.cssText = 'position: fixed; top: 20px; left: 50%; transform: translateX(-50%); ' +
'background: {bg_color}; color: white; padding: 12px 20px; border-radius: 4px; ' +
'box-shadow: 0 2px 12px rgba(0,0,0,0.3); z-index: 99999; font-size: 14px; ' +
'max-width: 600px; word-wrap: break-word; display: flex; align-items: center; gap: 10px;';
// 添加消息内容
var msgSpan = document.createElement('span');
msgSpan.textContent = '{safe_message}';
notif.appendChild(msgSpan);
// 添加关闭按钮
var closeBtn = document.createElement('span');
closeBtn.innerHTML = '×';
closeBtn.style.cssText = 'cursor: pointer; font-size: 18px; font-weight: bold; margin-left: 10px;';
closeBtn.onclick = function() {{ notif.remove(); }};
notif.appendChild(closeBtn);
// 添加到页面
document.body.appendChild(notif);
// 5秒后自动消失
setTimeout(function() {{
if (notif && notif.parentNode) {{
notif.style.transition = 'opacity 0.3s';
notif.style.opacity = '0';
setTimeout(function() {{ notif.remove(); }}, 300);
}}
}}, 5000);
}})();
"""
# 在第一个窗口中执行 JavaScript
webview.windows[0].evaluate_js(js_code)
except Exception as e:
print(f"显示通知失败: {str(e)}")
"""
处理流程：
1、解析任务json,提取对应的companyName 组组装user_info实例化 AmzoneApprove
@@ -385,6 +463,8 @@ class ApproveTask:
"""
from datetime import datetime
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if level == "ERROR":
self.show_notification(message, "error")
print(f"[{timestamp}] [ApproveTask] [{level}] {message}")
def process_task(self, task_data: dict):
@@ -439,6 +519,7 @@ class ApproveTask:
shop_name = shop_item.get("shopName", "未知店铺")
self.log(f"[{idx}/{len(items)}] 开始处理店铺: {shop_name}")
self.show_notification(f"开始处理店铺: {shop_name}", "info")
try:
self.process_shop(shop_item, country_codes, task_id,risk_listing_filter)
@@ -502,11 +583,11 @@ class ApproveTask:
self.log(f"尝试打开店铺 {shop_name} (第 {retry + 1}/{max_retries} 次)")
# 如果不是第一次尝试，先杀进程
if retry > 0:
self.log("重试前先杀掉浏览器进程...")
kill_process("v6")
kill_process("v5")
time.sleep(2)
# if retry > 0:
# self.log("重试前先杀掉浏览器进程...")
# kill_process("v6")
# kill_process("v5")
# time.sleep(2)
# 组装用户信息并创建驱动
user_info = {
@@ -522,7 +603,8 @@ class ApproveTask:
else:
self.log(f"打开店铺失败: {browser}", "WARNING")
driver = None
# 判断是否需要登录
except Exception as e:
import traceback
self.log(f"打开店铺异常: {str(e)}", "ERROR")
@@ -535,7 +617,8 @@ class ApproveTask:
# 检查是否成功打开
if not driver or not browser or browser == "店铺不存在":
self.log(f"店铺 {shop_name} 打开失败，已重试 {max_retries} 次，跳过该店铺", "ERROR")
error_msg = f"店铺 {shop_name} 打开失败，已重试 {max_retries} 次，跳过该店铺"
self.log(error_msg, "ERROR")
# 从执行列表中移除
if shop_name in runing_shop:
del runing_shop[shop_name]
@@ -589,7 +672,9 @@ class ApproveTask:
# 转换国家代码为中文名称
country_name = self.country_info.get(country_code, country_code)
self.log(f"开始处理国家: {country_name} ({country_code})")
info_mes = f"开始处理国家: {country_name} ({country_code})"
self.log(info_mes)
self.show_notification(info_mes, "info")
# 更新当前处理的国家
if task_id in runing_task:
@@ -630,7 +715,8 @@ class ApproveTask:
# 如果切换失败，直接返回
if not switch_success:
self.log(f"切换到国家 {country_name} 失败，已重试 {max_retries} 次，跳过该国家", "ERROR")
error_message = f"切换到国家 {country_name} 失败，已重试 {max_retries} 次，跳过该国家"
self.log(error_message, "ERROR")
if task_id in runing_task:
runing_task[task_id]["processed_countries"] += 1
return
@@ -819,3 +905,5 @@ if __name__ == '__main__':
print(f"ASIN {asin} 的处理结果: {status}")
print("已完成操作")
# for i in range(0):
# print(i)