diff --git a/app/amazon/__pycache__/del_brand.cpython-39.pyc b/app/amazon/__pycache__/del_brand.cpython-39.pyc new file mode 100644 index 0000000..656d682 Binary files /dev/null and b/app/amazon/__pycache__/del_brand.cpython-39.pyc differ diff --git a/app/amazon/__pycache__/main.cpython-39.pyc b/app/amazon/__pycache__/main.cpython-39.pyc new file mode 100644 index 0000000..d1ed31b Binary files /dev/null and b/app/amazon/__pycache__/main.cpython-39.pyc differ diff --git a/app/amazon/del_brand.py b/app/amazon/del_brand.py index cdce0fd..9bb2a94 100644 --- a/app/amazon/del_brand.py +++ b/app/amazon/del_brand.py @@ -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() diff --git a/app/amazon/main.py b/app/amazon/main.py index 17aeaaf..fd8472c 100644 --- a/app/amazon/main.py +++ b/app/amazon/main.py @@ -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对象 + + try: + while self.running: + try: + # 使用较长超时时间等待任务,减少空等待异常 + # 超时后继续循环检查self.running状态,避免卡死 + task_data = JSON_TASK_QUEUE.get(block=True, timeout=30) + + # 检查任务类型 + 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") - # 检查任务类型 - task_type = task_data.get("type", "") - if task_type != "delete-brand-run": - self.log(f"未知任务类型: {task_type},跳过", "WARNING") - continue - - # 处理任务 - self.log(f"接收到删除品牌任务,开始处理...") - self.process_task(task_data) - - 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,48 +350,89 @@ 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") - - if len(sku_ls) == 0: - status = "查询不到" - self.log(f"ASIN {asin} 未找到商品", "WARNING") - else: - # 删除所有找到的SKU - success_count = 0 - for sku in sku_ls: + for retry in range(max_retries): + try: + self.log(f"处理ASIN {asin} (第 {retry + 1}/{max_retries} 次)") + + # 如果不是第一次尝试,先刷新页面 + 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") - - 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 + self.log(f"刷新页面失败: {str(e)}", "WARNING") + + # 搜索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: @@ -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") diff --git a/app/blueprints/__pycache__/main.cpython-39.pyc b/app/blueprints/__pycache__/main.cpython-39.pyc index c27011e..2026e60 100644 Binary files a/app/blueprints/__pycache__/main.cpython-39.pyc and b/app/blueprints/__pycache__/main.cpython-39.pyc differ diff --git a/app/blueprints/main.py b/app/blueprints/main.py index a9717dc..82e5ce5 100644 --- a/app/blueprints/main.py +++ b/app/blueprints/main.py @@ -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: # 使用流式请求 diff --git a/app/config.py b/app/config.py index a371ba2..cf3f6fc 100644 --- a/app/config.py +++ b/app/config.py @@ -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) diff --git a/app/main.py b/app/main.py index 4cd2924..955832a 100644 --- a/app/main.py +++ b/app/main.py @@ -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)