提交品牌删除更新

app/blueprints/brand.py

@@ -4,12 +4,10 @@
 import os
 import json
 import threading
-import datetime
 import traceback
 import zipfile
 import io
 import time
-import tempfile
 import requests
 from queue import Queue, Empty
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -18,7 +16,7 @@ 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
+from config import bucket_path,JAVA_API_BASE
 from brand_spider.main import single_file_handle, TaskCancelledError
 
 brand_bp = Blueprint('brand', __name__)
@@ -39,6 +37,64 @@ _task_event_lock = threading.Lock()
 # }
 _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):
@@ -54,10 +110,14 @@ def _push_task_event(task_id, event):
     # 任务进入终态时，顺便清理行级进度缓存
     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 文件的绝对路径列表（保留旧逻辑，不再用于品牌工具文件夹上传）"""
+    """返回文件夹下所有 .xlsx 文件的绝对路径列表"""
     if not folder_path or not os.path.isdir(folder_path):
         return []
     paths = []
@@ -67,26 +127,6 @@ def _expand_folder_xlsx(folder_path):
     return paths
 
 
-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
-
-
 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):
@@ -99,7 +139,7 @@ def _upload_local_xlsx_to_oss(local_path, user_id):
     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/{int(time.time() * 1000)}/{user_id}_{safe_base}"
+    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()
@@ -109,347 +149,125 @@ def _upload_local_xlsx_to_oss(local_path, user_id):
         return None, str(e)
 
 
-def _run_brand_files(file_paths, task_id=None, check_cancelled=None, strategy="Terms"):
-    """对多个 xlsx 执行 single_file_handle（线程池多线程），返回 (result_paths, error_message)。
-    check_cancelled: 可调用对象，返回 True 时中止并返回 (result_paths, 'cancelled')。
-    task_id 存在时会在每处理完一个文件后更新 progress_current。
-    strategy: 品牌匹配方式，默认 Terms，可选 Simple。"""
-    os.makedirs(BRAND_OUTPUT_DIR, exist_ok=True)
-    subdir = str(task_id) if task_id else ('immediate_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S'))
-    out_dir = os.path.join(BRAND_OUTPUT_DIR, subdir)
-    os.makedirs(out_dir, exist_ok=True)
+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)
 
-    # 先筛出有效路径并生成 (fp, out_path) 列表
-    tasks = []
-    for fp in file_paths:
-        if not fp or not isinstance(fp, str) or not os.path.isfile(fp):
-            continue
-        base = os.path.splitext(os.path.basename(fp))[0]
-        safe_base = "".join(c if c.isalnum() or c in '-_' else '_' for c in base)
-        out_path = os.path.join(out_dir, safe_base + '_result.xlsx')
-        tasks.append((fp, out_path))
-
-    if not tasks:
-        return [], None
-
-    result_paths = []
-    result_lock = threading.Lock()
-    current = [0]  # 用列表以便在闭包中修改
-
-    def process_one(fp, out_path, file_index, files_total):
-        # 行级进度回调：由 single_file_handle 在循环中调用，直接写入内存字典，不落库
-        def progress_callback(current_line, total_lines, extra=None):
-            if not task_id:
-                return
-            info = {
-                'file_index': file_index,
-                'file_total': files_total,
-                'file_path': fp,
-                'current_line': int(current_line or 0),
-                'total_lines': int(total_lines or 0),
-            }
-            if isinstance(extra, dict):
-                info.update(extra)
-            with _task_line_progress_lock:
-                _task_line_progress[task_id] = info
-
-        single_file_handle(fp, out_path, strategy, check_cancelled=check_cancelled, progress_callback=progress_callback)
-        return out_path
-
-    max_workers = min(8, len(tasks))
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        future_to_out = {
-            executor.submit(process_one, fp, out_path, idx + 1, len(tasks)): out_path
-            for idx, (fp, out_path) in enumerate(tasks)
-        }
-        try:
-            for future in as_completed(future_to_out):
-                if check_cancelled and check_cancelled():
-                    for f in future_to_out:
-                        f.cancel()
-                    return result_paths, 'cancelled'
-                try:
-                    out_path = future.result()
-                    with result_lock:
-                        result_paths.append(out_path)
-                        current[0] += 1
-                        if task_id:
-                            try:
-                                conn = get_db()
-                                with conn.cursor() as cur:
-                                    cur.execute(
-                                        "UPDATE brand_crawl_tasks SET progress_current = %s WHERE id = %s",
-                                        (current[0], task_id)
-                                    )
-                                conn.commit()
-                                conn.close()
-                            except Exception:
-                                print(traceback.format_exc())
-                except TaskCancelledError:
-                    for f in future_to_out:
-                        f.cancel()
-                    return result_paths, 'cancelled'
-                except Exception as e:
-                    print(traceback.format_exc())
-                    print(e)
-                    for f in future_to_out:
-                        f.cancel()
-                    return result_paths, str(e)
-        except Exception as e:
-            print(traceback.format_exc())
-            return result_paths, str(e)
-    return result_paths, None
+    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 _upload_results_to_oss(source_paths, local_result_paths, task_id):
+
+def _background_brand_task(task_id,data):
     """
-    将源文件和结果文件上传到 OSS，并打包成 zip（分两个文件夹：源文件、结果文件）上传。
-    返回 (result_data, error_message)，result_data 为 {"urls": [url1, ...], "zip_url": "..."}，出错时为 (None, err)。
+        后台执行爬虫任务
+        参数示例：
+            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"
+                ]
+              }
+            ]
+          }
     """
-    if not local_result_paths or not task_id:
-        return None, "无结果文件"
     try:
-        from ali_oss import upload_file as oss_upload_file
-    except ImportError:
-        return None, "OSS 模块未配置"
-    urls = []
-    subdir = f"{OSS_PREFIX}/{task_id}"
-    for fp in local_result_paths:
-        if not fp or not os.path.isfile(fp):
-            continue
-        name = os.path.basename(fp)
-        key = f"{bucket_path}{subdir}/{name}"
-        try:
-            with open(fp, "rb") as f:
-                content = f.read()
-            url = oss_upload_file(content, key)
-            urls.append(url)
-        except Exception as e:
-            return None, f"上传结果文件失败: {e}"
-    if not urls:
-        return None, "没有可上传的结果文件"
-    # 打包成 zip：源文件 -> 源文件/，结果文件 -> 结果文件/
-    buf = io.BytesIO()
-    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
-        for fp in (source_paths or []):
-            if fp and os.path.isfile(fp):
-                zf.write(fp, os.path.join("源文件", os.path.basename(fp)))
-        for fp in local_result_paths:
-            if fp and os.path.isfile(fp):
-                zf.write(fp, os.path.join("结果文件", os.path.basename(fp)))
-    buf.seek(0)
-    zip_key = f"{bucket_path}{subdir}/result.zip"
-    try:
-        zip_url = oss_upload_file(buf.getvalue(), zip_key)
-    except Exception as e:
-        return {"urls": urls, "zip_url": None}, None  # 单文件链接已保存，zip 失败不报错
-    return {"urls": urls, "zip_url": zip_url}, None
-
-
-def _background_brand_task(task_id):
-    """后台执行单个品牌爬虫任务并更新数据库"""
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "SELECT id, user_id, file_paths, strategy, `desc` FROM brand_crawl_tasks WHERE id = %s AND status = 'pending'",
-                (task_id,)
-            )
-            row = cur.fetchone()
-        conn.close()
-        if not row:
-            return
-        file_paths = row.get('file_paths')
-        if isinstance(file_paths, str):
-            file_paths = json.loads(file_paths) if file_paths else []
-        # 优先使用表里存的 strategy 字段，兼容旧数据则从 desc 解析
-        strategy = (row.get('strategy') or '').strip() or None
-        if not strategy:
-            desc_val = (row.get('desc') or '').strip()
-            strategy = "Simple" if desc_val.startswith('[Simple]') else "Terms"
-        if strategy not in ('Terms', 'Simple'):
-            strategy = 'Terms'
-        if not file_paths:
-            try:
-                conn = get_db()
-                with conn.cursor() as cur:
-                    cur.execute(
-                        "UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
-                        ('没有有效文件路径', task_id)
-                    )
-                conn.commit()
-                conn.close()
-            except Exception:
-                pass
-            _push_task_event(task_id, {'status': 'failed', 'error_message': '没有有效文件路径'})
-            return
-        # 将 file_paths 解析为本地路径：URL 下载到 BRAND_OUTPUT_DIR 对应任务文件夹下（兼容旧数据中的本地路径）
-        os.makedirs(BRAND_OUTPUT_DIR, exist_ok=True)
-        task_subdir = str(task_id) if task_id else ('immediate_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S'))
-        task_dir = os.path.join(BRAND_OUTPUT_DIR, task_subdir)
-        os.makedirs(task_dir, exist_ok=True)
-
-        local_paths = []
-        downloaded_files_to_cleanup_on_fail = []
-        for idx, p in enumerate(file_paths):
-            if not p or not isinstance(p, str):
+        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
-            p = p.strip()
-            if _is_url(p):
-                try:
-                    parsed = urlparse(p)
-                    url_name = os.path.basename(parsed.path or '')
-                    if not url_name:
-                        url_name = f"input_{idx + 1}.xlsx"
-                    safe_name = "".join(c if c.isalnum() or c in '-_.' else '_' for c in url_name)
-                    if not safe_name.lower().endswith('.xlsx'):
-                        safe_name = safe_name + '.xlsx'
+            for i in range(0, len(brand_ls), 5):
+                chunk = brand_ls[i:i + 5]
+                tasks.append((chunk, fileUrl))
 
-                    base_no_ext, ext = os.path.splitext(safe_name)
-                    dest = os.path.join(task_dir, safe_name)
-                    n = 1
-                    while os.path.exists(dest):
-                        dest = os.path.join(task_dir, f"{base_no_ext}_{n}{ext}")
-                        n += 1
-
-                    r = requests.get(p, timeout=120, stream=True)
-                    r.raise_for_status()
-                    with open(dest, "wb") as f:
-                        for chunk in r.iter_content(chunk_size=1024 * 1024):
-                            if chunk:
-                                f.write(chunk)
-                    local_paths.append(dest)
-                    downloaded_files_to_cleanup_on_fail.append(dest)
-                except Exception as e:
-                    for tf in downloaded_files_to_cleanup_on_fail:
-                        try:
-                            os.unlink(tf)
-                        except Exception:
-                            pass
-                    try:
-                        conn = get_db()
-                        with conn.cursor() as cur:
-                            cur.execute(
-                                "UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
-                                (f'下载文件失败: {str(e)}', task_id)
-                            )
-                        conn.commit()
-                        conn.close()
-                    except Exception:
-                        pass
-                    _push_task_event(task_id, {'status': 'failed', 'error_message': f'下载文件失败: {str(e)}'})
+        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
-            elif os.path.isfile(p):
-                local_paths.append(p)
-        if not local_paths:
-            try:
-                conn = get_db()
-                with conn.cursor() as cur:
-                    cur.execute(
-                        "UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
-                        ('没有有效的本地或可下载文件', task_id)
-                    )
-                conn.commit()
-                conn.close()
-            except Exception:
-                pass
-            _push_task_event(task_id, {'status': 'failed', 'error_message': '没有有效的本地或可下载文件'})
-            return
-        try:
-            # 原子地将 pending 改为 running，避免用户先点取消时仍被覆盖为 running
-            try:
-                conn = get_db()
-                with conn.cursor() as cur:
-                    cur.execute(
-                        "UPDATE brand_crawl_tasks SET status = 'running', progress_total = %s, progress_current = 0 WHERE id = %s AND status = 'pending'",
-                        (len(local_paths), task_id)
-                    )
-                    n = cur.rowcount
-                conn.commit()
-                conn.close()
-                if n == 0:
-                    conn = get_db()
-                    with conn.cursor() as cur:
-                        cur.execute("SELECT status FROM brand_crawl_tasks WHERE id = %s", (task_id,))
-                        r = cur.fetchone()
-                    conn.close()
-                    if r and (r.get('status') or '') == 'cancelled':
+                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
-            except Exception:
-                pass
-
-            def check_cancelled():
-                try:
-                    conn = get_db()
-                    with conn.cursor() as cur:
-                        cur.execute("SELECT status FROM brand_crawl_tasks WHERE id = %s", (task_id,))
-                        row = cur.fetchone()
-                    conn.close()
-                    return row and (row.get('status') or '') == 'cancelled'
-                except Exception:
-                    return False
-
-            result_paths, err = _run_brand_files(local_paths, task_id=task_id, check_cancelled=check_cancelled, strategy=strategy)
-            if err:
-                try:
-                    conn = get_db()
-                    with conn.cursor() as cur:
-                        st = 'cancelled' if err == 'cancelled' else 'failed'
-                        cur.execute(
-                            "UPDATE brand_crawl_tasks SET status = %s, error_message = %s WHERE id = %s",
-                            (st, err if st == 'failed' else None, task_id)
-                        )
-                    conn.commit()
-                    conn.close()
-                except Exception as e:
-                    traceback.print_exc()
-                    print(e)
-                    pass
-                _push_task_event(task_id, {'status': st, 'error_message': err if st == 'failed' else None})
-                return
-            result_data, upload_err = _upload_results_to_oss(local_paths, result_paths, task_id)
-            if upload_err:
-                try:
-                    conn = get_db()
-                    with conn.cursor() as cur:
-                        cur.execute(
-                            "UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
-                            (upload_err, task_id)
-                        )
-                    conn.commit()
-                    conn.close()
-                except Exception:
-                    pass
-                _push_task_event(task_id, {'status': 'failed', 'error_message': upload_err})
-                return
-            try:
-                conn = get_db()
-                with conn.cursor() as cur:
-                    cur.execute(
-                        "UPDATE brand_crawl_tasks SET status = 'success', result_paths = %s WHERE id = %s",
-                        (json.dumps(result_data), task_id)
-                    )
-                conn.commit()
-                conn.close()
-            except Exception:
-                pass
-            _push_task_event(task_id, {'status': 'success'})
-        finally:
-            pass
+                        return
+                    future.result()
+        _push_task_event(task_id, {'status': 'success'})
     except Exception as e:
-        try:
-            conn = get_db()
-            with conn.cursor() as cur:
-                cur.execute(
-                    "UPDATE brand_crawl_tasks SET status = 'failed', error_message = %s WHERE id = %s",
-                    (str(e), task_id)
-                )
-            conn.commit()
-            conn.close()
-        except Exception:
-            pass
+        print("执行出错",traceback.format_exc())
         _push_task_event(task_id, {'status': 'failed', 'error_message': str(e)})
+    print("执行完成")
 
 
 def _parse_json(val, default=None):
@@ -494,6 +312,9 @@ def _normalize_result_paths(raw):
 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:
@@ -503,12 +324,33 @@ def api_brand_expand_folder():
     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:
@@ -542,65 +384,58 @@ def api_brand_run():
         if not paths:
             return jsonify({'success': False, 'error': '没有有效的 xlsx 文件路径'}), 400
         # 先上传到 OSS，获取链接再入库
-        user_id = session['user_id']
+        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
-            urls.append(url)
-        desc_core = ', '.join(os.path.basename(p) for p in paths)[:500] if paths else ''
-        desc = f'[{strategy}] ' + (desc_core or '')
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "INSERT INTO brand_crawl_tasks (user_id, file_paths, `desc`, strategy, status, task_type) VALUES (%s, %s, %s, %s, 'pending', %s)",
-                (user_id, json.dumps(urls), desc or None, strategy, task_type)
-            )
-            task_id = cur.lastrowid
-        conn.commit()
-        conn.close()
-        threading.Thread(target=_background_brand_task, args=(task_id,), daemon=True).start()
-        return jsonify({'success': True, 'task_id': task_id})
+            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": ""})
+        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:
+            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
+# @login_required
 def api_brand_tasks():
     """GET: 获取当前用户的任务列表；POST: 添加任务（后台执行）"""
     if request.method == 'GET':
         try:
-            conn = get_db()
-            with conn.cursor() as cur:
-                cur.execute(
-                    """SELECT id, file_paths, `desc`, strategy, status, result_paths, error_message,
-                          COALESCE(progress_current, 0) AS progress_current,
-                          COALESCE(progress_total, 0) AS progress_total,
-                          created_at, updated_at
-                       FROM brand_crawl_tasks WHERE user_id = %s ORDER BY id DESC LIMIT 100""",
-                    (session['user_id'],)
-                )
-                rows = cur.fetchall()
-            conn.close()
-            items = []
-            for r in rows:
-                items.append({
-                    'id': r['id'],
-                    'file_paths': _parse_json(r['file_paths'], []),
-                    'desc': (r.get('desc') or '').strip() or None,
-                    'strategy': (r.get('strategy') or 'Terms').strip() or 'Terms',
-                    'status': r.get('status') or 'pending',
-                    'result_paths': _parse_json(r['result_paths']),
-                    'error_message': (r.get('error_message') or '').strip() or None,
-                    'progress_current': int(r.get('progress_current') or 0),
-                    'progress_total': int(r.get('progress_total') or 0),
-                    'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
-                    'updated_at': r['updated_at'].strftime('%Y-%m-%d %H:%M') if r.get('updated_at') else '',
-                })
-            return jsonify({'success': True, 'items': items})
+            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:
@@ -621,111 +456,95 @@ def api_brand_tasks():
         paths = [p.strip() for p in paths if p and isinstance(p, str)]
         if not paths:
             return jsonify({'success': False, 'error': '请提供至少一个文件路径或链接'}), 400
-        user_id = session['user_id']
+        user_id, err = _get_user_id_or_error()
+        if err:
+            return err
         urls = []
         for p in paths:
-            if _is_url(p):
-                urls.append(p)
-            elif os.path.isfile(p):
-                url, err = _upload_local_xlsx_to_oss(p, user_id)
-                if err:
-                    return jsonify({'success': False, 'error': f'上传文件失败: {os.path.basename(p)} - {err}'}), 500
-                urls.append(url)
-            else:
-                return jsonify({'success': False, 'error': f'无效路径或链接: {p[:80]}'}), 400
-        desc_core = ', '.join(os.path.basename(urlparse(u).path or u) for u in urls)[:500] if urls else ''
-        desc = f'[{strategy}] ' + (desc_core or '')
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "INSERT INTO brand_crawl_tasks (user_id, file_paths, `desc`, strategy, status, task_type) VALUES (%s, %s, %s, %s, 'pending', %s)",
-                (user_id, json.dumps(urls), desc or None, strategy, task_type)
-            )
-            task_id = cur.lastrowid
-        conn.commit()
-        conn.close()
-        # threading.Thread(target=_background_brand_task, args=(task_id,), daemon=True).start()
-        return jsonify({'success': True, 'task_id': task_id})
+            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
+# @login_required
 def api_brand_task_detail(task_id):
     """获取单个任务详情（用于轮询状态）"""
     try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                """SELECT id, file_paths, `desc`, strategy, status, result_paths, error_message,
-                      COALESCE(progress_current, 0) AS progress_current,
-                      COALESCE(progress_total, 0) AS progress_total,
-                      created_at, updated_at
-                   FROM brand_crawl_tasks WHERE id = %s AND user_id = %s""",
-                (task_id, session['user_id'])
-            )
-            row = cur.fetchone()
-        conn.close()
-        if not row:
-            return jsonify({'success': False, 'error': '任务不存在'}), 404
-        return jsonify({
-            'success': True,
-            'task': {
-                'id': row['id'],
-                'file_paths': _parse_json(row['file_paths'], []),
-                'desc': (row.get('desc') or '').strip() or None,
-                'strategy': (row.get('strategy') or 'Terms').strip() or 'Terms',
-                'status': row.get('status') or 'pending',
-                'result_paths': _parse_json(row['result_paths']),
-                'error_message': (row.get('error_message') or '').strip() or None,
-                'progress_current': int(row.get('progress_current') or 0),
-                'progress_total': int(row.get('progress_total') or 0),
-                'created_at': row['created_at'].strftime('%Y-%m-%d %H:%M') if row.get('created_at') else '',
-                'updated_at': row['updated_at'].strftime('%Y-%m-%d %H:%M') if row.get('updated_at') else '',
-            }
-        })
+        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
+# @login_required
 def api_brand_task_line_progress(task_id):
     """
     获取任务的“当前文件行级进度”（当前行数/总行数）。
     该信息仅存放在内存字典 _task_line_progress 中，不写入数据库。
     """
     try:
-        with _task_line_progress_lock:
-            info = _task_line_progress.get(task_id)
-        if not info:
-            return jsonify({'success': True, 'has_progress': False})
-        # 为了避免暴露完整路径，只返回文件名
-        file_name = os.path.basename(info.get('file_path') or '') if info.get('file_path') else ''
-        resp = {
-            'file_index': int(info.get('file_index') or 0),
-            'file_total': int(info.get('file_total') or 0),
-            'file_name': 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})
+        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
+# @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, session['user_id'])
+                    (task_id, user_id)
                 )
                 row = cur.fetchone()
         finally:
@@ -778,21 +597,28 @@ def api_brand_task_events(task_id):
 
 
 @brand_bp.route('/api/brand/tasks/<int:task_id>/cancel', methods=['POST'])
-@login_required
+# @login_required
 def api_brand_task_cancel(task_id):
     """取消正在执行或等待中的任务（pending/running 均可取消）"""
     try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "UPDATE brand_crawl_tasks SET status = 'cancelled' WHERE id = %s AND user_id = %s AND status IN ('pending', 'running')",
-                (task_id, session['user_id'])
-            )
-            n = cur.rowcount
-        conn.commit()
-        conn.close()
-        if n == 0:
-            return jsonify({'success': False, 'error': '任务不存在或无法取消'}), 400
+        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:
@@ -804,17 +630,21 @@ def api_brand_task_cancel(task_id):
 def api_brand_task_delete(task_id):
     """删除任务（仅限非 running 状态）"""
     try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "DELETE FROM brand_crawl_tasks WHERE id = %s AND user_id = %s AND status != 'running'",
-                (task_id, session['user_id'])
-            )
-            n = cur.rowcount
-        conn.commit()
-        conn.close()
-        if n == 0:
-            return jsonify({'success': False, 'error': '任务不存在或正在执行中无法删除'}), 400
+        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
@@ -825,11 +655,15 @@ def api_brand_task_delete(task_id):
 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, session['user_id'])
+                (task_id, user_id)
             )
             row = cur.fetchone()
         conn.close()

app/brand_spider/main.py

@@ -3,10 +3,7 @@ import re
 import random
 import requests
 import time
-import datetime
-import openpyxl
 import unicodedata
-from openpyxl import Workbook
 
 from brand_spider.web_dec import decrypt_via_service, get_guid
 
@@ -77,7 +74,7 @@ def search(hashsearch, data, proxies=None):
     # print("原始结果", enc_text)
 
     # 若返回 Forbidden，则从 config 的 proxy_url 获取代理 IP 后重试，并开启 3 分钟代理窗口
-    if enc_text.strip() == '{"message":"Forbidden"}' and CONFIG_PROXY_URL:
+    if (enc_text.strip() == '{"message":"Forbidden"}' or "Too Many Requests" in enc_text) and CONFIG_PROXY_URL:
         try:
             proxy_resp = requests.get(CONFIG_PROXY_URL, timeout=10)
             print("代理请求结果->：", proxy_resp.text)
@@ -125,7 +122,7 @@ class TaskCancelledError(Exception):
     pass
 
 
-def single_file_handle(file_path, output_path, strategy="Terms", check_cancelled=None, progress_callback=None):
+def single_file_handle(brand, strategy="Terms"):
     status_info = {
         "Ended": "已结束",
         "Expired": "已过期的",
@@ -471,208 +468,123 @@ def single_file_handle(file_path, output_path, strategy="Terms", check_cancelled
         "欧洲联盟"
     ]
     check_staus = ["已过期的","已结束"]
-    # 1. 读取 Excel 活动工作表
-    wb = openpyxl.load_workbook(file_path, data_only=True)
-    active_sheet = wb.active
-    active_sheet_name = active_sheet.title
-    print(active_sheet_name)
-
-    # 2. 将数据转换为列表嵌套字典
-    rows = list(active_sheet.iter_rows(values_only=True))
-    if rows:
-        columns = list(rows[0])
-        data_rows = []
-        for row in rows[1:]:
-            row_dict = {}
-            for idx, col in enumerate(columns):
-                row_dict[col] = row[idx] if idx < len(row) else None
-            data_rows.append(row_dict)
-    else:
-        columns = []
-        data_rows = []
-
-    # 3. 确保必要列存在
-    # required_cols = ["状态", "国家", "时间"]
-    # for col in required_cols:
-    #     if col not in columns:
-    #         columns.append(col)
-    #         for row_dict in data_rows:
-    #             row_dict[col] = ""
-
-    # 4. 提取唯一品牌列表
-    brand_ls = set()
-    for row in data_rows:
-        brand = row.get("品牌")
-        if brand:
-            brand_ls.add(brand)
 
     # 5. 初始化不符合品牌的数据列表
     faild_data = []
     # 初始化查询失败的品牌的数据列表
     query_faild_data = []
 
-    # 6. 对每个品牌进行处理
-    brands = list(brand_ls)
-    total_brands = len(brands)
-    for idx, brand in enumerate(brands, start=1):
-        if check_cancelled and check_cancelled():
-            raise TaskCancelledError()
 
-        # 行级进度回调：当前第 idx 条 / total_brands
-        if progress_callback and total_brands > 0:
-            try:
-                progress_callback(idx, total_brands, extra={'current_brand': str(brand)})
-            except Exception:
-                # 回调失败不影响主流程
-                pass
+    # Simple 策略下最多翻 3 页（0,30,60），直到命中 554 行判断
+    max_pages = 3 if strategy == "Simple" else 1
+    page = 0
+    matched = False
 
-        # Simple 策略下最多翻 3 页（0,30,60），直到命中 554 行判断
-        max_pages = 3 if strategy == "Simple" else 1
-        page = 0
-        matched = False
+    while page < max_pages and not matched:
+        try:
+            as_structure_dict = {
+                "_id": create_code_generator(),
+                "boolean": "AND",
+                "bricks": [
+                    {
+                        "_id": create_code_generator(),
+                        "key": "brandName",
+                        "value": brand,
+                        "strategy": strategy
+                    }
+                ]
+            }
+            # 构造 asStructure 内部 JSON 对象
+            as_structure_dict = as_structure_dict
+            # print(as_structure_dict)
 
-        while page < max_pages and not matched:
-            try:
-                as_structure_dict = {
-                    "_id": create_code_generator(),
-                    "boolean": "AND",
-                    "bricks": [
-                        {
-                            "_id": create_code_generator(),
-                            "key": "brandName",
-                            "value": brand,
-                            "strategy": strategy
-                        }
-                    ]
-                }
-                # 构造 asStructure 内部 JSON 对象
-                as_structure_dict = as_structure_dict
-                # print(as_structure_dict)
+            # 将内部对象转为 JSON 字符串（作为 asStructure 字段的值）
+            as_structure_str = json.dumps(as_structure_dict, ensure_ascii=False)
+            data = {
+                "sort": "score desc",
+                "rows": "30",
+                "asStructure": as_structure_str,
+                "fg": "_void_"
+            }
 
-                # 将内部对象转为 JSON 字符串（作为 asStructure 字段的值）
-                as_structure_str = json.dumps(as_structure_dict, ensure_ascii=False)
-                data = {
-                    "sort": "score desc",
-                    "rows": "30",
-                    "asStructure": as_structure_str,
-                    "fg": "_void_"
-                }
-                # 翻页：第二页 start=30，第三页 start=60（仅 Simple 策略）
-                if strategy == "Simple" and page > 0:
-                    data["start"] = str(30 * page)
+            # 翻页：第二页 start=30，第三页 start=60（仅 Simple 策略）
+            if strategy == "Simple" and page > 0:
+                data["start"] = str(30 * page)
 
-                # print("请求参数", as_structure_dict, "页码:", page)
-                hashsearch = get_guid()
-                enc_text = search(hashsearch, data)
-                print(f"【{hashsearch}】待解密-->", enc_text)
-                dec_text = decrypt_via_service(hashsearch, enc_text)
-                # print(type(dec_text))
-                dec_text = str(dec_text)
-                # print(f"解密之后的结果-->", dec_text)
-                data = json.loads(dec_text)
-            except Exception as e:
-                # 仅第一页请求失败时记录为查询失败品牌
-                if page == 0:
-                    import traceback
-                    traceback.print_exc()
-                    safe_brand = brand.encode('gbk', errors='replace').decode('gbk')
-                    print(f"品牌：{safe_brand},处理失败:{e}")
-                    query_faild_data.append({"品牌": brand, "时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
-                break
+            # print("请求参数", as_structure_dict, "页码:", page)
+            hashsearch = get_guid()
+            enc_text = search(hashsearch, data)
+            print(f"【{hashsearch}】待解密-->", enc_text)
+            dec_text = decrypt_via_service(hashsearch, enc_text)
+            # print(type(dec_text))
+            dec_text = str(dec_text)
+            # print(f"解密之后的结果-->", dec_text)
+            data = json.loads(dec_text)
+        except Exception as e:
+            # 仅第一页请求失败时记录为查询失败品牌
+            if page == 0:
+                import traceback
+                traceback.print_exc()
+                safe_brand = brand.encode('gbk', errors='replace').decode('gbk')
+                print(f"品牌：{safe_brand},处理失败:{e}")
+                # query_faild_data.append({"brand": brand, "time": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
+                query_faild_data.append(brand)
+            break
 
-            # print(data)
+        # print(data)
+
+        if data.get("response", {}).get("numFound", 0) > 0:
+            if strategy == "Simple":
+                new_brand_name = clean_text(brand)
+                _brand = new_brand_name.lower()
+            else:
+                _brand = brand
+
+            docs = data["response"]["docs"]
+            for d in docs:
+                office = d.get("office")
+                status = d.get("status")
+                registrationDate = d.get("registrationDate")
+                status_name = status_info.get(status, status)
+                brand_name = d.get("brandName")
+                if isinstance(brand_name, list):
+                    brand_name = "|".join(brand_name)
 
-            if data.get("response", {}).get("numFound", 0) > 0:
                 if strategy == "Simple":
-                    new_brand_name = clean_text(brand)
-                    _brand = new_brand_name.lower()
-                else:
-                    _brand = brand
+                    new_brand_name = clean_text(brand_name)
+                    new_brand_name = new_brand_name.lower()
+                    if _brand != new_brand_name:
+                        continue
 
-                docs = data["response"]["docs"]
-                for d in docs:
-                    office = d.get("office")
-                    status = d.get("status")
-                    registrationDate = d.get("registrationDate")
-                    status_name = status_info.get(status, status)
-                    brand_name = d.get("brandName")
-                    if isinstance(brand_name, list):
-                        brand_name = "|".join(brand_name)
+                check_office = d.get("designation")
 
-                    if strategy == "Simple":
-                        new_brand_name = clean_text(brand_name)
-                        new_brand_name = new_brand_name.lower()
-                        if _brand != new_brand_name:
+                check_office.append(office)
+                for i in set(check_office):
+                    office_name = country_info.get(i, i)
+                    # print(office_name,office_name in check_value and status_name not in check_staus)
+                    if office_name in check_value and status_name not in check_staus:
+                        # 命中 554 行判断，记录并结束当前品牌后续翻页
+                        # data_rows = [row for row in data_rows if row.get("品牌") != brand]
+                        # print(office_name, office in ["EM","QZ"] and i not in ["EM","QZ"])
+                        if office in ["EM","QZ"] and i not in ["EM","QZ"]:
                             continue
 
-                    check_office = d.get("designation")
+                        if strategy == "Simple":
+                            faild_data.append({"brand": brand, "country": office_name, "status": status_name}) # "reason":"|".join(d.get("brandName","")
+                            matched = True
+                        else:
+                            # print("添加",brand,office_name)
+                            faild_data.append({"brand": brand, "country": office_name, "status": status_name})
+                        # break
+                if matched:
+                    break
 
-                    check_office.append(office)
-                    for i in set(check_office):
-                        office_name = country_info.get(i, i)
-                        # print(office_name,office_name in check_value and status_name not in check_staus)
-                        if office_name in check_value and status_name not in check_staus:
-                            # 命中 554 行判断，记录并结束当前品牌后续翻页
-                            data_rows = [row for row in data_rows if row.get("品牌") != brand]
-                            # print(office_name, office in ["EM","QZ"] and i not in ["EM","QZ"])
-                            if office in ["EM","QZ"] and i not in ["EM","QZ"]:
-                                continue
+        # 如果 Simple 策略下本页未命中，则翻下一页；其他策略不翻页
+        if not matched:
+            page += 1
 
-                            if strategy == "Simple":
-                                faild_data.append({"品牌": brand, "国家": office_name, "状态": status_name, "原因":"|".join(d.get("brandName","")),"时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
-                                matched = True
-                            else:
-                                # print("添加",brand,office_name)
-                                faild_data.append({"品牌": brand, "国家": office_name, "状态": status_name, "时间": datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")})
-                            # break
-                    if matched:
-                        break
 
-            # 如果 Simple 策略下本页未命中，则翻下一页；其他策略不翻页
-            if not matched:
-                page += 1
-                # else:
-                #     # 更新主表数据
-                #     for row in data_rows:
-                #         if row.get("品牌") == brand:
-                #             row["状态"] = status_name
-                #             row["国家"] = office_name
-                #             row["时间"] = datetime.datetime.now().strftime("%y-%m-%d %H:%M:%S")  # 品牌匹配行
-                        # else:
-                        #     # 非匹配行：时间 = 当前国家值（模拟原 mask 逻辑）
-                        #     row["时间"] = ""
-
-    # 7. 创建输出工作簿
-    out_wb = Workbook()
-    out_wb.remove(out_wb.active)  # 删除默认 sheet
-
-    # 主工作表
-    main_ws = out_wb.create_sheet(title=active_sheet_name)
-    main_ws.append(columns)
-    for row_dict in data_rows:
-        main_ws.append([row_dict.get(col, "") for col in columns])
-
-    # 不符合品牌工作表
-    faild_ws = out_wb.create_sheet(title="不符合品牌")
-    if strategy == "Simple":
-        faild_columns = ["品牌", "国家", "状态","原因"]
-    else:
-        faild_columns = ["品牌", "国家", "状态"]
-    faild_ws.append(faild_columns)
-    for row_dict in faild_data:
-        # print("写入",row_dict)
-        faild_ws.append([row_dict.get(col, "") for col in faild_columns])
-
-    # 查询失败品牌工作表
-    query_faild_ws = out_wb.create_sheet(title="查询失败品牌")
-    query_faild_columns = ["品牌", "时间"]
-    query_faild_ws.append(query_faild_columns)
-    for row_dict in query_faild_data:
-        query_faild_ws.append([row_dict.get(col, "") for col in query_faild_columns])
-
-    out_wb.save(output_path)
-    print("生成完成--->", output_path)
-    return output_path
+    return faild_data,query_faild_data
 
 
 if __name__ == '__main__':

app/config.py

@@ -6,9 +6,7 @@ STITCH_WORKFLOW_ID = "7608813873483300907"
 import os
 from dotenv import load_dotenv
 load_dotenv()
-_base_url = os.getenv("base_url","http://159.75.121.33:15124")
-java_api_base = os.getenv("java_api_base", "http://127.0.0.1:18080")
-JAVA_API_BASE = java_api_base
+_base_url = os.getenv("base_url","http://159.75.121.33:15124")
 
 # MySQL 配置
 mysql_host = os.getenv("mysql_host")
@@ -18,8 +16,9 @@ mysql_database = os.getenv("mysql_database","aiimage")
 proxy_url = os.getenv("proxy_url")
 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")
+
 
 cache_path = "./user_data"
 
@@ -39,8 +38,8 @@ os.environ['OSS_ACCESS_KEY_SECRET'] = accessKeySecret
 os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6"
 
 
-debug = True
-version = "1.0.3"
+debug = False
+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