Files
crawler-plugin/app/amazon/approve.py
2026-05-24 21:59:21 +08:00

948 lines
47 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import time
import re
import traceback
import requests
from DrissionPage._pages.chromium_tab import ChromiumTab
from config import runing_task, runing_shop,DELETE_BRAND_API_BASE
from datetime import datetime
from amazon.amazon_base import AmamzonBase,kill_process,TaskBase
from amazon.tool import show_notification
class AmzoneApprove(AmamzonBase):
mark_name = "产品风险审批"
def __init__(self, user_info: dict, socket_port: int = 19890):
super().__init__(user_info,socket_port)
self.already_asin = set()
def reset_already_asin(self):
self.already_asin = set()
self.log("already_asin 已重置")
def wait_loaded(self):
# 等待加载完成
try:
load_ele = self.tab.eles(
'xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]//div[contains(@class,"Loader-module__loader")]|//kat-panel[@data-testid="kat-panel-ActionPanelContent"]/div[@data-f1-component]//div[contains(@class,"==")]/div[contains(@class,"==")]/span',
timeout=3)
if len(load_ele) > 0:
load_ele[0].wait.deleted(timeout=5, raise_err=False)
except Exception as e:
print("等待加载中消失出错", e)
def clear_tab(self):
for i in ["亚马逊","您的账户是否符合条件?"]:
all_tab = self.browser.get_tabs(title=i)
close_tab = []
for tab in all_tab:
tab_id = tab if isinstance(tab,str) else tab.tab_id
if self.tab.tab_id == tab_id:
continue
close_tab.append(tab)
print("需要关闭的标签页",close_tab)
print("当前操作的tab_id",self.tab.tab_id)
self.browser.close_tabs(close_tab)
def handle_repair_product(self,tab:ChromiumTab,headle_type=""):
def wait_loaded(wait_type="contentWrapper"):
# 等待加载完成
print("------------【修复商品信息】等待加载完成-----------------")
for _ in range(3):
try:
load_ele = tab.eles(
'xpath://div[@class="contentWrapper"]//div[contains(@class,"==")]/span[not(node())]|//kat-spinner[@id="ahd-app-page-spinner"]',
timeout=10)
if len(load_ele) > 0:
load_ele[0].wait.deleted(timeout=20, raise_err=False)
if wait_type == "contentWrapper":
contentWrapper = tab.eles('xpath://div[@class="contentWrapper"]//h4',timeout=20)
if len(contentWrapper) > 0:
contentWrapper[0].wait.displayed(timeout=20, raise_err=True)
break
else:
time.sleep(2)
break
if wait_type == "content":
contentWrapper = tab.eles('xpath://div[@class="contentWrapper"]',timeout=20)
if len(contentWrapper) > 0:
contentWrapper[0].wait.displayed(timeout=20, raise_err=True)
break
else:
time.sleep(2)
break
if wait_type == "section":
section = tab.eles('xpath://section[@class="ahd-product-policy__section"]',timeout=20)
if len(section) > 0:
section[0].wait.displayed(timeout=20, raise_err=True)
break
except Exception as e:
print("等待加载中消失出错", e)
print("-----------------------------")
def close_contentWrapper():
# 关闭按钮
close_x_btn = tab.eles('xpath://div[@class="flyoutPanelContent"]/span',timeout=5)
contentWrapper = tab.eles('xpath://div[@class="flyoutPanelContent"]/div[@class="contentWrapper"]',timeout=5)
if len(close_x_btn) > 0 and len(contentWrapper)>0 :
if len(contentWrapper[0].children()) > 0:
close_x_btn[0].wait.displayed(timeout=5, raise_err=False)
close_x_btn[0].click()
tab.wait.doc_loaded(timeout=60,raise_err=False)
wait_loaded(wait_type="section")
# product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy"]')
product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy" and @class="ahd-product-policy-table-row-wrapper"]|//div[@class="ahd-product-policy-table-row-wrapper-nested" and @data-testid="ahd-desktop-product-policy"]')
if len(product_policy) == 0:
print("未找到提交处理区域,可能页面未加载完成或者页面结构发生变化")
time.sleep(5)
# product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy"]')
product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy" and @class="ahd-product-policy-table-row-wrapper"]|//div[@class="ahd-product-policy-table-row-wrapper-nested" and @data-testid="ahd-desktop-product-policy"]')
print("等待5秒后再次查找提交处理区域",len(product_policy))
# submit_btn_ls = tab.eles('xpath://div[@id="ahd-product-policies-table"]/div//div[@data-testid="nextStepsMetricWrapper"]//kat-button[@label="提交"]',timeout=10)
# if len(submit_btn_ls) == 0:
# return
now_index = 0
submit_index = 0
for _ in range(3):
try:
# 先全部点开
chevron_right_ls = tab.eles('xpath://kat-icon[@name="chevron-right"]',timeout=10)
for ele in chevron_right_ls:
ele.click()
time.sleep(0.5)
except Exception as e:
print("【修复商品信息】全部点开失败",e)
try:
# todo 遍历
print("【商品修复产品卡片】数量",len(product_policy))
print("【商品修复产品卡片】列表",product_policy)
for index in range(now_index,len(product_policy)):
product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy" and @class="ahd-product-policy-table-row-wrapper"]|//div[@class="ahd-product-policy-table-row-wrapper-nested" and @data-testid="ahd-desktop-product-policy"]')
now_index = index
if index > len(product_policy)-1:
print("提交按钮区域发生变化,停止处理")
break
message_link = product_policy[index].eles('xpath:.//a[@data-testid="ahd-reason-message-link"]')
message = message_link[0].text
if headle_type in ['查看冻结原因']:
need_heanle = any(i in message for i in ["制造商详细联系信息","警告和安全信息","负责人详细联系信息"])
else:
need_heanle = True
if not need_heanle:
print(f"{headle_type}】【{message}】不需要处理,跳过")
continue
spread_btn = product_policy[index].eles('xpath:.//kat-icon[@name="chevron-right"]',timeout=5)
print(f"{index},是否存在展开按钮",len(spread_btn))
if len(spread_btn) > 0:
spread_btn[0].click()
time.sleep(0.5)
submit_btn_ls = product_policy[index].eles('xpath:.//kat-button[@label="提交"]',timeout=5)
# try:
# submit_btn = product_policy[index].ele('xpath:.//kat-button[@label="提交"]',timeout=5)
# except Exception as e:
# print(datetime.now().strftime("%Y-%m-%d %H:%M:%S") ,"没有查找到提交按钮,跳过!")
# close_contentWrapper()
# continue
# print("提交按钮数量:",len(submit_btn))
print("提交按钮列表:",submit_btn_ls)
if len(submit_btn_ls) > 0:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"查找到提交按钮")
# submit_btn = submit_btn[0]
submit_btn = tab.eles('xpath://kat-button[@label="提交"]',timeout=10)[submit_index]
submit_index += 1
else:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S") ,"没有查找到提交按钮,跳过!")
close_contentWrapper()
continue
# time.sleep(10)
# 关闭按钮
close_x_btn = tab.eles('xpath://div[@class="flyoutPanelContent"]/span',timeout=5)
contentWrapper = tab.eles('xpath://div[@class="flyoutPanelContent"]/div[@class="contentWrapper"]',timeout=5)
if len(close_x_btn) > 0 and len(contentWrapper)>0 :
if len(contentWrapper[0].children()) > 0:
close_x_btn[0].wait.displayed(timeout=5, raise_err=False)
close_x_btn[0].click()
print("开始操作",submit_btn)
print("操作的ID 为:",submit_btn.attr('id'))
submit_btn.wait.enabled(timeout=5, raise_err=False)
submit_btn.click()
# 先直接等待contentWrapper出现不要进行复杂的元素查找
try:
# 使用简单的等待,避免触发页面事件
for check_count in range(10):
contentWrapper_exists = tab.eles('xpath://div[@class="contentWrapper"]', timeout=2)
if len(contentWrapper_exists) > 0:
print(f"【弹窗已出现】第{check_count+1}次检查")
time.sleep(2) # 再等待2秒让内容完全加载
break
time.sleep(1)
except Exception as e:
print(f"等待弹窗出现异常: {e}")
# wait_loaded(wait_type="content")
for j in range(4):
if j == 0:
timeout = 5
else:
timeout = 10
contentWrapper = tab.eles('xpath://div[@class="contentWrapper"]//h4',timeout=timeout)
print("标题的数量",len(contentWrapper))
if len(contentWrapper) > 0:
contentWrapper[0].wait.displayed(timeout=20, raise_err=True)
break
else:
# 关闭按钮
close_contentWrapper()
time.sleep(2)
submit_btn.click()
need_input = tab.eles('xpath://div[@class="contentWrapper"]//kat-input[@data-testid="kat-input-dew:ump_epr_resgitration_number_title"]',timeout=5)
if len(need_input) > 0:
print("需要输入注册号,不符合操作要求,跳过...")
close_contentWrapper()
continue
target_title = tab.eles('xpath://div[@class="contentWrapper"]//section//h4[text()="警告和安全信息"]',timeout=10)
print(f"序号{index}【警告和安全信息】数量 ->",len(target_title))
if len(target_title) > 0:
is_Continue = False
for j in range(10):
not_start = tab.eles('xpath://div[@class="contentWrapper"]//div[@aria-label="安全证明"]//kat-label',timeout=2)
if len(not_start) > 0:
status_lebel = not_start[0].attr("text")
print(f"序号{index}【警告和安全信息】状态 ->",status_lebel)
if status_lebel and str(status_lebel).strip() != "" and status_lebel != "未开始":
print("安全证明-不是未开始状态,不需要操作")
close_contentWrapper()
continue
if status_lebel == "未开始":
is_Continue = True
break
else:
time.sleep(1)
# not_start = tab.eles('xpath://div[@class="contentWrapper"]//div[@aria-label="安全证明"]//kat-label[@text="未开始"]',timeout=10)
# if len(not_start) == 0:
# print("安全证明-不是未开始状态,不需要操作")
# close_contentWrapper()
# continue
print(f"序号{index}【警告和安全信息】是否需要继续 ->",is_Continue)
if not is_Continue:
continue
not_start = not_start[0]
not_start.wait.displayed(timeout=5, raise_err=False)
not_start.click()
wait_loaded()
check = tab.ele('xpath://div[@class="contentWrapper"]//kat-checkbox[@name="value"]',timeout=10)
check.wait.displayed(timeout=5, raise_err=False)
check.click()
save_btn = tab.ele('xpath://div[@class="contentWrapper"]//kat-button[@variant="primary"]',timeout=10)
save_btn.wait.enabled(timeout=5, raise_err=False)
save_btn.click()
wait_loaded()
close_btn = tab.ele('xpath://div[@class="contentWrapper"]//kat-button[@label="关闭"]',timeout=10)
close_btn.wait.displayed(timeout=5, raise_err=False)
close_btn.click()
print(f"序号{index}【警告和安全信息】操作完成")
continue
save_btn_ls = tab.eles('xpath://div[@class="contentWrapper"]//kat-button[@variant="primary"]',timeout=10)
print(f"序号{index}【选择保存】是否存在->",len(save_btn_ls))
if len(save_btn_ls) > 0:
option_selection_ls = tab.eles('xpath://div[@class="contentWrapper"]//div[@role="option"]',timeout=5)
for option in option_selection_ls:
option.click()
time.sleep(0.5)
save_btn_ls[0].wait.enabled(timeout=5, raise_err=False)
save_btn_ls[0].click()
wait_loaded()
close_contentWrapper()
print(f"序号{index}【选择保存】操作完成")
continue
submit_btn_ls = tab.eles('xpath://div[@id="ahd-product-policies-table"]/div//div[@data-testid="nextStepsMetricWrapper"]//kat-button[@label="提交"]',timeout=10)
if len(submit_btn_ls) > 0:
submit_btn_ls[0].click()
submit_btn_ls[0].wait.deleted(timeout=5, raise_err=False)
wait_loaded()
close_contentWrapper()
break
except Exception as e:
print(f"【handle_repair_product】处理修复商品信息异常", traceback.format_exc())
tab.refresh()
tab.wait.doc_loaded(timeout=60,raise_err=False)
time.sleep(2)
wait_loaded(wait_type="section")
# product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy"]')
product_policy = tab.eles('xpath://div[@data-testid="ahd-desktop-product-policy" and @class="ahd-product-policy-table-row-wrapper"]|//div[@class="ahd-product-policy-table-row-wrapper-nested" and @data-testid="ahd-desktop-product-policy"]')
print("修复商品信息处理完成!")
tab.close()
def run_page_action(self):
self.log(f"==================={self.mark_name}=======================")
self.log("开始执行...")
num = 0
retry_num = 0
max_retry_num = 5
while retry_num < max_retry_num: # 最多重试3次
# if num > 3: #测试
# return
# 等待加载完成
try:
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)
# 获取当前页码
try:
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
# 总页数
total_page = page_pamel[0].sr.eles('xpath:.//ul[@class="pages"]//span[@class="page__inner"][last()]')
if len(total_page) > 0:
total_page = total_page[-1].text
else:
total_page = 0
self.log(f"当前页码: {current_page} / 总页数: {total_page}")
except Exception as e:
self.log(f"获取页码失败:{e}")
# 保存当前URL用于失败重试时访问
try:
self.url = self.tab.url
self.log(f"获取当前的链接:{self.url}")
except Exception as e:
self.url = None
self.log(f"获取当前的链接失败:{e}")
sku_ls = self.tab.eles("xpath://div[@data-sku]",timeout=10)
self.log(f"获取到 {len(sku_ls)}")
if len(sku_ls) == 0:
retry_num += 1
print(f"没有获取到 ASIN 商品,重试{retry_num}/{max_retry_num}")
self.tab.refresh()
self.tab.wait.doc_loaded(raise_err=False, timeout=120)
if retry_num == max_retry_num:
raise RuntimeError(f"与页面的连接已断开,没有获取到 ASIN 商品,重试{retry_num}/{max_retry_num}")
continue
# for sku_ele in sku_ls[0:2]:
for sku_ele in sku_ls:
# solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]')
asin = sku_ele.ele('xpath:.//div[contains(@class,"JanusSplitBox-module__container")]//div[contains(@class,"JanusSplitBox-module__panel--") and contains(string(.),"ASIN")]/..//div[last()]').text
self.log(f"*************** {asin} *****************")
self.log(f"ASIN {asin} 存在问题,正在点击解决...")
try:
if asin in self.already_asin:
self.log(f"{asin} 已经处理过了,跳过")
continue
xp_ls = [f'.//kat-link[@label="{i}"]' for i in ["解决商品信息问题","修复被禁止显示的商品","解决商品移除风险","查看冻结原因"]]
xp = "|".join(xp_ls)
# solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]|.//kat-link[@label="修复被禁止显示的商品"]|.//kat-link[@label="解决商品移除风险"]')
solve_problem = sku_ele.eles(f'xpath:{xp}')
if len(solve_problem) == 0:
# //a[text0=”添加缺失的商品详情”]
lack_info = sku_ele.eles('xpath:.//a[text()="添加缺失的商品详情"]')
if len(lack_info) > 0:
self.log(f"{asin} 需要添加缺失的商品详情,跳过...")
yield (asin,"添加缺失的商品详情")
continue
self.log(f"{asin},没有处理入口,不处理")
yield (asin,"没有处理入口,不处理")
self.already_asin.add(asin)
continue
solve_problem[0].click()
label = solve_problem[0].attr('label')
self.log(f"【当前类别】{label}")
action_panel = self.tab.ele('xpath://kat-panel-wrapper[@data-testid="kat-panel-wrapper-ActionPanelContent"]')
action_panel.wait.displayed(timeout=5, raise_err=False)
self.wait_loaded()
time.sleep(1)
# 如果存在“请求批准”按钮,则直接返回跳过
request_approval_btn = self.tab.eles('xpath://kat-panel-wrapper[@data-testid="kat-panel-wrapper-ActionPanelContent"]//div[@data-testid="section-header" and contains(string(.),"移除")]',timeout=5)
if len(request_approval_btn) > 0:
# 新增"查看冻结原因" 的情况下,需要点击修复商品详情
if "查看冻结原因" in label:
repair_produ_btn_xp = '//kat-panel-wrapper[@data-testid="kat-panel-wrapper-ActionPanelContent"]//kat-button[@variant="secondary" and @class="action-button" and contains(string(.),"修复商品信息")]'
repair_produ_btn = self.tab.eles(f'xpath:{repair_produ_btn_xp}',timeout=5)
self.log(f"{asin}【查看冻结原因】修复商品信息数量,{len(repair_produ_btn)}")
if len(repair_produ_btn) > 0:
self.log(f"{asin}【查看冻结原因】修复商品信息数量,开始操作")
self.clear_tab()
repair_produ_btn[0].click()
time.sleep(3)
# 获取最新的tab
new_tab = self.browser.latest_tab
if isinstance(new_tab, str):
new_tab = self.browser.get_tab(id_or_num=new_tab)
self.handle_repair_product(new_tab,headle_type="查看冻结原因")
else:
self.log(f"{i}【查看冻结原因】修复商品信息没有,不需要操作")
else:
self.log(f"ASIN {asin} 存在请求批准按钮,不需要处理,跳过...")
panel_close_btn = self.tab.ele('xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]',
timeout=5).sr('xpath:.//button[@class="close"]')
panel_close_btn.click()
action_panel.wait.deleted(timeout=3, raise_err=False)
yield (asin,"请求批准")
self.already_asin.add(asin)
continue
# 无需采取任何操作
not_operate = self.tab.eles('xpath://kat-alert[contains(@description,"如果您之前已提交更改,则这些更改当前正在处理中") and not(@dismissed)]',timeout=3)
if len(not_operate) > 0:
self.log(f"ASIN {asin} 无需采取任何操作,跳过...")
yield (asin,"无需操作")
self.already_asin.add(asin)
continue
# 解决商品信息违规问题 按钮
kat_box_ls = action_panel.eles('xpath:.//kat-box[@variant="white"]',timeout=5)
# if len(kat_box_ls) == 0:
# print(f"ASIN {asin} 的操作面板中未找到选项,跳过...")
# continue
self.log(f"ASIN {asin}需要处理的有:{len(kat_box_ls)}个问题.")
# 解决商品信息违规问题 按钮,有多少个都要处理
for i in range(len(kat_box_ls)):
self.log(f"开始处理第{i}个问题")
if len(kat_box_ls) <= i:
self.log(f"ASIN {asin} 的操作面板中选项数量发生变化,停止处理...")
break
kat_box_ls[i].click()
time.sleep(0.5)
self.wait_loaded()
# TODO 增加 “出现警告和安全信息时候下滑” 的情况
safe_handle_ls = self.tab.eles('xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]//div[@aria-label="安全证明"]//kat-label[@text="未开始"]',
timeout=15)
if len(safe_handle_ls) > 0:
self.log("存在安全证明,正在处理...")
safe_handle_ls[0].click()
time.sleep(0.5)
self.wait_loaded()
check_box = self.tab.ele('xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]//kat-checkbox')
check_box.click()
else:
problem_selection_ls = self.tab.eles(
'xpath://kat-panel-wrapper[@data-testid="kat-panel-wrapper-ActionPanelContent"]//div[@data-testid="registry-list"]//div[@data-testid="registry"]',
timeout=10)
self.log(f"待选择数量,{len(problem_selection_ls)}")
for problem_selection in problem_selection_ls:
problem_selection.click()
print("点击选择完成")
time.sleep(0.5)
# 点击保存
confirm_btn = self.tab.eles('xpath://kat-panel-wrapper[@data-testid="kat-panel-wrapper-ActionPanelContent"]//kat-button[@variant="primary"]')
if len(confirm_btn) > 0:
confirm_btn[0].wait.displayed(timeout=3, raise_err=False)
confirm_btn[0].click()
self.wait_loaded()
# 等待关闭按钮出现
try:
close_btn = self.tab.eles('xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]//kat-button[@label="关闭"]',
timeout=10)
if len(close_btn)>0:
close_btn[0].click()
except Exception as e:
self.log(f"点击关闭按钮失败,{e}")
kat_box_ls = action_panel.eles('xpath:.//kat-box[@variant="white"]', timeout=5)
# 修复商品信息
repair_product = action_panel.eles('xpath:.//kat-button[@class="action-button" and not(@variant="secondary")]',timeout=5)
if len(repair_product) > 0:
self.log(f"ASIN {asin} 存在修复商品信息按钮,正在点击...")
self.clear_tab()
repair_product[0].click()
time.sleep(3)
# 获取最新的tab
new_tab = self.browser.latest_tab
if isinstance(new_tab, str):
new_tab = self.browser.get_tab(id_or_num=new_tab)
self.handle_repair_product(new_tab)
yield (asin,"处理完成")
self.already_asin.add(asin)
self.log(f"*********************{asin} 处理完成*************************")
try:
# 全部操作完成,关闭
panel_close_btn= self.tab.ele('xpath://kat-panel[@data-testid="kat-panel-ActionPanelContent"]',
timeout=5).sr('xpath:.//button[@class="close"]')
panel_close_btn.click()
action_panel.wait.deleted(timeout=3, raise_err=False)
self.clear_tab()
except Exception as e:
self.log(f"关闭操作面板失败,{e}")
except Exception as e:
yield (asin, "处理失败")
self.already_asin.add(asin)
self.log(f"{asin} 操作出错:{traceback.format_exc()}")
# 判断是否存在需要翻页的情况
page_pamel = self.tab.eles('xpath://kat-pagination',timeout=5)
if len(page_pamel) == 0:
break
next_page_btn = page_pamel[0].sr('xpath:.//span[@part="pagination-nav-right"]')
class_str = next_page_btn.attr('class')
if "end" in class_str:
break
next_page_btn.click()
num += 1
self.log(f"【程序计算】正在翻页,已翻 {num} 页...")
# already_asin = set()
except Exception as e:
self.log(f"处理审批操作异常,{e}")
traceback.print_exc()
retry_num += 1
self.tab.refresh()
self.tab.wait.doc_loaded(raise_err=False,timeout=120)
class ApproveTask(TaskBase):
"""审批任务处理类:负责处理产品风险审批任务"""
task_name = "产品风险审批-TASK"
def process_task(self, task_data: dict):
"""处理审批任务主入口
Args:
task_data: 任务数据
"""
try:
data = task_data.get("data", {})
task_id = data.get("taskId")
items = data.get("items", [])
country_codes = data.get("country_codes", [])
risk_listing_filter = data.get("risk_listing_filter", "")
# 测试
# risk_listing_filter = "DetailPageRemoved" #data.get("risk_listing_filter", "")
if not task_id:
self.log("任务ID为空跳过", "WARNING")
return
if not items:
self.log("店铺列表为空,跳过", "WARNING")
return
if not country_codes:
self.log("国家列表为空,跳过", "WARNING")
return
self.log(f"开始处理审批任务 {task_id},共 {len(items)} 个店铺,{len(country_codes)} 个国家")
runing_task[task_id] = {
"status": "running",
"start_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"total_shops": len(items),
"processed_shops": 0,
"total_countries": len(country_codes) * len(items),
"processed_countries": 0,
"total_asins": 0,
"processed_asins": 0,
"success_count": 0,
"failed_count": 0,
"stop_requested": False
}
# 遍历处理每个店铺
for idx, shop_item in enumerate(items, 1):
# 检查是否收到暂停请求
if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
self.log(f"检测到任务 {task_id} 的暂停请求,停止处理", "WARNING")
runing_task[task_id]["status"] = "stopped"
return
shop_name = shop_item.get("shopName", "未知店铺")
self.log(f"[{idx}/{len(items)}] 开始处理店铺: {shop_name}")
show_notification(f"开始处理店铺: {shop_name}", "info")
try:
self.process_shop(shop_item, country_codes, task_id,risk_listing_filter)
# 更新已处理店铺数
if task_id in runing_task:
runing_task[task_id]["processed_shops"] += 1
except Exception as e:
self.log(f"处理店铺 {shop_name} 失败: {str(e)}", "ERROR")
self.log(traceback.format_exc(), "ERROR")
# 更新任务状态
if task_id in runing_task:
if runing_task[task_id].get("stop_requested", False):
runing_task[task_id]["status"] = "stopped"
self.log(f"任务 {task_id} 已被暂停!")
else:
runing_task[task_id]["status"] = "completed"
self.log(f"任务 {task_id} 处理完成!")
except Exception as e:
self.log(f"任务处理失败: {traceback.format_exc()}", "ERROR")
if task_id:
if task_id in runing_task:
runing_task[task_id]["status"] = "failed"
runing_task[task_id]["error"] = str(e)
def process_shop(self, shop_item: dict, country_codes: list, task_id: int, risk_listing_filter: str):
"""处理单个店铺
Args:
shop_item: 店铺信息
country_codes: 国家代码列表
task_id: 任务ID
risk_listing_filter: 风险商品筛选条件
"""
shop_name = shop_item.get("shopName", "未知店铺")
company_name = shop_item.get("companyName", "")
if not company_name:
self.log(f"店铺 {shop_name} 的公司名称为空,跳过", "WARNING")
return
if task_id in runing_task:
runing_task[task_id]["current_shop"] = shop_name
# 将店铺添加到正在执行中的店铺列表
start_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
runing_shop[shop_name] = start_time
self.log(f"店铺 {shop_name} 已添加到执行列表,账号: {company_name},开始时间: {start_time}")
# 店铺打开重试最多3次
driver = None
max_retries = 3
iskill = False
# 处理每个国家
for country_code in country_codes:
# 检查是否收到暂停请求
if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
self.log(f"检测到任务 {task_id} 的暂停请求,停止处理国家", "ERROR")
driver.close_store()
break
# 打开店铺
driver = self.open_shop(cls=AmzoneApprove,max_retries=max_retries, company_name=company_name,
shop_name=shop_name, iskill=iskill)
if driver is None:
self.log(f"任务 {task_id} 启动店铺失败,结束任务", "ERROR")
for country_code in country_codes:
self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
return
current_url = None
max_retries = 200 #只要没有完成,一直重试
for _ in range(max_retries):
try:
self.process_country(driver, country_code, task_id, shop_name,risk_listing_filter,target_url=current_url)
driver.reset_already_asin()
driver.close_store()
break
except Exception as e:
self.log(f"处理国家 {country_code} 失败: {str(e)}", "ERROR")
self.log(traceback.format_exc(), "ERROR")
if "与页面的连接已断开" in str(e):
iskill = True
current_url = driver.url
# 更新已处理国家数
if task_id in runing_task:
runing_task[task_id]["processed_countries"] += 1
self.log(f"{task_id}任务处理完成")
# 最后回传,标记完成
try:
self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
driver.close_store() #完成退出店铺
except Exception as e:
self.log(f"回传结果失败: {str(e)}", "ERROR")
# 从正在执行中的店铺列表中移除
if shop_name in runing_shop:
del runing_shop[shop_name]
self.log(f"店铺 {shop_name} 已从执行列表中移除")
def process_country(self, driver: AmzoneApprove, country_code: str, task_id: int,
shop_name: str, risk_listing_filter: str,target_url:str=None):
"""
处理单个国家的审批任务
"""
# 转换国家代码为中文名称
country_name = self.country_info.get(country_code, country_code)
info_mes = f"开始处理国家: {country_name} ({country_code})"
self.log(info_mes)
show_notification(info_mes, "info")
# 更新当前处理的国家
if task_id in runing_task:
runing_task[task_id]["current_country"] = country_name
# 切换国家最多重试3次
max_retries = 5
switch_success, switch_success_pg, search_success,sku_ls = self.action_init(driver, country_name, shop_name,
risk_listing_filter)
# 如果切换失败,直接返回
if not switch_success or not switch_success_pg or not search_success:
error_message = f"切换到国家({switch_success})/库存页面({switch_success_pg})/搜索筛选到指定选项({search_success}) {country_name} 失败,已重试 {max_retries} 次,跳过该国家"
self.log(error_message, "ERROR")
if task_id in runing_task:
runing_task[task_id]["processed_countries"] += 1
return
# # 如果没有需要审批的商品,直接返回
if len(sku_ls) == 0:
self.log(f"国家 {country_name} 没有需要审批的商品")
if task_id in runing_task:
runing_task[task_id]["processed_countries"] += 1
return
self.log(f"国家 {country_name}{len(sku_ls)} 个需要审批的商品,开始处理...")
if target_url is not None: #从失败的链接继续
self.log(f"开始访问:{target_url}")
driver.tab.get(url=target_url)
driver.tab.wait.doc_loaded(timeout=60,raise_err=False)
result = []
# 处理所有需要审批的商品通过yield获取结果
for asin, status in driver.run_page_action():
# 检查是否收到暂停请求
if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
self.log(f"检测到任务 {task_id} 的暂停请求停止处理ASIN", "WARNING")
break
self.log(f"ASIN {asin} 处理结果: {status}")
if status in ["请求批准", "添加缺失的商品详情"]:
country_data = {
"status": "",
"shopName": shop_name,
"productAsinSku": "",
"done": False,
"removeAsin": asin,
"removeStatus": status
}
else:
country_data = {
"status": status,
"shopName": shop_name,
"productAsinSku": asin,
"done": False,
"removeAsin": "",
"removeStatus": ""
}
result.append(country_data)
if len(result) > 10:
self.post_result_batch(task_id, shop_name, country_code, result)
result = []
if len(result) > 0:
self.post_result_batch(task_id, shop_name, country_code, result)
self.log(f"国家 {country_name} 处理完成")
def post_result(self, task_id: int, shop_name: str, country_code: str, asin: str, status: str,is_done: bool = False):
"""回传处理结果到API
Args:
task_id: 任务ID
shop_name: 店铺名称
country_code: 国家代码
asin: ASIN
status: 处理状态
"""
url = f"{DELETE_BRAND_API_BASE}/api/product-risk-resolve/tasks/{task_id}/result"
if status in ["请求批准","添加缺失的商品详情"]:
country_data = {
"status": "",
"shopName": shop_name,
"productAsinSku": "",
"done": is_done,
"removeAsin": asin,
"removeStatus": status
}
else:
country_data = {
"status": status,
"shopName": shop_name,
"productAsinSku": asin,
"done": is_done,
"removeAsin": "",
"removeStatus": ""
}
payload = {
"shops": [
{
"error": "",
"countries": {
country_code : [ country_data ]
},
"shopName": shop_name
}
]
}
max_retries = 3
for retry in range(max_retries):
try:
self.log("=====================================")
self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)")
self.log(f"回传URL: {url}")
self.log(f"回传数据: {payload}")
response = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=30,
verify=False
)
self.log(f"回传结果: {response.text}")
if response.status_code == 200:
self.log(f"结果回传成功: {asin} - {status}")
return
else:
self.log(f"结果回传失败,状态码: {response.status_code}", "WARNING")
print("=====================================")
except Exception as e:
self.log(f"调用API异常: {str(e)}", "ERROR")
print("=====================================")
# 如果还有重试机会,等待后继续
if retry < max_retries - 1:
time.sleep(2)
self.log(f"已达到最大重试次数,结果回传最终失败", "ERROR")
def post_result_batch(self, task_id: int, shop_name: str, country_code: str, country_data_ls:list):
"""
批量回传处理结果到API
"""
url = f"{DELETE_BRAND_API_BASE}/api/product-risk-resolve/tasks/{task_id}/result"
payload = {
"shops": [
{
"error": "",
"countries": {
country_code: country_data_ls
},
"shopName": shop_name
}
]
}
max_retries = 3
for retry in range(max_retries):
try:
self.log("=====================================")
self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)")
self.log(f"回传URL: {url}")
self.log(f"回传数据: {payload}")
response = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=30,
verify=False
)
self.log(f"回传结果: {response.text}")
if response.status_code == 200:
self.log(f"结果回传成功")
return
else:
self.log(f"结果回传失败,状态码: {response.status_code}", "WARNING")
print("=====================================")
except Exception as e:
self.log(f"调用API异常: {str(e)}", "ERROR")
print("=====================================")
# 如果还有重试机会,等待后继续
if retry < max_retries - 1:
time.sleep(2)
self.log(f"已达到最大重试次数,结果回传最终失败", "ERROR")
if __name__ == '__main__':
user_info = {
"company": "rongchuang123",
"username": "自动化_Robot",
"password": "#20zsg25"
}
shop_name = "郭亚芳"
country = "德国"
kill_process('v6')
driver = AmzoneApprove(user_info)
browser = driver.open_shop(shop_name)
sw_suc = driver.SwitchingCountries(country)
driver.SwitchPage()
risk_listing_filter = "DetailPageRemoved"
for _ in range(3):
try:
sku_ls = driver.search(filter_type=risk_listing_filter)
break
except Exception as e:
print(e)
driver.tab.refresh()
if len(sku_ls) > 0:
print("有数据,开始操作")
for asin, status in driver.run_page_action():
print(f"ASIN {asin} 的处理结果: {status}")
print("已完成操作")
# for i in range(0):
# print(i)