Merge branch 'master' of https://gitee.com/TeaCodeNice/crawler-plugin

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()
 

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:
+        self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
-                # 阻塞式获取任务，避免CPU空转
+        futures = []  # 保存所有提交的任务Future对象
-                task_data = JSON_TASK_QUEUE.get(block=True, timeout=1)
+        
+        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)
         
-        # 切换国家
+        # 切换国家，最多重试3次
-        try:
+        max_retries = 3
-            switch_success = driver.SwitchingCountries(country)
+        switch_success = False
-            if not switch_success:
+        
-                self.log(f"切换到国家 {country} 失败，跳过该国家", "ERROR")
+        for retry in range(max_retries):
-                return
+            try:
-        except Exception as e:
+                self.log(f"尝试切换到国家 {country} (第 {retry + 1}/{max_retries} 次)")
-            self.log(f"切换国家 {country} 异常: {str(e)}", "ERROR")
+                
+                # 如果不是第一次尝试，先刷新页面
+                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:
+        for retry in range(max_retries):
-            # 搜索ASIN
+            try:
-            sku_ls = driver.search(asin=asin)
+                self.log(f"处理ASIN {asin} (第 {retry + 1}/{max_retries} 次)")
-            self.log(f"搜索到 {len(sku_ls)} 个SKU")
+                
-            
+                # 如果不是第一次尝试，先刷新页面
-            if len(sku_ls) == 0:
+                if retry > 0:
-                status = "查询不到"
+                    self.log("重试前刷新页面...")
-                self.log(f"ASIN {asin} 未找到商品", "WARNING")
-            else:
-                # 删除所有找到的SKU
-                success_count = 0
-                for sku in sku_ls:
                     try:
-                        suc = driver.del_action(sku)
+                        driver.tab.refresh()
-                        if suc:
+                        time.sleep(3)
-                            success_count += 1
-                            self.log(f"SKU 删除成功")
-                        else:
-                            self.log(f"SKU 删除失败", "WARNING")
                     except Exception as e:
-                        self.log(f"删除SKU异常: {str(e)}", "ERROR")
+                        self.log(f"刷新页面失败: {str(e)}", "WARNING")
-                        
+                
-                if success_count > 0:
+                # 搜索ASIN
-                    status = "成功"
+                sku_ls = driver.search(asin=asin)
-                    self.log(f"ASIN {asin} 删除成功 ({success_count}/{len(sku_ls)})")
+                self.log(f"搜索到 {len(sku_ls)} 个SKU")
-                    # 更新成功计数
+                
-                    if task_id in runing_task:
+                if len(sku_ls) == 0:
-                        runing_task[task_id]["success_count"] += 1
+                    status = "查询不到"
+                    self.log(f"ASIN {asin} 未找到商品", "WARNING")
+                    break  # 查询不到商品，无需重试
                 else:
-                    status = "失败"
+                    # 删除所有找到的SKU，如果任何一个失败则重新开始整个流程
-                    self.log(f"ASIN {asin} 所有SKU删除失败", "ERROR")
+                    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")
 
 

app/blueprints/main.py

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

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)

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)