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

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

Browse Source 
modified:   app/amazon/__pycache__/del_brand.cpython-39.pyc
	modified:   app/amazon/__pycache__/main.cpython-39.pyc
	modified:   app/amazon/__pycache__/match_action.cpython-39.pyc
	modified:   app/amazon/__pycache__/tool.cpython-39.pyc
	modified:   app/amazon/approve.py
	modified:   app/amazon/del_brand.py
	modified:   app/amazon/main.py
	modified:   app/amazon/match_action.py
	modified:   app/amazon/tool.py
	modified:   app/assets/delete-brand.js
	modified:   app/new_web_source/delete-brand.html
This commit is contained in:
铭坤
2026-04-11 23:00:55 +08:00
parent 7589b4e418
commit ad2bb0a6ea
10 changed files with 133 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/amazon/__pycache__/approve.cpython-39.pyc
BIN
View File
Binary file not shown.
app/amazon/__pycache__/del_brand.cpython-39.pyc
BIN
View File
Binary file not shown.
app/amazon/__pycache__/main.cpython-39.pyc
BIN
View File
Binary file not shown.
app/amazon/__pycache__/match_action.cpython-39.pyc
BIN
View File
Binary file not shown.
app/amazon/__pycache__/tool.cpython-39.pyc
BIN
View File
Binary file not shown.
app/amazon/approve.py
+33 -18
View File
@@ -7,6 +7,8 @@ from config import runing_task, runing_shop
from datetime import datetime
from amazon.del_brand import AmamzonBase, kill_process
from amazon.tool import get_shop_info,show_notification
# 导入 webview 用于前端通知
try:
import webview
@@ -92,17 +94,18 @@ class AmzoneApprove(AmamzonBase):
print("等待加载中消失出错", e)
def clear_tab(self):
all_tab = self.browser.get_tabs(title="亚马逊")
close_tab = []
for tab in all_tab:
tab_id = tab if isinstance(tab,str) else tab.tab_id
if self.tab.tab_id == tab:
continue
close_tab.append(tab)
print("需要关闭的标签页",close_tab)
print("当前操作的tab_id",self.tab.tab_id)
self.browser.close_tabs(close_tab)
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):
def wait_loaded():
# 等待加载完成
@@ -118,7 +121,7 @@ class AmzoneApprove(AmamzonBase):
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:
print("未找到提交按钮，可能页面未加载完成或者页面结构发生变化")
return
# return
for _ in range(3):
try:
# todo 遍历
@@ -265,6 +268,7 @@ class AmzoneApprove(AmamzonBase):
action_panel.wait.displayed(timeout=5, raise_err=False)
self.wait_loaded()
time.sleep(1)
try:
# 如果存在“请求批准”按钮，则直接返回跳过
@@ -342,7 +346,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" and not(@variant="secondary")]',timeout=5)
if len(repair_product) > 0:
print(f"ASIN {asin} 存在修复商品信息按钮，正在点击...")
self.clear_tab()
@@ -556,11 +560,9 @@ class ApproveTask:
if not task_id:
self.log("任务ID为空，跳过", "WARNING")
return
if not items:
self.log("店铺列表为空，跳过", "WARNING")
return
if not country_codes:
self.log("国家列表为空，跳过", "WARNING")
return
@@ -673,16 +675,28 @@ class ApproveTask:
if browser and browser != "店铺不存在":
self.log(f"成功打开店铺 {shop_name}")
break
# break
else:
self.log(f"打开店铺失败: {browser}", "WARNING")
driver = None
continue
# 判断是否需要登录
need_login = driver.need_login()
print("【是否需要登录】：",need_login)
if need_login:
self.log(f"店铺 {shop_name} 需要登录，正在登录...")
password = ""
# 获取店铺凭证
response = get_shop_info(shop_name)
print("【获取店铺凭证返回】：",response.text)
shop_data = response.json()
if not shop_data:
mes = f"获取店铺凭证失败，响应数据: {shop_data.get('message', '未知错误')}"
self.log(mes, "ERROR")
show_notification(mes, "ERROR")
continue
password = shop_data["data"]["password"]
login_success = driver.login(password)
if login_success:
@@ -697,7 +711,8 @@ class ApproveTask:
else:
self.log(f"店铺 {shop_name} 登录失败", "WARNING")
driver = None
else:
break
except Exception as e:
import traceback
self.log(f"打开店铺异常: {traceback.format_exc()}", "INFO")
app/amazon/del_brand.py
+13 -6
View File
@@ -489,7 +489,9 @@ class AmamzonBase(ZiniaoDriver):
处理流程：
"""
need_login_ele = self.tab.eles('xpath://h1[@class="a-spacing-small"]',timeout=5)
time.sleep(3) # 等待页面可能的登录元素加载,避免跳转等等
self.tab.wait.doc_loaded(timeout=30, raise_err=False)
need_login_ele = self.tab.eles('xpath://h1[@class="a-spacing-small"]|//span[contains(text(),"登录")]',timeout=5)
if len(need_login_ele) >0:
print("检测到需要登录元素")
return True
@@ -497,13 +499,15 @@ class AmamzonBase(ZiniaoDriver):
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()
pwd_input = self.tab.eles('xpath://input[@type="password"]',timeout=10)
if len(pwd_input) > 0:
pwd_input[0].input(password)
submit_btn = self.tab.eles('xpath://input[@id="signInSubmit"]',timeout=10)
if len(submit_btn) > 0:
submit_btn[0].click()
self.tab.wait.doc_loaded()
opt_code_input = self.tab.ele('xpath://input[@name="otpCode"]',timeout=10)
opt_code_input = self.tab.ele('xpath://input[@name="otpCode"]',timeout=60)
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() != "":
@@ -512,6 +516,9 @@ class AmamzonBase(ZiniaoDriver):
submit_btn.click()
self.tab.wait.doc_loaded()
return True
time.sleep(1)
submit_btn = self.tab.ele('xpath://input[@id="auth-signin-button"]',timeout=10)
submit_btn.click()
except Exception as e:
print("登录过程中发生异常:" + traceback.format_exc())
return False
app/amazon/main.py
+19 -3
View File
@@ -10,6 +10,9 @@ from amazon.approve import ApproveTask
from amazon.match_action import MatchTak
from amazon.tool import get_shop_info,show_notification
class TaskMonitor:
"""任务监控器：负责监控队列并执行品牌删除任务"""
@@ -64,13 +67,13 @@ class TaskMonitor:
if task_type == "delete-brand-run":
# 提交删除品牌任务到线程池
self.log(f"接收到删除品牌任务，提交到线程池处理...")
self.log(f"接收到【删除品牌】任务，提交到线程池处理...")
future = self.executor.submit(self._process_task_wrapper, task_data)
futures.append(future)
elif task_type in task_type_info:
# 提交产品风险审批任务到线程池
self.log(f"接收到产品风险审批任务，提交到线程池处理...")
self.log(f"接收到任务，提交到线程池处理...")
future = self.executor.submit(self._process_approve_task_wrapper, task_data, task_type_info[task_type])
futures.append(future)
@@ -242,10 +245,22 @@ class TaskMonitor:
driver = None
# 判断是否需要登录
driver.tab.wait.doc_loaded(timeout=30, raise_err=False) # 等待页面加载，避免过早判断登录状态
need_login = driver.need_login()
print("【是否需要登录】：",need_login)
if need_login:
self.log(f"店铺 {shop_name} 需要登录，正在登录...")
password = ""
# 获取店铺凭证
response = get_shop_info(shop_name)
print("【获取店铺凭证返回】：",response.text)
shop_data = response.json()
if not shop_data:
mes = f"获取店铺凭证失败，响应数据: {shop_data.get('message', '未知错误')}"
self.log(mes, "ERROR")
show_notification(mes, "ERROR")
continue
password = shop_data["data"]["password"]
login_success = driver.login(password)
if login_success:
@@ -611,6 +626,7 @@ class TaskMonitor:
verify=False # 忽略SSL证书验证
)
print("【结果提交】：",payload)
print("【结果提交返回】：",response.text)
if response.status_code == 200:
self.log(f"结果回传成功: {url}")
app/amazon/match_action.py
+29 -8
View File
@@ -7,7 +7,7 @@ from config import runing_task, runing_shop
from datetime import datetime
from amazon.del_brand import AmamzonBase, kill_process
from amazon.tool import show_notification
from amazon.tool import show_notification,get_shop_info
class AmzoneMatchAction(AmamzonBase):
@@ -273,6 +273,8 @@ class MatchTak:
try:
self.process_shop(shop_item, country_codes, task_id, risk_listing_filter, user_id, stage_index, final_stage)
# 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
@@ -299,6 +301,7 @@ class MatchTak:
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):
def process_shop(self, shop_item: dict, country_codes: list, task_id: int, risk_listing_filter: str, user_id=None, stage_index=None, final_stage: bool = True):
"""处理单个店铺
@@ -350,16 +353,27 @@ class MatchTak:
if browser and browser != "店铺不存在":
self.log(f"成功打开店铺 {shop_name}")
break
else:
self.log(f"打开店铺失败: {browser}", "WARNING")
driver = None
continue
# 判断是否需要登录
need_login = driver.need_login()
print("【是否需要登录】：",need_login)
if need_login:
self.log(f"店铺 {shop_name} 需要登录，正在登录...")
password = ""
# 获取店铺凭证
response = get_shop_info(shop_name)
print("【获取店铺凭证返回】：",response.text)
shop_data = response.json()
if not shop_data:
mes = f"获取店铺凭证失败，响应数据: {shop_data.get('message', '未知错误')}"
self.log(mes, "ERROR")
show_notification(mes, "ERROR")
continue
password = shop_data["data"]["password"]
login_success = driver.login(password)
if login_success:
@@ -374,6 +388,8 @@ class MatchTak:
else:
self.log(f"店铺 {shop_name} 登录失败", "WARNING")
driver = None
else:
break
except Exception as e:
import traceback
@@ -409,7 +425,9 @@ class MatchTak:
import traceback
self.log(f"处理国家 {country_code} 失败: {str(e)}", "ERROR")
self.log(traceback.format_exc(), "ERROR")
# 最后回传，标记完成
try:
# self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
if final_stage:
self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
else:
@@ -564,7 +582,6 @@ class MatchTak:
runing_task[task_id]["processed_countries"] += 1
self.log(f"国家 {country_name} 处理完成")
def post_stage_finished(self, task_id: int, user_id, stage_index):
import requests
from config import DELETE_BRAND_API_BASE
@@ -639,8 +656,8 @@ class MatchTak:
max_retries = 3
for retry in range(max_retries):
try:
print("================回传处理结果====================")
self.log(f"尝试回传结果 (第{retry + 1}/{max_retries} 次)")
print("================【匹配】=====================")
self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)")
self.log(f"回传URL: {url}")
self.log(f"回传数据: {payload}")
response = requests.post(
@@ -653,9 +670,10 @@ class MatchTak:
self.log(f"回传结果: {response.text}")
data = response.json() if response.text else {}
if response.status_code == 200 and isinstance(data, dict) and data.get("success"):
self.log(f"回传结果成功: {asin} - {status}")
self.log(f"结果回传成功: {asin} - {status}")
return
self.log(f"回传结果失败，状态码: {response.status_code}", "WARNING")
else:
self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING")
print("=====================================")
except Exception as e:
@@ -666,7 +684,10 @@ class MatchTak:
if retry < max_retries - 1:
time.sleep(2)
self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR")
raise RuntimeError("已达到最大重试次数，结果回传最终失败")
if __name__ == "__main__":
user_info = {
app/amazon/tool.py
+39
View File
@@ -5,6 +5,9 @@ except ImportError:
webview = None
import requests
from urllib.parse import quote
def show_notification(message: str, message_type: str = "error"):
"""显示 pywebview 顶层通知（5秒后自动消失）
@@ -75,3 +78,39 @@ def show_notification(message: str, message_type: str = "error"):
print(f"显示通知失败: {str(e)}")
def get_shop_info(shop_name: str, base_url: str = "http://8.136.19.173:18080") -> requests.Response:
"""
调用商铺凭证接口
:param shop_name: 商铺名称，将自动进行 URL 编码
:param base_url: API 基础地址，默认从 curl 中提取
:return: requests.Response 对象
"""
url = f"{base_url}/api/admin/shop-manages/credential"
params = {"shopName": shop_name}
headers = {
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Pragma": "no-cache",
"Referer": "http://8.136.19.173:18080/doc.html",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36",
"X-Internal-Token": "59c6691199917a2095827b5188502029",
}
response = requests.get(
url,
params=params,
headers=headers,
verify=False,
timeout=30
)
return response