From 785d7563f936305aca7986e37e3a7cf036240a95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=93=AD=E5=9D=A4?= <10372463+ming-kun@user.noreply.gitee.com> Date: Wed, 1 Apr 2026 11:53:12 +0800 Subject: [PATCH] modified: app/ali_oss.py new file: app/amazon/del_brand.py new file: app/amazon/main.py modified: app/blueprints/__pycache__/brand.cpython-39.pyc modified: app/blueprints/__pycache__/communication.cpython-39.pyc modified: app/blueprints/__pycache__/main.cpython-39.pyc modified: app/blueprints/brand.py new file: "app/blueprints/brand_\345\244\207\344\273\275.py" modified: app/blueprints/communication.py modified: app/blueprints/main.py modified: app/config.py modified: app/main.py --- app/ali_oss.py | 1 + app/amazon/del_brand.py | 519 +++++++++++++ app/amazon/main.py | 421 ++++++++++ .../__pycache__/brand.cpython-39.pyc | Bin 20788 -> 20960 bytes .../__pycache__/communication.cpython-39.pyc | Bin 741 -> 2435 bytes .../__pycache__/main.cpython-39.pyc | Bin 4371 -> 4587 bytes app/blueprints/brand.py | 4 +- app/blueprints/brand_备份.py | 726 ++++++++++++++++++ app/blueprints/communication.py | 17 +- app/blueprints/main.py | 8 + app/config.py | 38 +- app/main.py | 126 ++- 12 files changed, 1779 insertions(+), 81 deletions(-) create mode 100644 app/amazon/del_brand.py create mode 100644 app/amazon/main.py create mode 100644 app/blueprints/brand_备份.py diff --git a/app/ali_oss.py b/app/ali_oss.py index 35fbe08..179fadd 100644 --- a/app/ali_oss.py +++ b/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( diff --git a/app/amazon/del_brand.py b/app/amazon/del_brand.py new file mode 100644 index 0000000..cdce0fd --- /dev/null +++ b/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) + + + diff --git a/app/amazon/main.py b/app/amazon/main.py new file mode 100644 index 0000000..17aeaaf --- /dev/null +++ b/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() diff --git a/app/blueprints/__pycache__/brand.cpython-39.pyc b/app/blueprints/__pycache__/brand.cpython-39.pyc index fff3b6d730843c2562e32b2cc7ae8b77afd0952c..7ab38191410dc45eaca633c7a578d3a656f0a5fd 100644 GIT binary patch delta 2625 zcmZ{m4RBP|6@c$~f7yRF8!*9S6G9Ru%g+uF42lT@5>v#mg-J$-2+}2Qu^E!3x%)Co ze9wqmVG)V3*H%UqMdU>Od;}C|67NrMn(GS# zx6!6cA}hmbtqxzME|Yvj{CbmTGz|Er(X7H>;AhQ)Fp&5k4pNj@Q!dBGYI3 z_fe#Zjf&%Q{I)t`*nWG^u)Vg&_HETi0a)MuzPPOp10S}E%mBJkvyfGsfaS3RW^f-o zc5Dw=-W+PtMzMT3)eM=_y3q#Moy-htOk-KqmD|~F*sbhVx2k3_C2G6L1uefF%mOBo za6*=MaMGD+2T>JrrrFa*0;3GDKz)nOYqKAAnsu{+ce#Q9J9d& zla;cEm6_~urB6dYa6v~CPImkOL=5ICC19< z4^r3w-ZFF?1sQtIM%@8Naa2ch^c)*u7UUF6MahA!+M2D8dfa4W7BQvM$nLy#8yVGXszL;7N zJ1a`c-5E|z2+PPa8ag(VfH+x^tSxv`>HgjnPg*J7N&(g0?ldP0UPV}r5I0Ag+KYKP zwFeNLMKYV7Os8%oGutwqUfDL2R+2kpWRj6zDrJHhzK7H~!Xt!Tgonk~qBSKk%`N<1 zQuPwb#kpu5w28k)qf08mgV%H@Qw#_}0DQ;@gcRk9p;60&x)AoyM{O9Txxk+kEtQYL zkT_R)Yk3JxH;hghT2jA;Z;}9B{D4?gb-1{a`f2qTj@N)6%>TA(HIxpag4ZChiYiau z%fCX35piqvJg60Y)irP^|HbMvU_3+mHDa)(Oz@gAxKZq_sfK6s&($0P*pOdW7Xi3U zbj^vu$^5=KH^ZQ`PvD?deuRe3B`iR699Q@}YJ7`u6p@H|Ic+8GAsi#jCe#wVgp-6h z1evPAKTYy+0<994!KA+})WoeW_O=8>d;M(KC=)%E-(Nqj%z6lWF+4pkC}c^iP|!P4 zyf=gT)8grQPw5Y$Zl`FPe?J@*r{<5vt|e4QLu481+e!5jHY@olQ{Q?htdq!uqKq6u~I7sS(x ztLtT*_!{bz-o8W9iv;QSG)dn@h~o0=ztGS;^CeQqBQG&^D#pYe4O<${Qt!)z?-6L3 z_zwx+C;WiWNB9xp6~b%cgNF6BPmuT%0>W5xX1bnhqobA)Tl zB$w{ml;lL*vXEVp@H*jV;`Jqq;fPRg*lGM66$6!`d1<9McEjHTZ(+05-CIY+bB&FL z4Dl9j)A}MaWhJ|NJ``n3tKjYYjZ3?f_3~WCbp9^6{RQEdgkKSIh>n=*?#%*e-a+(7 z5EDZ$j6e3Ji8BYrUpz8?_{7BX&rHRp< za=&~)eHRITBhaei`{YmJ?8-RI6d$c@F#dt6$%vo2v5KM}3`UnI(7OPkNF%fCfaYcpICA2!zMJt-@7 zpE%sQKv5w7W@}1YaFFa)5^f>fN%*9V_zjV|eV02zZk|uF`zD@K_q+6vC~I2+m&Mw) z)&4(|YYh;(7-?HrO&P{!yKj*Y_5GUgKH)cnadEM&Y{mf+4-sUh^c~|vqH^t1wX!C^ zqxJ;h0$~)<>A?>Xet|L;z7yYFTL))^l4yp<#fpRjrQ)STMfk^LD-VXB^6}<>pSS}O z@)aQC$xsxup!4nh`0aU?2FN$sDU#kG(6;7p6V4N8Dfv|b{(WG)R1B@#4heB(UE_ZM DEth=^ delta 2364 zcmZ{l3v3kS7J&aVGdug}Zf)0+QtX3ni*&EB7E7tTDJcc9U2JS5R8g1Cf|c!J|DF1% zv#h#kw}}R>Cm~lJaa%MI6G?DT)a%9T&5asF(TK0r_=*}aXoSQlqQP@!Q8Cdr^Uaxa z&VSDN&-rKelRa=?0!m|nfKSlhwWTi&^d$C|27+)5tAqQ4Rv-%?Os^#@u#~I{!hSWV zDzF5PRxF5kJm7~ikQOcNS2)1AzX07iph{QjJrH6WjNFB-lNvK)nx%AzwMJUC)Vwg% zakypC29eeb&2ViImewJ%FeV(EJ`QPDPNc3MikPb5Hnkkt3{O_2^`W$$6P9l2hNn{8 zt5k`!JE!GcmfLczSH`L2q2y2`r)>i%P2KR)S}hA%C7tn@Zp&l%9)PX;#=!LC)GWvk zO>a&$N(}!1xqw(LjDRLI!MkPoYB4HCVN@AaN451Xqm&Bum_823#iqSxiREQ^Z8YsG zMx?ZzmX|c8=@P3X?;jUH9T=)Gy_xe9{Z%Tl1uNy;_X2$VQ&4Si|6IG6osn6{c zkVy4;9l?m1F`4Ft24gd03*#Q_u5D1{eVDDC1}kuHZOx<#dS`-)4yE+=c!)uM!jEgG zz=QZx?fo!@qjk%m9AB%eg6&x5s=zaK+sdn{Jt>_+3R<9ne8}E4bp=$8agqi?SS_eB zcAMP6ik*0-{t^h`^iTurwyzAm0D7MFEAeF0WNd1j40Eu(u^t|^`x_?!R@?f;L4f60 zH9f5FC&!TxUKN^b-#EP%GS0#%DLdsj$M^?hCZTt)A}`^GCm0iiXxJ+w{NQ2iVT2fs z3@?L&%ZnLK&AMF0@*@mxtIRW;cA?e5$JQOwe4v53uS7SER>pW$N8({47}j# zjmyCl=?~)7Gj_Om(iuE8<4%~ssI_Y%eSrtbR&b?gVy4`Rhayv_9QZAa zY+}i&$7$qq_+#WsNMOgM{qVZ|*rgj){SEdwid8MM++1j{d=h(F>YJSX%av?&qCL&h zQ;b2zLB=x#Z10+a2U-?ZJ+_8NFu4Enzl=kS7Z^i~7a9L! zyo3+W?rGf0;;Rg9aB&9HSaN1Cm!&bRXbnYQCNW%o-hS_7+iQ$J`tQu-k3Bg%DfWf^ zVvHkFtx#faGGmDw5(CM2D(sT)af0_51;z)A4+*_twRrsAVa>a^d0suu zIzR7_SiZOcX4|cc{fhrM zSv%#|>~jLU+v~_W*uE0l@tO92RXeBRE4F>h_@2R|qwmu?UM_zmbjs5# zI&U4HvErV8!mm0P!WnE{*5EeciMScR6|Y)8Q#mW_%<{N4^C9+ZXDnr`W<1N_f{Xb5 zgZKoNcimT9;T)2`^V_G;(q3Ml#s7EBhEsUDYemU-te~F?ArX-cTb+QoyFY@r&2#lV|ciGN2Al9&-wWa#+Qs;gg(!(Fbt@PJ`!g zUvwdC#S_sqRN&34s>@zsU*`qza6P)cebqG(<+oir`J7aowBVAPhRwKqlugbj$!6(w d1|MJfHsd444~(A~XBibZuV)i<;p07Xeg!}WM+*P| diff --git a/app/blueprints/__pycache__/communication.cpython-39.pyc b/app/blueprints/__pycache__/communication.cpython-39.pyc index d33fdb6f402a84505b3e2b2bc2c6df41c077e0cc..5e777ba257047121286f5df3092bac78b3d39ebf 100644 GIT binary patch literal 2435 zcma)8>rWI{6u);~`(9QBs>WK=N6hx6HcgutW2zOKK7wt)7&j!7Vefz=GrQY6Ggio& zpo>;pC}3-)mDF7&RZOV%5kSkX(qT0rPnV(oTxP|FuBDlIK50d=$d9)WRd-* zs~4z|b6DuOozEFvy(~oa&t2ViK)d0iwnMFI`@Xi1)z1#M9ByGT>gJ4Gx9VtiPbQJg zTNbRdnq%a1-TV{$$_9^s3h)r&ejjez0pg(JxGWG{!KY9WuyHYc-4TnBMOmr+o4PbVicHS>CUk{VPO!c9G|q15h-|*I`gD=646F5{_9;J z2X=SVZk+apE-lPVEXgUbaD{x}7&q=cdM+4pY{xb*N28g6KH{;zI^@h(>FWmv{)=80tY&VCePvCf z$HCTaUG|2r2evLuY_tr^Ol@+;-M^-3paMWtXDg592CJ9L)!#394=ye~91HNOetp3k z{+{DHJL`Q_`IoT0`**$JQLuv-_vJOe6hIev0f?$YzoZlXPNp*GvnF+A535-%drVi& ze71+BRDG{d2MFtcdEU19Jk z4I@ALWDYTjyY**WK$^HsWF=T$Mtd(1Bq>;Sy1%&?7ohB*Qe_%!0vs|@=T zju|*C>~}y0vyI3^;qwiopbL}hv2Hr#z5%WlzqtnmuTEX}#&1?H+*qs(^4(F_r9lZ7 zZ~t5^oo%PjgGsOf0zTg^Q)6J;N;iQhEj_6*4FiKFIai?D;lU#O2vv<-SAI3X$)~eF z2--HcL?TBy35ZE>9?;i-lYS2T9HY8M;83P70f)(OinzL8Udl_#8?zR*h`6%w9GmmHIqSFR#qGl zFQUr`E4cW9UKTuy^r9#Kh+MU?xiUAJxY2NcT#F~)qcwEU zJmc-i8vz8yAc1`d!NJDqeF&&VbZVBUZ&7R4CI+bybA*WX3=x|+k3k9h4ygygcNa>| zhr(zlV?Obbu(#rZj#-p)VeaNx8Vx6sZe0tuJA>}!;QHWZuib0+8)23tMH+>eM_D>O zwT{QpI!^F3ORG<)RTKKocF?W9q17Ig6)b62q~)8^O;m*t)a~uHlgw04#O#_OFmzIl7^rKJ?KK_Nvz}7g>kQ- z1zk!AV;JN7F4L5z5*D>EONY^j8Py-_>?L8atl)H+%9<+ZpKGab>td3TBBmSad6xxU L$|YJ2o2WhidfeH8cB`Rlt$0&0~7Z7JIV_)rT2R&CkSpBa@z@XP4oR1J|B<7bsv)El;iRM!eMxWOo($",resp.text) return True diff --git a/app/blueprints/brand_备份.py b/app/blueprints/brand_备份.py new file mode 100644 index 0000000..7336ef3 --- /dev/null +++ b/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/') +# @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//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//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//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/', 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/') +@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 + + diff --git a/app/blueprints/communication.py b/app/blueprints/communication.py index ef56333..ed391f4 100644 --- a/app/blueprints/communication.py +++ b/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}) + + + diff --git a/app/blueprints/main.py b/app/blueprints/main.py index 69519f9..a9717dc 100644 --- a/app/blueprints/main.py +++ b/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: # 使用流式请求 diff --git a/app/config.py b/app/config.py index fa1767f..a371ba2 100644 --- a/app/config.py +++ b/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 +# } +# } + + + + diff --git a/app/main.py b/app/main.py index 721bd10..4cd2924 100644 --- a/app/main.py +++ b/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 _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 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 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,