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

app/ali_oss.py

@@ -14,6 +14,7 @@ def upload_file(file_content: bytes, key: str):
     cfg.credentials_provider = credentials_provider
     cfg.region = region
     cfg.endpoint = endpoint
+    cfg.retry_max_attempts = 3
     client = oss.Client(cfg)
 
     result = client.put_object(

app/amazon/del_brand.py

@@ -0,0 +1,519 @@
+import traceback
+import winreg
+import subprocess
+import time
+import uuid
+import requests
+import json
+import os
+
+from typing import Literal
+from DrissionPage import Chromium
+from DrissionPage.common import By
+
+
+def kill_process(version: Literal["v5", "v6"]):
+    """杀紫鸟客户端进程（独立函数版本）"""
+    driver = ZiniaoDriver({})
+    driver.kill_process(version)
+
+
+class ZiniaoDriver:
+    """紫鸟浏览器自动化驱动类"""
+
+    def __init__(self, user_info: dict, socket_port: int = 19890):
+        """
+        初始化紫鸟浏览器驱动
+
+        Args:
+            user_info: 用户信息字典，包含 company, username, password
+            socket_port: 客户端通信端口，默认 19890
+        """
+        self.user_info = user_info
+        self.socket_port = socket_port
+        self.client_path = None
+        self.browser = None
+        self.tab = None
+        self.store_id = None
+
+    def get_zinaio_exe(self, protocol_name: str = "superbrowser"):
+        """
+        获取紫鸟安装目录
+
+        Args:
+            protocol_name: 协议名称，默认 "superbrowser"
+
+        Returns:
+            exe_path: 紫鸟浏览器可执行文件路径
+        """
+        try:
+            key_path = rf"SOFTWARE\Classes\{protocol_name}\shell\open\command"
+            key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path)
+            command, _ = winreg.QueryValueEx(key, "")
+            winreg.CloseKey(key)
+            if isinstance(command, str):
+                exe_path = command.split()[0]
+            else:
+                exe_path = command[0]
+            return exe_path
+        except FileNotFoundError:
+            try:
+                key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path)
+                command, _ = winreg.QueryValueEx(key, "")
+                winreg.CloseKey(key)
+                if isinstance(command, str):
+                    exe_path = command.split()[0]
+                else:
+                    exe_path = command[0]
+                return exe_path
+            except FileNotFoundError:
+                return None
+
+    def update_core(self):
+        """
+        下载所有内核，打开店铺前调用，需客户端版本5.285.7以上
+        因为http有超时时间，所以这个action适合循环调用，直到返回成功
+        """
+        data = {
+            "action": "updateCore",
+            "requestId": str(uuid.uuid4()),
+        }
+        data.update(self.user_info)
+        while True:
+            url = f'http://127.0.0.1:{self.socket_port}'
+            response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
+            result = response.json()
+            print(result)
+            if result is None:
+                print("等待客户端启动...")
+                time.sleep(2)
+                continue
+            if result.get("statusCode") is None or result.get("statusCode") == -10003:
+                print("当前版本不支持此接口，请升级客户端")
+                return
+            elif result.get("statusCode") == 0:
+                print("更新内核完成")
+                return
+            else:
+                print(f"等待更新内核: {json.dumps(result)}")
+                time.sleep(2)
+
+    def kill_process(self, version: Literal["v5", "v6"]):
+        """
+        杀紫鸟客户端进程
+
+        Args:
+            version: 客户端版本
+        """
+        if version == "v5":
+            process_name = 'SuperBrowser.exe'
+        else:
+            process_name = 'ziniao.exe'
+        os.system('taskkill /f /t /im ' + process_name)
+        time.sleep(3)
+
+    def get_browser_list(self) -> list:
+        """
+        获取浏览器列表
+
+        Returns:
+            list: 浏览器列表
+        """
+        request_id = str(uuid.uuid4())
+        data = {
+            "action": "getBrowserList",
+            "requestId": request_id
+        }
+        data.update(self.user_info)
+
+        url = f'http://127.0.0.1:{self.socket_port}'
+        response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
+        r = response.json()
+        if str(r.get("statusCode")) == "0":
+            print(r)
+            return r.get("browserList")
+        elif str(r.get("statusCode")) == "-10003":
+            print(f"login Err {json.dumps(r, ensure_ascii=False)}")
+            exit()
+        else:
+            print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
+            exit()
+
+    def open_store(self, store_info, isWebDriverReadOnlyMode=0, isprivacy=0,
+                   isHeadless=0, cookieTypeSave=0, jsInfo=""):
+        """
+        打开店铺
+
+        Args:
+            store_info: 店铺信息（browserId 或 browserOauth）
+            isWebDriverReadOnlyMode: 是否只读模式，默认 0
+            isprivacy: 隐私模式，默认 0
+            isHeadless: 无头模式，默认 0
+            cookieTypeSave: cookie保存类型，默认 0
+            jsInfo: 注入的JS信息，默认 ""
+
+        Returns:
+            dict: 返回结果
+        """
+        request_id = str(uuid.uuid4())
+        data = {
+            "action": "startBrowser",
+            "isWaitPluginUpdate": 0,
+            "isHeadless": isHeadless,
+            "requestId": request_id,
+            "isWebDriverReadOnlyMode": isWebDriverReadOnlyMode,
+            "cookieTypeLoad": 0,
+            "cookieTypeSave": cookieTypeSave,
+            "runMode": "1",
+            "isLoadUserPlugin": False,
+            "pluginIdType": 1,
+            "privacyMode": isprivacy
+        }
+        data.update(self.user_info)
+
+        if store_info.isdigit():
+            data["browserId"] = store_info
+        else:
+            data["browserOauth"] = store_info
+
+        if len(str(jsInfo)) > 2:
+            data["injectJsInfo"] = json.dumps(jsInfo)
+
+        url = f'http://127.0.0.1:{self.socket_port}'
+        response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
+        r = response.json()
+        if str(r.get("statusCode")) == "0":
+            return r
+        elif str(r.get("statusCode")) == "-10003":
+            print(f"login Err {json.dumps(r, ensure_ascii=False)}")
+            exit()
+        else:
+            print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
+            exit()
+
+    def get_browser(self, port) -> Chromium:
+        """
+        获取浏览器实例
+
+        Args:
+            port: 调试端口
+
+        Returns:
+            Chromium: DrissionPage浏览器实例
+        """
+        browser = Chromium(port)
+        return browser
+
+    def start_client(self):
+        """启动紫鸟客户端"""
+        self.client_path = self.get_zinaio_exe("superbrowser").strip('"')
+        print(self.client_path)
+        cmd = [self.client_path, '--run_type=web_driver', '--ipc_type=http',
+               '--port=' + str(self.socket_port)]
+        print(" ".join(cmd))
+        subprocess.Popen(cmd)
+        time.sleep(5)
+
+    def open_shop(self, shop_name: str):
+        """
+        打开指定店铺
+
+        Args:
+            shop_name: 店铺名称
+
+        Returns:
+            Chromium or str: 成功返回浏览器实例，失败返回错误信息
+        """
+        # 启动客户端
+        self.start_client()
+
+        # 更新内核
+        self.update_core()
+
+        # 获取店铺列表
+        shop_ls = self.get_browser_list()
+        self.store_id = None
+        # print(shop_ls)
+        for shop in shop_ls:
+            if shop.get("browserName") == shop_name:
+                self.store_id = shop.get('browserOauth')
+                break
+
+        if not self.store_id:
+            print("店铺不存在")
+            return "店铺不存在"
+
+        # 打开店铺
+        ret_json = self.open_store(self.store_id)
+        print(ret_json)
+        self.store_id = ret_json.get("browserOauth")
+        if self.store_id is None:
+            self.store_id = ret_json.get("browserId")
+
+        # 获取drissionpage浏览器会话
+        self.browser = self.get_browser(ret_json.get('debuggingPort'))
+
+        ip_check_url = ret_json.get("ipDetectionPage")
+        if not ip_check_url:
+            print("ip检测页地址为空，请升级紫鸟浏览器到最新版")
+            print(f"=====关闭店铺：{shop_name}=====")
+            self.close_store(self.store_id)
+            exit()
+        ip_usable = self.open_ip_check(self.browser, ip_check_url)
+        if ip_usable:
+            print("ip检测通过，打开店铺平台主页")
+            self.open_launcher_page(ret_json.get("launcherPage"), self.browser)
+        else:
+            print("IP检测不通过")
+        return self.browser
+
+    def close_store(self, browser_oauth=None):
+        """
+        关闭店铺
+
+        Args:
+            browser_oauth: 店铺OAuth标识，如果不提供则使用当前打开的店铺
+
+        Returns:
+            dict: 返回结果
+        """
+        if browser_oauth is None:
+            browser_oauth = self.store_id
+
+        request_id = str(uuid.uuid4())
+        data = {
+            "action": "stopBrowser",
+            "requestId": request_id,
+            "duplicate": 0,
+            "browserOauth": browser_oauth
+        }
+        data.update(self.user_info)
+
+        url = f'http://127.0.0.1:{self.socket_port}'
+        response = requests.post(url, json.dumps(data).encode('utf-8'), timeout=120)
+        r = response.json()
+        if str(r.get("statusCode")) == "0":
+            return r
+        elif str(r.get("statusCode")) == "-10003":
+            print(f"login Err {json.dumps(r, ensure_ascii=False)}")
+            exit()
+        else:
+            print(f"Fail {json.dumps(r, ensure_ascii=False)} ")
+            exit()
+
+    def open_launcher_page(self, launcher_page: str, browser: Chromium = None):
+        """
+        打开启动页面
+
+        Args:
+            launcher_page: 要打开的页面URL
+            browser: 浏览器实例，如果不提供则使用当前浏览器实例
+        """
+        if browser is None:
+            browser = self.browser
+
+        tab = browser.new_tab(url=launcher_page)
+        self.tab = tab
+        return tab
+
+    def open_ip_check(self, browser: Chromium, ip_check_url: str):
+        """
+        打开ip检测页检测ip是否正常
+        :param browser: drissionpage浏览器会话
+        :param ip_check_url ip检测页地址
+        :return 检测结果
+        """
+        try:
+            tab = browser.latest_tab
+            tab.get(ip_check_url)
+            success_button = tab.ele((By.XPATH, '//button[contains(@class, "styles_btn--success")]'),
+                                     timeout=60)  # 等待查找元素60秒
+            if success_button:
+                print("ip检测成功")
+                return True
+            else:
+                print("ip检测超时")
+                return False
+        except Exception as e:
+            print("ip检测异常:" + traceback.format_exc())
+            return False
+
+
+class AmazoneDriver(ZiniaoDriver):
+    """亚马逊专用驱动类，继承自ZiniaoDriver"""
+
+    def SwitchingCountries(self, country_name: str):
+        """
+        切换国家
+        操作：
+        1、//div[@class="dropdown-account-switcher-header-label"]/span[last()] 获取此元素文本，判断当前国家，如果与目标国家相同则不操作，否则执行下一步
+        2、点击 //div[@class="dropdown-account-switcher-header-label"] 打开下拉框
+        3、点击 //div[@class="dropdown-account-switcher-list-item"] 第一个展开国家列表
+        4、点击 //div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="国家名"] 切换到目标国家
+        5、等待页面加载完成，判断国家是否切换成功，成功则返回True，否则返回False
+
+        Args:
+            country_name: 目标国家名称
+
+        Returns:
+            bool: 切换成功返回True，失败返回False
+        """
+        try:
+            if self.browser is None:
+                print("浏览器实例不存在，请先打开店铺")
+                return False
+
+            # 获取当前标签页
+            tab = self.tab
+
+            # 步骤1：获取当前国家名称
+            print(f"正在检查当前国家...")
+            current_country_ele = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]',
+                                          timeout=10)
+            if current_country_ele:
+                current_country = current_country_ele.text.strip()
+                print(f"当前国家：{current_country}")
+
+                # 判断是否与目标国家相同
+                if current_country == country_name:
+                    print(f"当前已经是目标国家 {country_name}，无需切换")
+                    return True
+            else:
+                print("无法获取当前国家信息")
+                return False
+
+            # 步骤2：点击打开下拉框
+            print(f"正在打开国家切换下拉框...")
+            dropdown_header = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]', timeout=10)
+            if not dropdown_header:
+                print("找不到国家切换下拉框")
+                return False
+            dropdown_header.click()
+            time.sleep(1)  # 等待下拉框展开
+
+            # 步骤3：点击第一个展开国家列表
+            print(f"正在展开国家列表...")
+            first_item = tab.ele('xpath://div[@class="dropdown-account-switcher-list-item"]', timeout=10)
+            if not first_item:
+                print("找不到国家列表项")
+                return False
+            first_item.click()
+            time.sleep(1)  # 等待国家列表展开
+
+            # 步骤4：点击目标国家
+            print(f"正在切换到国家：{country_name}")
+            target_country_xpath = f'//div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="{country_name}"]'
+            target_country = tab.ele(f'xpath:{target_country_xpath}', timeout=10)
+            if not target_country:
+                print(f"找不到目标国家：{country_name}")
+                return False
+            target_country.click()
+
+            # 步骤5：等待页面加载完成并验证切换结果
+            print(f"等待页面加载...")
+            # time.sleep(3)  # 等待页面加载
+            self.tab.wait.doc_loaded()
+
+            # 再次检查当前国家
+            new_country_ele = tab.ele('xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]',
+                                      timeout=10)
+            if new_country_ele:
+                new_country = new_country_ele.text.strip()
+                if new_country == country_name:
+                    print(f"国家切换成功：{new_country}")
+                    return True
+                else:
+                    print(f"国家切换失败，当前国家：{new_country}，目标国家：{country_name}")
+                    return False
+            else:
+                print("无法验证切换结果")
+                return False
+
+        except Exception as e:
+            print(f"切换国家时发生异常：{traceback.format_exc()}")
+            return False
+
+    def SwitchPage(self):
+        """
+        切换至 管理所有库存页面
+        1、等待 //navigation-favorites-bar[@class="hydrated"] 出现
+        """
+        navigation = self.tab.ele('xpath://navigation-favorites-bar[@class="hydrated"]')
+        navigation.wait.displayed(raise_err=False)
+        page_btn = navigation.sr('xpath://internal-fav-bar-links[@data-internal="navigation"]').sr(
+            'xpath://a[@data-page-id="ezdpc-gui-inventory-mons"]')
+        page_btn.wait.displayed(raise_err=False)
+        page_btn.click(timeout=5)
+
+        self.tab.wait.doc_loaded()
+        # 等待搜索框出现
+        search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group')
+        search_region.wait.displayed(raise_err=False)
+
+    def search(self, asin):
+        search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group')
+        search_region.wait.displayed(raise_err=False)
+        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)
+
+        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)
+            time.sleep(1)
+        sku_ls = self.tab.eles("//div[@data-sku]")
+        return sku_ls
+
+    def del_action(self, sku_ele):
+        dropdown = sku_ele.ele('xpath:.//kat-dropdown-button[@variant="secondary"]')
+        dropdown.click()
+        time.sleep(1)
+
+        del_btn = dropdown.ele("xpath:.//button[@role='menuitem' and text()='删除商品信息']")
+        del_btn.wait.displayed(raise_err=False)
+        del_btn.click()
+
+        confirm_btn = self.tab.ele('xpath://kat-modal[@data-testid="action-modal"]//kat-button[@variant="primary"]')
+        confirm_btn.wait.displayed(raise_err=False)
+        confirm_btn.wait.enabled()
+        confirm_btn.click()
+
+        suc_alert = self.tab.ele("xpath://kat-alert[@variant='success']")
+        suc = suc_alert.wait.displayed(raise_err=False, timeout=5)
+        return suc
+
+
+if __name__ == '__main__':
+    # 使用示例
+    user_info = {
+        "company": "rongchuang123",
+        "username": "自动化_Robot",
+        "password": "#20zsg25"
+    }
+    shop_name = "郭亚芳"
+    kill_process("v6")
+
+    # 创建驱动实例
+    driver = AmazoneDriver(user_info)
+
+    # 打开店铺并获取浏览器实例
+    browser = driver.open_shop(shop_name)
+
+    country = "西班牙"
+    asin = "Voanos"
+
+    sw_suc = driver.SwitchingCountries(country)
+    driver.SwitchPage()
+    sku_ls = driver.search(asin=asin)
+    print("查询结果有：", len(sku_ls))
+
+    for sku in sku_ls:
+        suc = driver.del_action(sku)
+        print(sku, "删除结果", suc)
+
+
+

app/amazon/main.py

@@ -0,0 +1,421 @@
+import time
+import traceback
+import requests
+from datetime import datetime
+from typing import Dict, Any, List
+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
+
+
+class TaskMonitor:
+    """任务监控器：负责监控队列并执行品牌删除任务"""
+    
+    def __init__(self):
+        """初始化任务监控器"""
+        self.running = True
+        self.user_info = {
+            "company": ZN_COMPANY,
+            "username": ZN_USERNAME,
+            "password": ZN_PASSWORD
+        }
+        
+        self.chunk_index = 1  # 当前处理的分块索引
+
+    def log(self, message: str, level: str = "INFO"):
+        """日志输出
+        
+        Args:
+            message: 日志消息
+            level: 日志级别
+        """
+        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        print(f"[{timestamp}] [{level}] {message}")
+        
+    def start(self):
+        """启动任务监控（阻塞式循环）"""
+        self.log("任务监控器启动，开始监听队列...")
+        
+        while self.running:
+            try:
+                # 阻塞式获取任务，避免CPU空转
+                task_data = JSON_TASK_QUEUE.get(block=True, timeout=1)
+                
+                # 检查任务类型
+                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]):
+        """处理单个任务
+        
+        Args:
+            task_data: 任务数据
+        """
+        try:
+            data = task_data.get("data", {})
+            task_id = data.get("taskId")
+            items = data.get("items", [])
+            
+            if not task_id:
+                self.log("任务ID为空，跳过", "WARNING")
+                return
+                
+            # 初始化任务状态
+            self.update_task_status(
+                task_id,
+                status="running",
+                start_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                total_shops=len(items),
+                processed_shops=0,
+                total_asins=sum(shop.get("totalRows", 0) for shop in items),
+                processed_asins=0,
+                success_count=0,
+                failed_count=0,
+                stop_requested=False  # 暂停请求标志
+            )
+            
+            self.log(f"开始处理任务 {task_id}，共 {len(items)} 个店铺")
+            
+            # 遍历处理每个店铺
+            for idx, shop_data in enumerate(items, 1):
+                shop_name = shop_data.get("shopName", "未知店铺")
+                self.log(f"[{idx}/{len(items)}] 开始处理店铺: {shop_name}")
+                
+                try:
+                    self.process_shop(shop_data, task_id)
+                    # 更新已处理店铺数
+                    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 and runing_task[task_id].get("stop_requested", False):
+                self.update_task_status(task_id, status="stopped")
+                self.log(f"任务 {task_id} 已被暂停！")
+            else:
+                self.update_task_status(task_id, status="completed")
+                self.log(f"任务 {task_id} 处理完成！")
+            
+        except Exception as e:
+            self.log(f"任务处理失败: {traceback.format_exc()}", "ERROR")
+            if task_id:
+                self.update_task_status(task_id, status="failed", error=str(e))
+                
+    def process_shop(self, shop_data: Dict[str, Any], task_id: int):
+        """处理单个店铺（包含重试逻辑）
+        
+        Args:
+            shop_data: 店铺数据
+            task_id: 任务ID
+        """
+        shop_name = shop_data.get("shopName", "未知店铺")
+        result_id = shop_data.get("resultId")
+        countries = shop_data.get("countries", [])
+        
+        if not countries:
+            self.log(f"店铺 {shop_name} 没有国家数据，跳过", "WARNING")
+            return
+            
+        # 更新当前处理的店铺
+        self.update_task_status(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} 已添加到执行列表，开始时间: {start_time}")
+        
+        # 店铺打开重试最多3次
+        driver = None
+        browser = None
+        max_retries = 3
+        
+        for retry in range(max_retries):
+            try:
+                self.log(f"尝试打开店铺 {shop_name} (第 {retry + 1}/{max_retries} 次)")
+                
+                # 如果不是第一次尝试，先杀进程
+                if retry > 0:
+                    self.log("重试前先杀掉浏览器进程...")
+                    kill_process("v6")
+                    time.sleep(2)
+                
+                # 创建驱动并打开店铺
+                driver = AmazoneDriver(self.user_info)
+                browser = driver.open_shop(shop_name)
+                
+                if browser and browser != "店铺不存在":
+                    self.log(f"成功打开店铺 {shop_name}")
+                    break
+                else:
+                    self.log(f"打开店铺失败: {browser}", "WARNING")
+                    driver = None
+                    
+            except Exception as e:
+                self.log(f"打开店铺异常: {str(e)}", "ERROR")
+                driver = None
+                
+            # 如果还有重试机会，等待后继续
+            if retry < max_retries - 1:
+                time.sleep(3)
+                
+        # 检查是否成功打开
+        if not driver or not browser or browser == "店铺不存在":
+            self.log(f"店铺 {shop_name} 打开失败，已重试 {max_retries} 次，跳过该店铺", "ERROR")
+            return
+            
+        try:
+            # 处理每个国家
+            for country_data in countries:
+                # 检查是否收到暂停请求
+                if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
+                    self.log(f"检测到任务 {task_id} 的暂停请求，停止处理国家", "WARNING")
+                    break  # 跳出循环，进入finally关闭店铺
+                
+                try:
+                    self.process_country(driver, country_data, task_id, result_id, shop_data)
+                except Exception as e:
+                    country_name = country_data.get("country", "未知")
+                    self.log(f"处理国家 {country_name} 失败: {str(e)}", "ERROR")
+                    self.log(traceback.format_exc(), "ERROR")
+                    
+        finally:
+            # 关闭店铺
+            try:
+                if driver:
+                    self.log(f"关闭店铺 {shop_name}")
+                    driver.close_store()
+                    time.sleep(2)
+            except Exception as e:
+                self.log(f"关闭店铺失败: {str(e)}", "WARNING")
+            
+            # 从正在执行中的店铺列表中移除
+            if shop_name in runing_shop:
+                del runing_shop[shop_name]
+                self.log(f"店铺 {shop_name} 已从执行列表中移除")
+                
+    def process_country(self, driver: AmazoneDriver, country_data: Dict[str, Any], 
+                       task_id: int, result_id: int, shop_data: Dict[str, Any]):
+        """处理单个国家的所有ASIN
+        
+        Args:
+            driver: 亚马逊驱动实例
+            country_data: 国家数据
+            task_id: 任务ID
+            result_id: 结果ID
+            shop_data: 店铺数据
+        """
+        country = country_data.get("country", "未知")
+        items = country_data.get("items", [])
+        
+        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")
+            return
+            
+        # 切换到库存管理页面
+        try:
+            driver.SwitchPage()
+            self.log(f"已切换到库存管理页面")
+        except Exception as e:
+            self.log(f"切换页面失败: {str(e)}", "ERROR")
+            return
+            
+        # 处理每个ASIN
+        file_key = shop_data.get("fileKey", "")
+        source_filename = shop_data.get("sourceFilename", "")
+        total_rows = shop_data.get("totalRows", 0)
+        
+        for idx, asin_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")
+                return  # 返回到上层，会触发店铺关闭
+            
+            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)
+            except Exception as e:
+                asin = asin_item.get("asin", "未知")
+                self.log(f"处理ASIN {asin} 失败: {str(e)}", "ERROR")
+                # 继续处理下一个ASIN
+                
+    def process_asin(self, driver: AmazoneDriver, asin_item: Dict[str, Any], 
+                    country: str, task_id: int, result_id: int, file_key: str, 
+                    source_filename: str, total_rows: int):
+        """处理单个ASIN并回传结果
+        
+        Args:
+            driver: 亚马逊驱动实例
+            asin_item: ASIN数据项
+            country: 国家名称
+            task_id: 任务ID
+            result_id: 结果ID
+            file_key: 文件KEY
+            source_filename: 源文件名
+            total_rows: 总行数
+            chunk_index: 当前处理索引
+        """
+        asin = asin_item.get("asin", "")
+        row_index = asin_item.get("rowIndex", 0)
+        
+        # 更新当前处理的ASIN
+        self.update_task_status(task_id, current_asin=asin)
+        
+        status = "失败"
+        
+        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:
+                    try:
+                        suc = driver.del_action(sku)
+                        if suc:
+                            success_count += 1
+                            self.log(f"SKU 删除成功")
+                        else:
+                            self.log(f"SKU 删除失败", "WARNING")
+                    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
+                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 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
+            
+        # 回传结果到API
+        try:
+            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": status
+                        }]
+                    }]
+                }]
+            }
+            
+            self.post_result(task_id, payload)
+            self.log(f"ASIN {asin} 结果已回传，状态: {status}")
+            
+        except Exception as e:
+            self.log(f"回传结果失败: {str(e)}", "ERROR")
+            
+    def post_result(self, task_id: int, payload: Dict[str, Any]):
+        """回传结果到API
+        
+        Args:
+            task_id: 任务ID
+            payload: 结果数据
+        """
+        url = f"{DELETE_BRAND_API_BASE}/newApi/api/delete-brand/tasks/{task_id}/result"
+        
+        try:
+            response = requests.post(
+                url,
+                json=payload,
+                headers={"Content-Type": "application/json"},
+                timeout=30,
+                verify=False  # 忽略SSL证书验证
+            )
+            
+            if response.status_code == 200:
+                self.log(f"结果回传成功: {url}")
+            else:
+                self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING")
+                
+        except Exception as e:
+            self.log(f"调用API异常: {str(e)}", "ERROR")
+            
+    def update_task_status(self, task_id: int, **kwargs):
+        """更新任务状态
+        
+        Args:
+            task_id: 任务ID
+            **kwargs: 要更新的字段
+        """
+        if task_id not in runing_task:
+            runing_task[task_id] = {}
+            
+        runing_task[task_id].update(kwargs)
+        
+    def stop(self):
+        """停止监控"""
+        self.running = False
+        self.log("任务监控器已停止")
+
+
+def main():
+    """主函数：启动任务监控"""
+    # 创建并启动任务监控器
+    monitor = TaskMonitor()
+    
+    try:
+        monitor.start()
+    except KeyboardInterrupt:
+        monitor.log("接收到中断信号，正在停止...")
+        monitor.stop()
+    except Exception as e:
+        monitor.log(f"监控器异常退出: {traceback.format_exc()}", "ERROR")
+
+
+if __name__ == "__main__":
+    main()

app/blueprints/brand.py

@@ -187,8 +187,8 @@ def _run_brand_single(taskid,brand_ls, fileUrl,strategy,totalLines,chunkTotal,ch
                          #      "invalidBrands": invalidBrands,"queryFailedBrands": queryFailedBrands
                          #    }]})
                          json=data)
-    print(data)
-    print(taskid,brand_ls, fileUrl)
+    # print(data)
+    # print(taskid,brand_ls, fileUrl)
     print("提交结果",data,"\n-->",resp.text)
     return True
 

app/blueprints/brand_备份.py

@@ -0,0 +1,726 @@
+"""
+品牌爬虫蓝图：展开文件夹、运行任务、任务列表/详情、下载结果
+"""
+import os
+import json
+import threading
+import traceback
+import zipfile
+import io
+import time
+import requests
+from queue import Queue, Empty
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from urllib.parse import urlparse
+
+from flask import Blueprint, request, jsonify, session, send_file, redirect, Response
+
+from app_common import get_db, login_required, BASE_DIR
+from config import bucket_path,JAVA_API_BASE
+from brand_spider.main import single_file_handle, TaskCancelledError
+
+brand_bp = Blueprint('brand', __name__)
+BRAND_OUTPUT_DIR = os.path.join(BASE_DIR, 'brand_output')
+OSS_PREFIX = "brand_results"
+
+# 任务完成时推送给前端的 SSE 队列：task_id -> [Queue, ...]
+_task_event_queues = {}
+_task_event_lock = threading.Lock()
+
+# 任务行级进度（当前处理文件的行数/总行数），仅存内存，不落库：
+# task_id -> {
+#   'file_index': int,      # 当前处理第几个文件
+#   'file_total': int,      # 总文件数
+#   'file_path': str,       # 当前文件路径
+#   'current_line': int,    # 当前行号（或当前品牌序号）
+#   'total_lines': int,     # 总行数（或总品牌数）
+# }
+_task_line_progress = {}
+_task_line_progress_lock = threading.Lock()
+_cancelled_brand_tasks = set()
+_cancelled_brand_tasks_lock = threading.Lock()
+
+
+def _mark_task_cancelled(task_id):
+    with _cancelled_brand_tasks_lock:
+        _cancelled_brand_tasks.add(task_id)
+
+
+def _is_task_cancelled(task_id):
+    with _cancelled_brand_tasks_lock:
+        return task_id in _cancelled_brand_tasks
+
+
+def _get_uid_from_request_headers():
+    """
+    从请求头读取 uid。
+    前端会在 headers 里携带 uid，用于确定本次请求的归属用户。
+    """
+    uid = request.headers.get('uid')
+    if uid is None:
+        return None
+    uid = str(uid).strip()
+    if not uid:
+        return None
+    try:
+        return int(uid)
+    except Exception:
+        return None
+
+
+def _resolve_user_id():
+    """
+    优先使用请求头 uid；没有请求头 uid 时回退到 session['user_id']。
+    若二者同时存在但不一致，则视为无效请求并返回 None。
+    """
+    req_uid = _get_uid_from_request_headers()
+    if req_uid is not None:
+        sess_uid = session.get('user_id')
+        if sess_uid is not None:
+            try:
+                if int(sess_uid) != int(req_uid):
+                    return None
+            except Exception:
+                if str(sess_uid) != str(req_uid):
+                    return None
+        return req_uid
+    return session.get('user_id')
+
+
+def _get_user_id_or_error():
+    user_id = _resolve_user_id()
+    if not user_id:
+        return None, (jsonify({'success': False, 'error': '未登录或缺少uid'}), 401)
+    try:
+        return int(user_id), None
+    except Exception:
+        return None, (jsonify({'success': False, 'error': 'uid无效'}), 400)
+
+
+def _push_task_event(task_id, event):
+    """向订阅了该任务的所有 SSE 连接推送事件，并移除该任务的队列列表。
+    同时清理内存中的行级进度，不通过数据库中转行级进度。"""
+    with _task_event_lock:
+        queues = _task_event_queues.pop(task_id, [])
+    for q in queues:
+        try:
+            q.put_nowait(event)
+        except Exception:
+            pass
+    # 任务进入终态时，顺便清理行级进度缓存
+    with _task_line_progress_lock:
+        _task_line_progress.pop(task_id, None)
+    status = (event or {}).get('status') if isinstance(event, dict) else None
+    if status in ('success', 'failed', 'cancelled'):
+        with _cancelled_brand_tasks_lock:
+            _cancelled_brand_tasks.discard(task_id)
+
+
+def _expand_folder_xlsx(folder_path):
+    """返回文件夹下所有 .xlsx 文件的绝对路径列表"""
+    if not folder_path or not os.path.isdir(folder_path):
+        return []
+    paths = []
+    for name in os.listdir(folder_path):
+        if name.endswith('.xlsx') or name.endswith('.XLSX'):
+            paths.append(os.path.normpath(os.path.join(folder_path, name)))
+    return paths
+
+
+def _upload_local_xlsx_to_oss(local_path, user_id):
+    """将本地 xlsx 文件上传到 OSS，返回 (url, None) 或 (None, error_message)。"""
+    if not local_path or not os.path.isfile(local_path):
+        return None, "文件不存在"
+    try:
+        from ali_oss import upload_file as oss_upload_file
+    except ImportError:
+        return None, "OSS 模块未配置"
+    base = os.path.basename(local_path)
+    safe_base = "".join(c if c.isalnum() or c in '-_.' else '_' for c in base)
+    if not safe_base.endswith('.xlsx'):
+        safe_base = safe_base + '.xlsx'
+    key = f"{bucket_path}brand_input/{user_id}/{int(time.time() * 1000)}/{safe_base}"
+    try:
+        with open(local_path, "rb") as f:
+            content = f.read()
+        url = oss_upload_file(content, key)
+        return url, None
+    except Exception as e:
+        return None, str(e)
+
+
+def _run_brand_single(taskid,brand_ls, fileUrl,strategy,totalLines,chunkTotal,chunkIndex):
+    """"""
+    invalidBrands = [] # 不符合品牌的数据
+    queryFailedBrands = [] # 查询失败的数据
+    keptRows = []
+    for brand in brand_ls:
+        if _is_task_cancelled(taskid):
+            raise TaskCancelledError("任务已取消")
+        faild_data,query_faild_data = single_file_handle(brand,strategy)
+        invalidBrands.extend(faild_data)
+        queryFailedBrands.extend(query_faild_data)
+        if len(faild_data) == 0 and len(query_faild_data) == 0:
+            keptRows.append(brand)
+
+    data = { "strategy": strategy ,
+                          "files": [
+                            {
+                              "fileUrl": fileUrl,
+                              "originalFilename": "",
+                              "relativePath": "",
+                              "mainSheetName": "",
+                              "chunkIndex": chunkIndex,
+                              "chunkTotal": chunkTotal,
+                              "totalLines": totalLines,
+                              "keptRows": keptRows,
+                              "invalidBrands": invalidBrands,
+                              "queryFailedBrands": queryFailedBrands
+                            }
+                          ]
+                        }
+    resp = requests.post(f"{JAVA_API_BASE}/api/brand/tasks/{taskid}/result",headers={"accept":"application/json"},
+                         # json={ "strategy": strategy ,
+                         #  "files": [
+                         #    {  "fileUrl": fileUrl,"originalFilename": "","relativePath": "",
+                         #      "mainSheetName": "","columns": [],"keptRows": [],
+                         #      "invalidBrands": invalidBrands,"queryFailedBrands": queryFailedBrands
+                         #    }]})
+                         json=data)
+    # print(data)
+    # print(taskid,brand_ls, fileUrl)
+    print("提交结果",data,"\n-->",resp.text)
+    return True
+
+
+
+def _background_brand_task(task_id,data):
+    """
+        后台执行爬虫任务
+        参数示例：
+            data : {
+            "taskId": 348,
+            "strategy": "Terms",
+            "files": [
+              {
+                "fileIndex": 1,
+                "fileUrl": "ab8cce4878754bfd89b4cd3ed20e1395",
+                "originalFilename": "品牌样例.xlsx",
+                "relativePath": "店铺A/品牌样例.xlsx",
+                "sheetName": "Sheet1",
+                "columns": [
+                  "品牌",
+                  "ASIH",
+                  "状态",
+                  "时间"
+                ],
+                "rows": [
+                  {
+                    "品牌": "YQAUTEC",
+                    "ASIH": "B0F28NZ752",
+                    "状态": "",
+                    "时间": "",
+                    "__rowIndex": 2
+                  }
+                ],
+                "uniqueBrands": [
+                  "YQAUTEC"
+                ]
+              }
+            ]
+          }
+    """
+    try:
+        file_ls = data.get("files")
+        strategy = data.get("strategy")
+        if not isinstance(file_ls, list):
+            file_ls = []
+        tasks = []
+        for file_data in file_ls:
+            rows = file_data.get("rows") or []
+            brand_ls = [i.get("品牌") for i in rows if i.get("品牌")]
+            fileUrl = file_data.get("fileUrl")
+            if not brand_ls:
+                continue
+            for i in range(0, len(brand_ls), 5):
+                chunk = brand_ls[i:i + 5]
+                tasks.append((chunk, fileUrl))
+
+        if tasks:
+            max_workers = min(8, len(tasks))
+            with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                if _is_task_cancelled(task_id):
+                    _push_task_event(task_id, {'status': 'cancelled'})
+                    return
+                futures = [
+                    executor.submit(_run_brand_single, task_id, chunk, file_url, strategy,len(brand_ls),len(tasks),chunkIndex+1)
+                    for chunkIndex,(chunk, file_url) in enumerate(tasks)
+                ]
+                for future in as_completed(futures):
+                    if _is_task_cancelled(task_id):
+                        for f in futures:
+                            f.cancel()
+                        _push_task_event(task_id, {'status': 'cancelled'})
+                        return
+                    future.result()
+        _push_task_event(task_id, {'status': 'success'})
+    except Exception as e:
+        print("执行出错",traceback.format_exc())
+        _push_task_event(task_id, {'status': 'failed', 'error_message': str(e)})
+    print("执行完成")
+
+
+def _parse_json(val, default=None):
+    if val is None:
+        return default if default is not None else []
+    if isinstance(val, (list, dict)):
+        return val
+    try:
+        return json.loads(val)
+    except Exception:
+        return default if default is not None else []
+
+
+def _is_url(s):
+    if not isinstance(s, str) or not s.strip():
+        return False
+    return s.strip().startswith('http://') or s.strip().startswith('https://')
+
+
+def _normalize_result_paths(raw):
+    """
+    将 result_paths 规范化为 (url_list, zip_url)。
+    支持旧格式 list 或新格式 dict {"urls": [...], "zip_url": "..."}。
+    """
+    if not raw:
+        return [], None
+    if isinstance(raw, dict):
+        urls = raw.get('urls') or []
+        if not isinstance(urls, list):
+            urls = []
+        zip_url = raw.get('zip_url')
+        if zip_url and not isinstance(zip_url, str):
+            zip_url = None
+        return urls, zip_url
+    if isinstance(raw, list):
+        return raw, None
+    return [], None
+
+
+@brand_bp.route('/api/brand/expand-folder', methods=['POST'])
+@login_required
+def api_brand_expand_folder():
+    """展开文件夹，返回其下所有 .xlsx 文件路径列表"""
+    try:
+        _, err = _get_user_id_or_error()
+        if err:
+            return err
+        data = request.get_json() or {}
+        folder = (data.get('folder') or '').strip()
+        if not folder:
+            return jsonify({'success': False, 'error': '请提供 folder 路径'}), 400
+        paths = _expand_folder_xlsx(folder)
+        return jsonify({'success': True, 'paths': paths})
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+def _expand_folder_xlsx_recursive_items(folder_path):
+    """返回文件夹下所有 .xlsx 文件的绝对路径和相对路径列表（递归）"""
+    if not folder_path or not os.path.isdir(folder_path):
+        return []
+    items = []
+    root = os.path.normpath(folder_path)
+    for current_root, _, filenames in os.walk(root):
+        for name in filenames:
+            if not (name.endswith('.xlsx') or name.endswith('.XLSX')):
+                continue
+            absolute_path = os.path.normpath(os.path.join(current_root, name))
+            relative_path = os.path.relpath(absolute_path, root).replace('\\', '/')
+            items.append({
+                'absolutePath': absolute_path,
+                'relativePath': relative_path,
+            })
+    items.sort(key=lambda item: item['relativePath'])
+    return items
+
+@brand_bp.route('/api/brand/expand-folder-recursive', methods=['POST'])
+@login_required
+def api_brand_expand_folder_recursive():
+    """递归展开文件夹，返回 .xlsx 文件绝对路径及相对路径列表"""
+    try:
+        _, err = _get_user_id_or_error()
+        if err:
+            return err
+        data = request.get_json() or {}
+        folder = (data.get('folder') or '').strip()
+        if not folder:
+            return jsonify({'success': False, 'error': '请提供 folder 路径'}), 400
+        items = _expand_folder_xlsx_recursive_items(folder)
+        return jsonify({'success': True, 'items': items})
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/run', methods=['POST'])
+# @login_required
+def api_brand_run():
+    """立即运行：创建任务并后台执行，立即返回 task_id，前端可轮询进度与取消"""
+    try:
+        data = request.get_json() or {}
+        paths = data.get('paths') or []
+        # task_type: 1=立即执行，2=添加任务；此接口默认 1
+        try:
+            task_type = int(data.get('task_type') or 1)
+        except Exception:
+            task_type = 1
+        if task_type not in (1, 2):
+            task_type = 1
+        strategy = (data.get('strategy') or 'Terms').strip()
+        if strategy not in ('Terms', 'Simple'):
+            strategy = 'Terms'
+        if not isinstance(paths, list):
+            paths = []
+        paths = [p.strip() for p in paths if p and isinstance(p, str) and os.path.isfile(p.strip())]
+        if not paths:
+            return jsonify({'success': False, 'error': '没有有效的 xlsx 文件路径'}), 400
+        # 先上传到 OSS，获取链接再入库
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
+        urls = []
+        for p in paths:
+            url, err = _upload_local_xlsx_to_oss(p, user_id)
+            if err:
+                return jsonify({'success': False, 'error': f'上传文件失败: {os.path.basename(p)} - {err}'}), 500
+            base_name = os.path.basename(p)
+            urls.append({"fileUrl": url,"originalFilename": base_name,"relativePath": p })
+
+        # 请求提交
+        resp = requests.post(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id}",headers={
+            "content-type":"application/json",
+        },data=json.dumps({ "files": urls, "strategy": strategy,"taskType": 1,"archiveName": ""}))
+        print({ "files": urls, "strategy": strategy,"taskType": 1,"archiveName": ""})
+        print(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id}")
+        resp_data = resp.json()
+        # print(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id} 返回：",resp_data,"状态：",resp.status_code)
+        if resp_data.get("success"):
+            task_id = resp_data.get("data").get("taskId")
+            data = resp_data.get("data")
+            threading.Thread(target=_background_brand_task, args=(task_id,data), daemon=True).start()
+            return jsonify({'success': True, 'task_id': task_id})
+        else:
+            print(resp_data)
+            return jsonify({'success': False, 'error': "请求接口失败"}), 500
+
+    except Exception as e:
+        traceback.print_exc()
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/tasks', methods=['GET', 'POST'])
+# @login_required
+def api_brand_tasks():
+    """GET: 获取当前用户的任务列表；POST: 添加任务（后台执行）"""
+    if request.method == 'GET':
+        try:
+            user_id, err = _get_user_id_or_error()
+            if err:
+                return err
+            resp = requests.get(f"{JAVA_API_BASE}/api/brand/tasks", params={
+                "userId": user_id
+            })
+            resp_data = resp.json()
+            print("获取列表",resp.text)
+            if resp_data.get("success"):
+                items = resp_data.get("data").get("items")
+                return jsonify({'success': True, 'items': items})
+            return jsonify({'success': True, 'items': []})
+        except Exception as e:
+            traceback.print_exc()
+            print("获取列表失败",e)
+            return jsonify({'success': False, 'error': str(e)}), 500
+    # POST
+    try:
+        data = request.get_json() or {}
+        paths = data.get('paths') or []
+        # task_type: 1=立即执行，2=添加任务；此接口默认 2
+        try:
+            task_type = int(data.get('task_type') or 2)
+        except Exception:
+            task_type = 2
+        if task_type not in (1, 2):
+            task_type = 2
+        strategy = (data.get('strategy') or 'Terms').strip()
+        if strategy not in ('Terms', 'Simple'):
+            strategy = 'Terms'
+        if not isinstance(paths, list):
+            paths = []
+        paths = [p.strip() for p in paths if p and isinstance(p, str)]
+        if not paths:
+            return jsonify({'success': False, 'error': '请提供至少一个文件路径或链接'}), 400
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
+        urls = []
+        for p in paths:
+            url, err = _upload_local_xlsx_to_oss(p, user_id)
+            if err:
+                return jsonify({'success': False, 'error': f'上传文件失败: {os.path.basename(p)} - {err}'}), 500
+            base_name = os.path.basename(p)
+            urls.append({"fileUrl": url, "originalFilename": base_name, "relativePath": p})
+        params = {
+            "userId": user_id
+        }
+        # resp = requests.post(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id}",headers={
+        #     "content-type":"application/json",
+        # },data=json.dumps({ "files": urls, "strategy": strategy,"taskType": 1,"archiveName": ""}))
+        req_data = {"files": urls, "strategy": strategy, "taskType": 2, "archiveName": ""}
+        resp = requests.post(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id}", headers={"content-type": "application/json"},
+                             data=json.dumps(req_data))
+        print(req_data)
+        # print(f"{JAVA_API_BASE}/api/brand/tasks?userId={user_id}，返回", resp.text)
+        resp_data = resp.json()
+        if resp_data.get("success"):
+            task_id = resp_data.get("data").get("taskId")
+            data = resp_data.get("data")
+            return jsonify({'success': True, 'task_id': task_id})
+        return jsonify({'success': False, 'task_id': "请求异常"})
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/tasks/<int:task_id>')
+# @login_required
+def api_brand_task_detail(task_id):
+    """获取单个任务详情（用于轮询状态）"""
+    try:
+        resp = requests.get(f"{JAVA_API_BASE}/api/brand/tasks/{task_id}")
+        resp_data = resp.json()
+        if resp_data.get("success"):
+            task = resp_data["data"]["task"]
+            return jsonify({
+                'success': True,
+                'task': task
+            })
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/tasks/<int:task_id>/line-progress')
+# @login_required
+def api_brand_task_line_progress(task_id):
+    """
+    获取任务的“当前文件行级进度”（当前行数/总行数）。
+    该信息仅存放在内存字典 _task_line_progress 中，不写入数据库。
+    """
+    try:
+        resp = requests.get(f"{JAVA_API_BASE}/api/brand/tasks/{task_id}")
+        resp_data = resp.json()
+        if resp_data.get("success"):
+            if resp_data["data"]["line_progress"]["has_progress"]:
+                info = resp_data["data"]["line_progress"]["info"]
+                resp = {
+                    'file_index': int(info.get('file_index') or 0),
+                    'file_total': int(info.get('file_total') or 0),
+                    'file_name': info.get("file_name",""),
+                    'current_line': int(info.get('current_line') or 0),
+                    'total_lines': int(info.get('total_lines') or 0),
+                }
+                return jsonify({'success': True, 'has_progress': True, 'info': resp})
+        raise RuntimeError("获取进度失败")
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/tasks/<int:task_id>/events')
+# @login_required
+def api_brand_task_events(task_id):
+    """SSE：任务完成时后端主动推送事件，前端监听后隐藏进度条与取消按钮"""
+    user_id, err = _get_user_id_or_error()
+    if err:
+        return err
+
+    def _task_status_event():
+        conn = get_db()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    "SELECT status, error_message FROM brand_crawl_tasks WHERE id = %s AND user_id = %s",
+                    (task_id, user_id)
+                )
+                row = cur.fetchone()
+        finally:
+            conn.close()
+        if not row:
+            return None
+        st = (row.get('status') or '').lower()
+        if st in ('success', 'failed', 'cancelled'):
+            return {'status': st, 'error_message': (row.get('error_message') or '').strip() or None}
+        return None
+
+    # 若任务已处于终态，直接返回一条事件后结束
+    ev = _task_status_event()
+    if ev is not None:
+        def _one_shot():
+            yield "data: " + json.dumps(ev, ensure_ascii=False) + "\n\n"
+        return Response(
+            _one_shot(),
+            mimetype='text/event-stream',
+            headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
+        )
+
+    # 否则注册队列，等待后台任务完成时推送
+    q = Queue()
+    with _task_event_lock:
+        _task_event_queues.setdefault(task_id, []).append(q)
+
+    def _stream():
+        try:
+            while True:
+                try:
+                    event = q.get(timeout=20)
+                    yield "data: " + json.dumps(event, ensure_ascii=False) + "\n\n"
+                    return
+                except Empty:
+                    yield ": keepalive\n\n"
+        finally:
+            with _task_event_lock:
+                lst = _task_event_queues.get(task_id, [])
+                if q in lst:
+                    lst.remove(q)
+                if not lst:
+                    _task_event_queues.pop(task_id, None)
+
+    return Response(
+        _stream(),
+        mimetype='text/event-stream',
+        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'}
+    )
+
+
+@brand_bp.route('/api/brand/tasks/<int:task_id>/cancel', methods=['POST'])
+# @login_required
+def api_brand_task_cancel(task_id):
+    """取消正在执行或等待中的任务（pending/running 均可取消）"""
+    try:
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
+
+
+        resp = requests.post(
+            f"{JAVA_API_BASE}/api/brand/tasks/{task_id}/cancel",
+            headers={"accept": "application/json"},
+            timeout=20
+        )
+        try:
+            resp_data = resp.json()
+        except Exception:
+            return jsonify({'success': False, 'error': '取消接口返回非 JSON'}), 500
+        if not resp_data.get('success'):
+            return jsonify({'success': False, 'error': '第三方取消任务失败'}), 400
+
+        _mark_task_cancelled(task_id)
+        _push_task_event(task_id, {'status': 'cancelled'})
+        return jsonify({'success': True})
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/tasks/<int:task_id>', methods=['DELETE'])
+@login_required
+def api_brand_task_delete(task_id):
+    """删除任务（仅限非 running 状态）"""
+    try:
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
+
+        resp = requests.delete(
+            f"{JAVA_API_BASE}/api/brand/tasks/{task_id}",
+            headers={"accept": "application/json"},
+            timeout=20
+        )
+        try:
+            resp_data = resp.json()
+        except Exception:
+            return jsonify({'success': False, 'error': '取消接口返回非 JSON'}), 500
+        if not resp_data.get('success'):
+            return jsonify({'success': False, 'error': '第三方取消任务失败'}), 400
+        return jsonify({'success': True})
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+
+@brand_bp.route('/api/brand/download/<int:task_id>')
+@login_required
+def api_brand_download(task_id):
+    """下载任务结果：优先使用 OSS zip_url 重定向；否则本地/单链接/多链接按原逻辑处理"""
+    try:
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
+
+        conn = get_db()
+        with conn.cursor() as cur:
+            cur.execute(
+                "SELECT result_paths FROM brand_crawl_tasks WHERE id = %s AND user_id = %s",
+                (task_id, user_id)
+            )
+            row = cur.fetchone()
+        conn.close()
+        if not row or not row.get('result_paths'):
+            return jsonify({'success': False, 'error': '无结果可下载'}), 404
+        raw = row['result_paths']
+        if isinstance(raw, str):
+            raw = _parse_json(raw)
+        url_list, zip_url = _normalize_result_paths(raw)
+        # 新格式：有 zip_url 直接重定向到 OSS
+        if zip_url and _is_url(zip_url):
+            return redirect(zip_url, code=302)
+
+        # 兼容旧格式：result_paths 可能为 list（本地路径或 url）
+        if not url_list and isinstance(raw, list):
+            url_list = raw
+        local_paths = [p for p in url_list if isinstance(p, str) and not _is_url(p) and os.path.isfile(p)]
+        url_paths = [p.strip() for p in url_list if isinstance(p, str) and _is_url(p)]
+
+        if len(url_paths) == 1 and not local_paths:
+            return redirect(url_paths[0], code=302)
+
+        if len(local_paths) == 1 and not url_paths:
+            return send_file(
+                local_paths[0],
+                as_attachment=True,
+                download_name=os.path.basename(local_paths[0])
+            )
+
+        if local_paths or url_paths:
+            buf = io.BytesIO()
+            with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
+                for p in local_paths:
+                    zf.write(p, os.path.basename(p))
+                for url in url_paths:
+                    try:
+                        import requests as req
+                        r = req.get(url, timeout=30, stream=True)
+                        r.raise_for_status()
+                        name = os.path.basename(urlparse(url).path) or ('file_%s' % (url_paths.index(url)))
+                        if not name or name == 'file_%s' % url_paths.index(url):
+                            name = 'download_%s' % url_paths.index(url)
+                        zf.writestr(name, r.content)
+                    except Exception:
+                        pass
+            buf.seek(0)
+            return send_file(
+                buf,
+                mimetype='application/zip',
+                as_attachment=True,
+                download_name='brand_task_%s.zip' % task_id
+            )
+
+        return jsonify({'success': False, 'error': '结果文件不存在或链接不可用'}), 404
+    except Exception as e:
+        return jsonify({'success': False, 'error': str(e)}), 500
+
+

app/blueprints/communication.py

@@ -4,7 +4,7 @@ from queue import Empty, Full
 
 from flask import Blueprint, jsonify, request
 
-from config import JSON_TASK_QUEUE
+from config import JSON_TASK_QUEUE,runing_task
 
 
 communication_bp = Blueprint('communication', __name__)
@@ -65,3 +65,18 @@ def del_brand():
         _latest_del_brand_ts = time.time()
         print("【删除品牌】返回任务数据", item)
         return jsonify({'success': True, 'data': item})
+    
+
+@communication_bp.route('/api/amazon/get_all_detail', methods=['GET', 'POST'])
+def get_all_detail():
+    """获取正在执行的删除品牌任务详情"""
+    request_id = _resolve_request_id()
+    if not request_id:  #如果没有提供 id 参数，直接返回所有正在执行的任务详情
+        return jsonify({'success': True, 'data': runing_task})
+    task_info = runing_task.get(request_id)
+    if not task_info:
+        return jsonify({'success': False, 'message': '任务不存在', 'data': None})
+    return jsonify({'success': True, 'data': task_info})
+
+
+

app/blueprints/main.py

@@ -110,6 +110,14 @@ 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)
 
     try:
         # 使用流式请求

app/config.py

@@ -20,6 +20,14 @@ proxy_mode = int(os.getenv("proxy_mode",1))
 client_name=os.getenv("client_name") + ".exe"
 JAVA_API_BASE = os.getenv("java_api_base", "http://127.0.0.1:18080")
 
+# 紫鸟浏览器配置
+ZN_COMPANY = os.getenv("zn_company", "")
+ZN_USERNAME = os.getenv("zn_username", "")
+ZN_PASSWORD = os.getenv("zn_password", "")
+
+# 删除品牌API配置
+DELETE_BRAND_API_BASE = os.getenv("DELETE_BRAND_API_BASE", JAVA_API_BASE)
+
 
 cache_path = "./user_data"
 
@@ -40,7 +48,7 @@ os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6"
 
 
 debug = True
-version = "1.0.22"
+version = "1.0.13"
 APP_UPDATE_URL = f"{_base_url}/api/version/latest"  # 检测更新接口地址，返回 { "file_url": "...", "version": "x.x.x" }
 os.environ['APP_VERSION'] = version
 os.environ['APP_UPDATE_URL'] = APP_UPDATE_URL
@@ -49,6 +57,34 @@ os.environ['APP_UPDATE_URL'] = APP_UPDATE_URL
 # 队列
 JSON_TASK_QUEUE = Queue()
 
+#删除品牌，正在执行中的任务
+runing_task = {
+
+}
+# 正在执行中的店铺
+runing_shop = {}
+
+# 
+# runing_task = {
+#     "497": {
+#         "status": "running",              # running/completed/failed
+#         "start_time": "2026-04-01 10:00:00",
+#         "stop_requested": False,
+#         "current_shop": "魏振峰",
+#         "current_country": "德国",
+#         "current_asin": "B0D451RQRG",
+#         "total_shops": 1,
+#         "processed_shops": 0,
+#         "total_asins": 14,
+#         "processed_asins": 3,
+#         "success_count": 2,
+#         "failed_count": 1
+#     }
+# }
+
+
+
+
 
 
 

app/main.py

@@ -1,26 +1,35 @@
 """
 基于 pywebview 实现的跨平台 UI，需先登录方可使用
 """
-from config import  cache_path,debug,version,JAVA_API_BASE,JSON_TASK_QUEUE
+from config import  cache_path,debug,version,JAVA_API_BASE,JSON_TASK_QUEUE,runing_task,runing_shop
 import datetime
 import json
 import sys
+import shutil
 import os
 os.makedirs(cache_path,exist_ok=True)
 if not debug:
     today = datetime.datetime.now().strftime("%Y_%m_%d")
     sys.stdout = open(os.path.join(cache_path,f'{today}.log'), 'a',buffering=1)
-    # sys.stderr = sys.stdout
+    sys.stderr = sys.stdout
 
 import webview
 import threading
 import time
 import requests
+import subprocess
 
 
 with open("version.txt","w",encoding="utf-8") as file:
     file.write(json.dumps({"version":version}))
 
+print(f"""
+========================================================
+版本： {version}
+========================================================
+
+""")
+
 
 # 获取当前脚本所在目录
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -170,7 +179,6 @@ class WindowAPI:
 
     def save_template_xlsx(self):
         """弹窗选择保存位置，将品牌文档格式模板 xlsx 保存到用户选择的位置。"""
-        import shutil
         template_name = '品牌文档格式_模板.xlsx'
         template_path = os.path.join(BASE_DIR, 'static', template_name)
         if not os.path.isfile(template_path):
@@ -189,7 +197,6 @@ class WindowAPI:
         except Exception as e:
             return {'success': False, 'error': str(e)}
 
-
     def save_template_zip(self):
         """弹窗选择保存位置，将品牌文档格式模板 xlsx 保存到用户选择的位置。"""
         import shutil
@@ -234,6 +241,21 @@ class WindowAPI:
         except Exception as e:
             return {'success': False, 'error': str(e)}
 
+    def open_external_url(self, url):
+        if not url or not str(url).strip():
+            return {'success': False, 'error': '链接为空'}
+        target = str(url).strip()
+        try:
+            if sys.platform.startswith('win'):
+                os.startfile(target)
+            elif sys.platform == 'darwin':
+                subprocess.Popen(['open', target])
+            else:
+                subprocess.Popen(['xdg-open', target])
+            return {'success': True}
+        except Exception as e:
+            return {'success': False, 'error': str(e)}
+
     def enqueue_json(self, data):
         """保存任务到队列"""
         try:
@@ -245,83 +267,33 @@ class WindowAPI:
                 payload = json.loads(data)
             if not isinstance(payload, (dict, list)):
                 return {'success': False, 'error': '仅支持 JSON 对象或数组'}
+            if payload.get("type") == "delete-brand-run" and payload.get("data").get("taskId") in runing_task:
+                return {'success': False, 'error': '已存在正在执行的删除品牌任务'}
+            # 判断当前店铺是否正在运行中
+            if payload.get("data").get("items"):
+                shop_name = payload["data"]["items"][0]["shopName"]
+                if shop_name in runing_shop:
+                    return {'success': False, 'error': '当前店铺正在执行中,请等待店铺完成之后重试'}
             JSON_TASK_QUEUE.put(payload)
             return {'success': True, 'queue_size': JSON_TASK_QUEUE.qsize()}
         except Exception as e:
             return {'success': False, 'error': str(e)}
 
-def _select_folder_win32():
-    """Windows: 用 ctypes 调用系统文件夹选择对话框，打包成 exe 后也能正常弹窗（不依赖 tkinter）"""
-    try:
-        import ctypes
-        from ctypes import wintypes
-        BIF_RETURNONLYFSDIRS = 0x00000001
-        BIF_NEWDIALOGSTYLE = 0x00000040
-        MAX_PATH = 260
-        shell32 = ctypes.windll.shell32  # type: ignore[attr-defined]
-        # BROWSEINFOW
-        class BROWSEINFOW(ctypes.Structure):
-            _fields_ = [
-                ("hwndOwner", wintypes.HWND),
-                ("pidlRoot", ctypes.c_void_p),
-                ("pszDisplayName", ctypes.c_wchar_p),
-                ("lpszTitle", ctypes.c_wchar_p),
-                ("ulFlags", wintypes.UINT),
-                ("lpfn", ctypes.c_void_p),
-                ("lParam", wintypes.LPARAM),
-                ("iImage", ctypes.c_int),
-            ]
-        display_name_buf = ctypes.create_unicode_buffer(MAX_PATH)
-        bi = BROWSEINFOW()
-        bi.hwndOwner = 0
-        bi.pidlRoot = 0
-        bi.pszDisplayName = ctypes.cast(display_name_buf, ctypes.c_wchar_p)
-        bi.lpszTitle = "选择保存图片的文件夹"
-        bi.ulFlags = BIF_RETURNONLYFSDIRS | BIF_NEWDIALOGSTYLE
-        bi.lpfn = 0
-        bi.lParam = 0
-        bi.iImage = 0
-        pidl = shell32.SHBrowseForFolderW(ctypes.byref(bi))
-        if not pidl:
-            return ''
-        path_buf = ctypes.create_unicode_buffer(MAX_PATH)
-        if not shell32.SHGetPathFromIDListW(pidl, path_buf):
-            return ''
-        ole32 = ctypes.windll.ole32  # type: ignore[attr-defined]
-        ole32.CoTaskMemFree(pidl)
-        return path_buf.value or ''
-    except Exception as e:
-        print("Windows 文件夹选择失败:", e)
-        return ''
+    def get_detail_del(self,task_id):
+        """获取正在执行的删除品牌任务详情"""
+        task_info = runing_task.get(task_id)
+        if not task_info:
+            return {'success': False, 'error': '任务不存在'}
+        return {'success': True, 'task_info': task_info}
     
-
-def _select_folder_tk():
-    import tkinter as tk
-    from tkinter import filedialog
-
-    root = tk.Tk()
-    root.withdraw()
-    root.attributes('-topmost', True)
-
-    # 使用 after 确保 askdirectory 在主循环启动后执行
-    def ask():
-        nonlocal path
-        path = filedialog.askdirectory(title='选择保存图片的文件夹')
-        root.quit()  # 关闭主循环
-
-    path = ''
-    root.after(100, ask)
-    root.mainloop()  # 启动主循环，直到 root.quit() 被调用
-    try:
-        root.destroy()
-    except:
-        pass
-    return path
-
-
-def select_folder():
-    """弹窗选择文件夹，返回路径（空字符串表示取消）。Windows 打包 exe 时优先用系统 API 避免 tkinter 不弹窗。"""
-    return _select_folder_tk()
+    def stop_task(self,task_id):
+        """停止正在执行的删除品牌任务"""
+        task_info = runing_task.get(task_id)
+        if not task_info:
+            return {'success': False, 'error': '任务不存在'}
+        # 这里可以设置一个标志位，实际的删除品牌任务需要定期检查这个标志位来决定是否停止
+        task_info['stop_requested'] = True
+        return {'success': True, 'message': '已请求停止任务'}
 
 
 def start_flask():
@@ -350,7 +322,7 @@ def main():
     )
     api = WindowAPI(window)
     window.expose(api.close, api.minimize, api.maximize, api.toggle_maximize, generate_images, api.save_image, api.save_image_to_folder, api.select_folder, api.select_brand_xlsx_files, api.select_brand_folder,
-                  api.save_file_from_url, api.save_template_xlsx,api.save_template_zip,api.upload_file_to_java,api.enqueue_json)
+                  api.save_file_from_url, api.save_template_xlsx,api.save_template_zip,api.upload_file_to_java,api.open_external_url,api.enqueue_json,api.get_detail_del)
     webview.start(
         debug=True,
         storage_path=cache_path,