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

new file: app/amazon/__pycache__/del_brand.cpython-39.pyc

Browse Source 
new file:   app/amazon/__pycache__/main.cpython-39.pyc
	modified:   app/amazon/del_brand.py
	modified:   app/amazon/main.py
	modified:   app/blueprints/__pycache__/main.cpython-39.pyc
	modified:   app/blueprints/main.py
	modified:   app/config.py
	modified:   app/main.py
	new file:   "app/web_source/brand-\346\227\247.html"
	modified:   app/web_source/brand.html
This commit is contained in:
铭坤
2026-04-02 00:03:16 +08:00
parent 63043c56d4
commit a0624221de
8 changed files with 274 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
app/amazon/__pycache__/del_brand.cpython-39.pyc Normal file
View File

Binary file not shown.

BIN
app/amazon/__pycache__/main.cpython-39.pyc Normal file
View File

Binary file not shown.

16
app/amazon/del_brand.py
View File

@@ -233,7 +233,7 @@ class ZiniaoDriver:
# 获取店铺列表
shop_ls = self.get_browser_list()
self.store_id = None
# print(shop_ls)
print(shop_ls)
for shop in shop_ls:
if shop.get("browserName") == shop_name:
self.store_id = shop.get('browserOauth')
@@ -456,16 +456,20 @@ class AmazoneDriver(ZiniaoDriver):
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)
search_input.input(asin,clear=True)
sku_ls = []
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)
# load_ele.wait.hidden(timeout=3, raise_err=False)
load_ele.wait.deleted(timeout=3, raise_err=False)
time.sleep(1)
sku_ls = self.tab.eles("//div[@data-sku]")
sku_ls = self.tab.eles("xpath://div[@data-sku]")
if len(sku_ls) > 0:
break
return sku_ls
def del_action(self, sku_ele):
@@ -473,7 +477,7 @@ class AmazoneDriver(ZiniaoDriver):
dropdown.click()
time.sleep(1)
del_btn = dropdown.ele("xpath:.//button[@role='menuitem' and text()='删除商品信息']")
del_btn = dropdown.sr("xpath:.//button[@role='menuitem' and @data-action='DeleteListing']")
del_btn.wait.displayed(raise_err=False)
del_btn.click()

293
app/amazon/main.py
View File

@@ -3,6 +3,7 @@ import traceback
import requests
from datetime import datetime
from typing import Dict, Any, List
from concurrent.futures import ThreadPoolExecutor, as_completed
from config import JSON_TASK_QUEUE, runing_task, runing_shop, DELETE_BRAND_API_BASE, ZN_COMPANY, ZN_USERNAME, ZN_PASSWORD
from amazon.del_brand import AmazoneDriver, kill_process
@@ -20,6 +21,11 @@ class TaskMonitor:
}
self.chunk_index = 1 # 当前处理的分块索引
self.max_workers = 5 # 线程池最大线程数
self.executor = None # 线程池执行器
# 在提交新任务前杀掉旧进程（确保环境干净）
kill_process("v6")
def log(self, message: str, level: str = "INFO"):
"""日志输出
@@ -32,28 +38,60 @@ class TaskMonitor:
print(f"[{timestamp}] [{level}] {message}")
def start(self):
"""启动任务监控（阻塞式循环）"""
self.log("任务监控器启动，开始监听队列...")
"""启动任务监控（使用线程池处理任务）"""
self.log(f"任务监控器启动，开始监听队列... (线程池大小: {self.max_workers})")
while self.running:
try:
# 阻塞式获取任务，避免CPU空转
task_data = JSON_TASK_QUEUE.get(block=True, timeout=1)
# 创建线程池
self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
futures = [] # 保存所有提交的任务Future对象
# 检查任务类型
task_type = task_data.get("type", "")
if task_type != "delete-brand-run":
self.log(f"未知任务类型: {task_type}，跳过", "WARNING")
continue
try:
while self.running:
try:
# 使用较长超时时间等待任务，减少空等待异常
# 超时后继续循环检查self.running状态，避免卡死
task_data = JSON_TASK_QUEUE.get(block=True, timeout=30)
# 处理任务
self.log(f"接收到删除品牌任务，开始处理...")
self.process_task(task_data)
# 检查任务类型
task_type = task_data.get("type", "")
if task_type != "delete-brand-run":
self.log(f"未知任务类型: {task_type}，跳过", "WARNING")
continue
# 提交任务到线程池
self.log(f"接收到删除品牌任务，提交到线程池处理...")
future = self.executor.submit(self._process_task_wrapper, task_data)
futures.append(future)
# 清理已完成的future对象，避免内存累积
futures = [f for f in futures if not f.done()]
except Exception as e:
# 静默处理队列为空的超时，只记录真正的异常
if "Empty" not in str(e):
self.log(f"任务监控异常: {str(e)}", "ERROR")
# 队列为空时不需要额外等待，直接继续循环
finally:
# 关闭监控时，等待所有任务完成
self.log("正在关闭任务监控器，等待所有任务完成...")
if self.executor:
self.executor.shutdown(wait=True)
self.log("所有任务已完成，监控器已关闭")
def _process_task_wrapper(self, task_data: Dict[str, Any]):
"""任务处理包装器（用于线程池调用）
Args:
task_data: 任务数据
"""
try:
self.log(f"线程 {id(task_data)} 开始处理任务...")
self.process_task(task_data)
self.log(f"线程 {id(task_data)} 任务处理完成")
except Exception as e:
self.log(f"线程 {id(task_data)} 任务处理异常: {traceback.format_exc()}", "ERROR")
except Exception as e:
if "Empty" not in str(e): # 忽略队列为空的超时异常
self.log(f"任务处理异常: {traceback.format_exc()}", "ERROR")
time.sleep(0.1) # 短暂等待避免异常循环
def process_task(self, task_data: Dict[str, Any]):
"""处理单个任务
@@ -222,14 +260,41 @@ class TaskMonitor:
self.log(f"开始处理国家: {country}，共 {len(items)} 个ASIN")
self.update_task_status(task_id, current_country=country)
# 切换国家
try:
switch_success = driver.SwitchingCountries(country)
if not switch_success:
self.log(f"切换到国家 {country} 失败，跳过该国家", "ERROR")
return
except Exception as e:
self.log(f"切换国家 {country} 异常: {str(e)}", "ERROR")
# 切换国家，最多重试3次
max_retries = 3
switch_success = False
for retry in range(max_retries):
try:
self.log(f"尝试切换到国家 {country} (第 {retry + 1}/{max_retries} 次)")
# 如果不是第一次尝试，先刷新页面
if retry > 0:
self.log("重试前刷新页面...")
try:
driver.tab.refresh()
time.sleep(3)
except Exception as e:
self.log(f"刷新页面失败: {str(e)}", "WARNING")
switch_success = driver.SwitchingCountries(country)
if switch_success:
self.log(f"成功切换到国家 {country}")
break
else:
self.log(f"切换到国家 {country} 失败", "WARNING")
except Exception as e:
self.log(f"切换国家 {country} 异常: {str(e)}", "ERROR")
# 如果还有重试机会，等待后继续
if retry < max_retries - 1:
time.sleep(2)
# 如果切换失败，回传该国家所有ASIN为失败状态
if not switch_success:
self.log(f"切换到国家 {country} 失败，已重试 {max_retries} 次，将所有ASIN标记为失败", "ERROR")
self._report_all_asins_failed(country, items, task_id, shop_data)
return
# 切换到库存管理页面
@@ -238,6 +303,8 @@ class TaskMonitor:
self.log(f"已切换到库存管理页面")
except Exception as e:
self.log(f"切换页面失败: {str(e)}", "ERROR")
# 切换页面失败也回传所有ASIN为失败
self._report_all_asins_failed(country, items, task_id, shop_data)
return
# 处理每个ASIN
@@ -254,7 +321,7 @@ class TaskMonitor:
try:
self.log(f"[{idx}/{len(items)}] 处理ASIN: {asin_item.get('asin', '')}")
self.process_asin(driver, asin_item, country, task_id, result_id,
file_key, source_filename, total_rows, idx)
file_key, source_filename, total_rows)
except Exception as e:
asin = asin_item.get("asin", "未知")
self.log(f"处理ASIN {asin} 失败: {str(e)}", "ERROR")
@@ -283,49 +350,90 @@ class TaskMonitor:
self.update_task_status(task_id, current_asin=asin)
status = "失败"
max_retries = 3 # 最多重试3次
try:
# 搜索ASIN
sku_ls = driver.search(asin=asin)
self.log(f"搜索到 {len(sku_ls)} 个SKU")
for retry in range(max_retries):
try:
self.log(f"处理ASIN {asin} (第 {retry + 1}/{max_retries} 次)")
if len(sku_ls) == 0:
status = "查询不到"
self.log(f"ASIN {asin} 未找到商品", "WARNING")
else:
# 删除所有找到的SKU
success_count = 0
for sku in sku_ls:
# 如果不是第一次尝试，先刷新页面
if retry > 0:
self.log("重试前刷新页面...")
try:
suc = driver.del_action(sku)
if suc:
success_count += 1
self.log(f"SKU 删除成功")
else:
self.log(f"SKU 删除失败", "WARNING")
driver.tab.refresh()
time.sleep(3)
except Exception as e:
self.log(f"删除SKU异常: {str(e)}", "ERROR")
self.log(f"刷新页面失败: {str(e)}", "WARNING")
if success_count > 0:
status = "成功"
self.log(f"ASIN {asin} 删除成功 ({success_count}/{len(sku_ls)})")
# 更新成功计数
if task_id in runing_task:
runing_task[task_id]["success_count"] += 1
# 搜索ASIN
sku_ls = driver.search(asin=asin)
self.log(f"搜索到 {len(sku_ls)} 个SKU")
if len(sku_ls) == 0:
status = "查询不到"
self.log(f"ASIN {asin} 未找到商品", "WARNING")
break # 查询不到商品，无需重试
else:
status = "失败"
self.log(f"ASIN {asin} 所有SKU删除失败", "ERROR")
# 更新失败计数
# 删除所有找到的SKU，如果任何一个失败则重新开始整个流程
success_count = 0
all_success = True # 标记是否所有SKU都删除成功
total_sku_count = len(sku_ls)
for sku in sku_ls:
try:
suc = driver.del_action(sku)
if suc:
success_count += 1
self.log(f"SKU 删除成功 ({success_count}/{total_sku_count})")
else:
self.log(f"SKU 删除失败", "WARNING")
all_success = False
break # 任何一个失败，退出循环，准备重试整个流程
except Exception as e:
self.log(f"删除SKU异常: {str(e)}", "ERROR")
all_success = False
break # 发生异常，退出循环，准备重试整个流程
# 判断删除结果
if all_success and success_count == total_sku_count:
status = "成功"
self.log(f"ASIN {asin} 所有SKU删除成功 ({success_count}/{total_sku_count})")
# 更新成功计数
if task_id in runing_task:
runing_task[task_id]["success_count"] += 1
break # 全部成功，跳出重试循环
else:
# 有失败的SKU
if retry < max_retries - 1:
self.log(f"有SKU删除失败，准备重试整个流程... ({retry + 1}/{max_retries})")
time.sleep(2)
continue # 继续下一次重试
else:
# 所有重试都用完了
if success_count > 0:
status = "部分成功"
self.log(f"ASIN {asin} 部分SKU删除成功 ({success_count}/{total_sku_count})", "WARNING")
if task_id in runing_task:
runing_task[task_id]["success_count"] += 1
else:
status = "失败"
self.log(f"ASIN {asin} 所有SKU删除失败", "ERROR")
if task_id in runing_task:
runing_task[task_id]["failed_count"] += 1
except Exception as e:
status = "删除异常"
self.log(f"处理ASIN {asin} 异常: {str(e)}", "ERROR")
# 如果还有重试机会，继续重试
if retry < max_retries - 1:
self.log(f"发生异常，准备重试... ({retry + 1}/{max_retries})")
time.sleep(2)
else:
# 所有重试都失败了，更新失败计数
if task_id in runing_task:
runing_task[task_id]["failed_count"] += 1
except Exception as e:
status = "删除异常"
self.log(f"处理ASIN {asin} 异常: {str(e)}", "ERROR")
# 更新失败计数
if task_id in runing_task:
runing_task[task_id]["failed_count"] += 1
# 更新已处理ASIN计数
if task_id in runing_task:
runing_task[task_id]["processed_asins"] += 1
@@ -355,10 +463,65 @@ class TaskMonitor:
self.post_result(task_id, payload)
self.log(f"ASIN {asin} 结果已回传，状态: {status}")
self.chunk_index += 1
except Exception as e:
self.log(f"回传结果失败: {str(e)}", "ERROR")
def _report_all_asins_failed(self, country: str, items: List[Dict[str, Any]],
task_id: int, shop_data: Dict[str, Any]):
"""将国家下所有ASIN标记为失败并回传
Args:
country: 国家名称
items: ASIN列表
task_id: 任务ID
shop_data: 店铺数据
"""
self.log(f"开始回传国家 {country} 下的 {len(items)} 个ASIN失败状态")
file_key = shop_data.get("fileKey", "")
source_filename = shop_data.get("sourceFilename", "")
total_rows = shop_data.get("totalRows", 0)
for asin_item in items:
asin = asin_item.get("asin", "")
try:
# 更新失败计数
if task_id in runing_task:
runing_task[task_id]["failed_count"] += 1
runing_task[task_id]["processed_asins"] += 1
# 回传失败状态
payload = {
"submissionId": "",
"files": [{
"fileKey": file_key,
"sourceFilename": source_filename,
"chunkIndex": self.chunk_index,
"chunkTotal": total_rows,
"processedRows": self.chunk_index,
"totalRows": total_rows,
"currentCountry": country,
"currentAsin": asin,
"countries": [{
"country": country,
"items": [{
"asin": asin,
"status": "失败"
}]
}]
}]
}
self.post_result(task_id, payload)
self.chunk_index += 1
self.log(f"ASIN {asin} 失败状态已回传")
except Exception as e:
self.log(f"回传ASIN {asin} 失败状态时出错: {str(e)}", "ERROR")
def post_result(self, task_id: int, payload: Dict[str, Any]):
"""回传结果到API
@@ -366,7 +529,7 @@ class TaskMonitor:
task_id: 任务ID
payload: 结果数据
"""
url = f"{DELETE_BRAND_API_BASE}/newApi/api/delete-brand/tasks/{task_id}/result"
url = f"{DELETE_BRAND_API_BASE}/api/delete-brand/tasks/{task_id}/result"
try:
response = requests.post(
@@ -376,7 +539,7 @@ class TaskMonitor:
timeout=30,
verify=False # 忽略SSL证书验证
)
print("【结果提交】：",payload)
if response.status_code == 200:
self.log(f"结果回传成功: {url}")
else:
@@ -404,7 +567,7 @@ class TaskMonitor:
def main():
"""主函数：启动任务监控"""
"""主函数:启动任务监控"""
# 创建并启动任务监控器
monitor = TaskMonitor()
@@ -413,7 +576,7 @@ def main():
except KeyboardInterrupt:
monitor.log("接收到中断信号，正在停止...")
monitor.stop()
except Exception as e:
except Exception:
monitor.log(f"监控器异常退出: {traceback.format_exc()}", "ERROR")

BIN
app/blueprints/__pycache__/main.cpython-39.pyc
View File

Binary file not shown.

18
app/blueprints/main.py
View File

@@ -110,14 +110,16 @@ def proxy(path):
if key.lower() in ['host', 'content-length', 'connection']:
continue
headers[key] = value
try:
print("=============================")
print("target_url：",target_url)
print("params：",params)
print("data：",request.get_data())
print("=============================")
except Exception as e:
print("打印失败",e)
ignore_url = [f"{JAVA_API_BASE}/api/delete-brand/tasks/batch",f"{JAVA_API_BASE}/api/delete-brand/history"]
if target_url not in ignore_url:
try:
print("=============================")
print("target_url：",target_url)
print("params：",params)
print("data：",request.get_data())
print("=============================")
except Exception as e:
print("打印失败",e)
try:
# 使用流式请求

4
app/config.py
View File

@@ -21,9 +21,11 @@ client_name=os.getenv("client_name") + ".exe"
JAVA_API_BASE = os.getenv("java_api_base", "http://127.0.0.1:18080")
# 紫鸟浏览器配置
from urllib.parse import unquote
ZN_COMPANY = os.getenv("zn_company", "")
ZN_USERNAME = os.getenv("zn_username", "")
ZN_PASSWORD = os.getenv("zn_password", "")
ZN_USERNAME = unquote(ZN_USERNAME)
ZN_PASSWORD = os.getenv("zn_password", "#20zsg25")
# 删除品牌API配置
DELETE_BRAND_API_BASE = os.getenv("DELETE_BRAND_API_BASE", JAVA_API_BASE)

19
app/main.py
View File

@@ -7,6 +7,7 @@ import json
import sys
import shutil
import os
from amazon.main import TaskMonitor
os.makedirs(cache_path,exist_ok=True)
if not debug:
today = datetime.datetime.now().strftime("%Y_%m_%d")
@@ -301,11 +302,29 @@ def start_flask():
run_app(host='127.0.0.1', port=PORT)
def start_task_monitor():
"""启动任务监控器（在独立线程中运行）"""
monitor = TaskMonitor()
try:
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 任务监控线程已启动")
monitor.start()
except Exception:
import traceback
print(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 任务监控线程异常退出: {traceback.format_exc()}")
def main():
if not os.path.exists(cache_path):
os.makedirs(cache_path,exist_ok=True)
# 启动 Flask 服务
t = threading.Thread(target=start_flask, daemon=True)
t.start()
# 启动任务监控线程
monitor_thread = threading.Thread(target=start_task_monitor, daemon=True)
monitor_thread.start()
# 等待 Flask 启动
time.sleep(2)