更新地址 更新后端

# Conflicts:
#	app/web_source/admin.html
#	app/web_source/brand-旧.html
#	app/web_source/home.html
#	app/web_source/index.html
#	app/web_source/login.html
#	backend/web_source/login.html
#	web_source/templates_backup/login.html

.gitignore

@@ -43,6 +43,7 @@ MANIFEST
 .installed.cfg
 
 # Build / packaging
+app.zip
 build/
 dist/
 develop-eggs/
@@ -108,3 +109,4 @@ xlsx/
 OPS_REDIS_MYSQL_OPTIMIZATION_NOTES.md
 架构.md
 *ts.%
+.omc

app/.env

@@ -1,9 +1,9 @@
-base_url=http://8.136.19.173:15124
+base_url=http://47.110.241.161:15124
 workflow_id=7608812635877900322
-mysql_host=8.136.19.173
+mysql_host=47.110.241.161
 mysql_user=aiimage
 
-proxy_url=https://api.jikip.com/ip-get?num=1&minute=1&format=json&area=all&protocol=1&mode=2&key=t24g6gi44ubufd8
+proxy_url=https://api.jikip.com/ip-get?num=1&minute=3&format=json&area=all&protocol=1&mode=2&key=t24g6gi44ubufd8
 proxy_mode=2
 
 zn_company=rongchuang123
@@ -11,10 +11,13 @@ zn_username=%E8%87%AA%E5%8A%A8%E5%8C%96_Robot
 
 client_name=ShuFuAI
 
+# java_api_base=http://api.aishufu.top:18080/
+# java_api_base=http://api.aishufu.top:18080/
+java_api_base=http://127.0.0.1:18080/
+
+# 与 Java 后端共享的 JWT 签名密钥，必须与 backend-java 的 AIIMAGE_JWT_SECRET 完全一致
+AIIMAGE_JWT_SECRET=please-change-this-secret-please-rotate-at-least-32-bytes
+
 
-# java_api_base=http://47.111.163.154:18080
-# java_api_base=http://127.0.0.1:18080
-# java_api_base=http://8.136.19.173:18080
-java_api_base=http://121.196.149.225:18080
 
 

app/.env copy

@@ -1,6 +1,6 @@
-base_url=http://8.136.19.173:15124
+base_url=http://47.111.163.154:15124
 workflow_id=7608812635877900322
-mysql_host=8.136.19.173
+mysql_host=47.111.163.154
 mysql_user=aiimage
 
 proxy_url=https://api.jikip.com/ip-get?num=1&minute=1&format=json&area=all&protocol=1&mode=2&key=t24g6gi44ubufd8
@@ -9,7 +9,7 @@ proxy_mode=2
 client_name=ShuFuAI
 
 
-java_api_base=http://127.0.0.1:18080
+java_api_base=http://api.aishufu.top:18080/
-# java_api_base=http://8.136.19.173:18080
+# java_api_base=http://api.aishufu.top:18080/
 
 

app/.env.example

@@ -1,5 +1,5 @@
 workflow_id=7608812635877900322
-mysql_host=8.136.19.173
+mysql_host=47.111.163.154
 mysql_user=aiimage
 
 proxy_url=https://api.jikip.com/ip-get?num=1&minute=1&format=json&area=all&protocol=1&mode=2&key=t24g6gi44ubufd8
@@ -11,8 +11,13 @@ zn_username=%E8%87%AA%E5%8A%A8%E5%8C%96_Robot
 client_name=ShuFuAI
 
 
-java_api_base=http://47.111.163.154:18080
+java_api_base=http://api.aishufu.top:18080/
-# java_api_base=http://127.0.0.1:18080
+# java_api_base=http://api.aishufu.top:18080/
-# java_api_base=http://8.136.19.173:18080
+# java_api_base=http://api.aishufu.top:18080/
+
+# 与 Java 后端共享的 JWT 签名密钥，必须与 backend-java 的 AIIMAGE_JWT_SECRET 完全一致
+AIIMAGE_JWT_SECRET=please-change-this-secret-please-rotate-at-least-32-bytes
+# JWT cookie 名称，默认 aiimage_token；改动需与 Java 端 aiimage.auth.cookie-name 保持一致
+# AIIMAGE_AUTH_COOKIE_NAME=aiimage_token
 
 

app/amazon/amazon_base.py

@@ -417,8 +417,8 @@ class AmamzonBase(ZiniaoDriver):
         """
         try:
             timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-            if level == "ERROR":
+            # if level == "ERROR":
-                show_notification(message, "error")
+            #     show_notification(message, "error")
             print(f"[{timestamp}] [{self.mark_name}] [{level}] {message}")
         except Exception as e:
             print(f"输出出错,{e}")
@@ -689,8 +689,8 @@ class TaskBase:
         """
         try:
             timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-            if level == "ERROR":
+            # if level == "ERROR":
-                show_notification(message, "error")
+            #     show_notification(message, "error")
             print(f"[{timestamp}] [{self.task_name}] [{level}] {message}")
         except Exception as e:
             print(f"输出出错,{e}")
@@ -739,7 +739,7 @@ class TaskBase:
                     if not shop_data:
                         mes = f"获取店铺凭证失败，响应数据: {shop_data.get('message', '未知错误')}"
                         self.log(mes, "ERROR")
-                        show_notification(mes, "ERROR")
+                        # show_notification(mes, "ERROR")
                         continue
 
                     password = shop_data["data"]["password"]

app/amazon/approve.py

@@ -308,8 +308,9 @@ class AmzoneApprove(AmamzonBase):
         self.log("开始执行...")
         num = 0
         retry_num = 0
-
         max_retry_num = 5
+        total_page = 0
+        current_page = 0
         while retry_num < max_retry_num:  # 最多重试3次
             # if num > 3:  #测试
             #     return
@@ -325,13 +326,13 @@ class AmzoneApprove(AmamzonBase):
                     page_pamel = self.tab.eles('xpath://kat-pagination',timeout=5)
                     if len(page_pamel) > 0:
                         current_page = page_pamel[0].sr('xpath:.//ul[@class="pages"]//li[@aria-current="true"]').text
+                        current_page = int(current_page.strip())
 
                         # 总页数
                         total_page = page_pamel[0].sr.eles('xpath:.//ul[@class="pages"]//span[@class="page__inner"][last()]')
                         if len(total_page) > 0:
-                            total_page = total_page[-1].text
+                            total_page = int(total_page[-1].text.strip())
-                        else:
+
-                            total_page = 0
                         self.log(f"当前页码: {current_page} / 总页数: {total_page}")
                 except Exception as e:
                     self.log(f"获取页码失败:{e}")
@@ -543,6 +544,14 @@ class AmzoneApprove(AmamzonBase):
                 self.tab.refresh()
                 self.tab.wait.doc_loaded(raise_err=False,timeout=120)
 
+        #检查页数是否相等，不相等则继续
+        self.log(f"开始检查页数，当前页数 {current_page} / {total_page}")
+        if total_page != 0 and current_page!= 0 and current_page < total_page:
+            self.log(f"检查到页数还未完成，重启浏览器继续")
+            raise RuntimeError(f"与页面的连接已断开，检查到页数还未完成，重启浏览器继续")
+
+
+
 class ApproveTask(TaskBase):
     """审批任务处理类：负责处理产品风险审批任务"""
     task_name = "产品风险审批-TASK"
@@ -674,12 +683,12 @@ class ApproveTask(TaskBase):
                 return
 
             current_url = None
+            max_retries = 200  #只要没有完成，一直重试
             for _ in range(max_retries):
                 try:
                     self.process_country(driver, country_code, task_id, shop_name,risk_listing_filter,target_url=current_url)
                     driver.reset_already_asin()
                     driver.close_store()
-
                     break
                 except Exception as e:
                     self.log(f"处理国家 {country_code} 失败: {str(e)}", "ERROR")
@@ -777,7 +786,7 @@ class ApproveTask(TaskBase):
                 }
             result.append(country_data)
 
-            if len(result) > 20:
+            if len(result) > 10:
                 self.post_result_batch(task_id, shop_name, country_code, result)
                 result = []
 

app/amazon/chrome_base.py

@@ -9,7 +9,7 @@ from datetime import datetime
 from DrissionPage import Chromium, ChromiumOptions
 from collections import defaultdict
 
-from config import base_dir
+from config import base_dir,debug
 from amazon.tool import get_shop_info,show_notification
 
 
@@ -46,6 +46,7 @@ class ChromeAmzoneBase:
         杀死当前谷歌浏览器进程，并使用 drissionpage 启动谷歌浏览器，使用系统安装的浏览器默认用户文件夹
         """
         print("正在关闭现有的chromium浏览器进程...")
+        if not debug:
             os.system('taskkill /f /t /im chrome.exe')
         time.sleep(2)
 
@@ -70,8 +71,8 @@ class ChromeAmzoneBase:
         """
         try:
             timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
-            if level == "ERROR":
+            # if level == "ERROR":
-                show_notification(message, "error")
+            #     show_notification(message, "error")
             print(f"[{timestamp}] [{self.mark_name}] [{level}] {message}")
         except Exception as e:
             print(f"输出出错,{e}")
@@ -116,20 +117,20 @@ class ChromeAmzoneBase:
                 # print(f"当前地址信息: {current_text}")
                 # 先检查标识
                 if mark is not None and mark in current_text:
-                    print(f"邮编检测到标识: {mark}，无需修改")
+                    self.log(f"邮编检测到标识: {mark}，无需修改")
                     return True
                 # 检查是否已经包含目标邮编
                 if zip_code in current_text:
-                    print(f"邮编已经设置为: {zip_code}，无需修改")
+                    self.log(f"邮编已经设置为: {zip_code}，无需修改")
                     return True
 
             # 需要设置邮编
-            print(f"正在设置邮编为: {zip_code}")
+            self.log(f"正在设置邮编为: {zip_code}")
 
             # 点击地址选择按钮
             location_link = self.tab.ele('xpath://a[@id="nav-global-location-popover-link"]', timeout=10)
             if not location_link:
-                print("找不到地址设置按钮")
+                self.log("找不到地址设置按钮")
                 return False
 
             location_link.click()
@@ -193,3 +194,13 @@ class ChromeAmzoneBase:
                 print("浏览器已关闭")
         except Exception as e:
             print(f"关闭浏览器时出错: {str(e)}")
+
+
+
+
+
+
+
+
+
+

app/amazon/detail_spider.py

@@ -13,7 +13,7 @@ from amazon.amazon_base import TaskBase
 from amazon.chrome_base import ChromeAmzoneBase
 
 
-from config import runing_task, runing_shop,base_dir,DELETE_BRAND_API_BASE
+from config import runing_task,runing_shop,base_dir,DELETE_BRAND_API_BASE
 
 
 class ChromeAmzone(ChromeAmzoneBase):
@@ -79,7 +79,8 @@ class ChromeAmzone(ChromeAmzoneBase):
         except Exception as e:
             error_msg = f"运行出错: {traceback.format_exc()}"
             self.log(error_msg)
-            show_notification(f"采集失败: {str(e)}", "error")
+            # show_notification(f"采集失败: {str(e)}", "error")
+            raise RuntimeError(error_msg)
             return {}
 
     def _scrape_data(self):
@@ -91,7 +92,8 @@ class ChromeAmzone(ChromeAmzoneBase):
         """
         data = {
             'image_url': "",
-            'title': ""
+            'title': "",
+            "sku" : ""
         }
 
         try:
@@ -100,9 +102,41 @@ class ChromeAmzone(ChromeAmzoneBase):
             title_ele = self.tab.ele('xpath://h1[@id="title"]',timeout=30)
             title = title_ele.text
             data["title"] = title
+            sku_ele_ls = self.tab.eles('xpath://ul[@class="a-unordered-list a-vertical a-spacing-mini"]', timeout=20)
+            if len(sku_ele_ls) > 0:
+                data["sku"] = sku_ele_ls[0].text
             imge_ele = self.tab.ele('xpath://div[@id="imgTagWrapperId"]//img',timeout=20)
+
+            image_url = ""
+            min_image_url = ""
+            data_a_dynamic_image = imge_ele.attr("data-a-dynamic-image")
+            if data_a_dynamic_image:
+                dynamic_image_json = json.loads(data_a_dynamic_image)
+                self.log(f"图片信息:{dynamic_image_json}")
+                max_area = 0
+                min_area = 0
+
+                for url, (width, height) in dynamic_image_json.items():
+                    area = width * height
+                    if area > max_area:
+                        max_area = area
+                        image_url = url
+                    if min_area == 0:
+                        min_area = area
+                        min_image_url = url
+                    if area < min_area:
+                        min_area = area
+                        min_image_url = url
+
+            if not image_url:
                 image_url = imge_ele.attr("src")
+
+            if not min_image_url:
+                min_image_url = imge_ele.attr("src")
+
             data["image_url"] = image_url
+            data["min_image_url"] = min_image_url
+
             return data
 
         except Exception as e:
@@ -246,6 +280,7 @@ class SpiderTask(TaskBase):
                             "country": i.get("country"),
                             "url": return_data.get("image_url"),
                             "title": return_data.get("title"),
+                            "sku": return_data.get("sku")
                         }
                          for i in items
                     ]
@@ -263,7 +298,7 @@ class SpiderTask(TaskBase):
                     result.append(res)
                     print("================")
                     is_done = gp_index == len(groups)-1
-                    if len(result) > 20 or is_done:
+                    if len(result) > 10 or is_done:
                         self.post_result(task_id=task_id,chunkIndex=gp_index+1,chunkTotal=len(groups),
                                          asin=asin,item_data=result,is_done=is_done)
                         result = []
@@ -273,18 +308,16 @@ class SpiderTask(TaskBase):
                     is_done = True
                     self.post_result(task_id=task_id, chunkIndex=len(groups), chunkTotal=len(groups),
                                      asin=asin, item_data=result, is_done=is_done)
-
-                try:
-                    chrome.close()
-                except Exception as e:
-                    print("退出浏览器出错",e)
                 # 更新已处理店铺数
                 if task_id in runing_task:
                     runing_task[task_id]["processed_shops"] += 1
             except Exception as e:
                 self.log(f"处理店铺 {task_id} 失败: {str(e)}", "ERROR")
                 self.log(traceback.format_exc(), "ERROR")
-
+            try:
+                chrome.close()
+            except Exception as e:
+                print("退出浏览器出错", e)
             # 更新任务状态
             if task_id in runing_task:
                 if runing_task[task_id].get("stop_requested", False):
@@ -351,7 +384,7 @@ class SpiderTask(TaskBase):
                 time.sleep(2)
 
         self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR")
-        raise RuntimeError("已达到最大重试次数，结果回传最终失败")
+        # raise RuntimeError("已达到最大重试次数，结果回传最终失败")
 
 if __name__ == '__main__':
     # spide = SpiderTask()

app/amazon/match_action.py

@@ -30,14 +30,17 @@ class AmzoneMatchAction(AmamzonBase):
         except Exception as e:
             print(f"【{self.mark_name}】等待加载中消失出错", e)
 
-    def run_page_action(self):
+    def run_page_action(self,skip_asin=[]):
         self.log(f"==================={self.mark_name}=======================")
         self.log("开始执行...")
         num = 0
         retry_num = 0
-        get_page_faild = 0
 
-        while retry_num < 3:  # 最多重试3次
+        max_retry_num = 5
+        total_page = 0
+        current_page = 0
+
+        while retry_num < max_retry_num:
             # if num > 3:  #测试
             #     return
             # 等待加载完成
@@ -45,6 +48,7 @@ class AmzoneMatchAction(AmamzonBase):
                 load_ele = self.tab.eles("xpath://div[contains(@class,'Loader-module__loader')]",timeout=5)
                 if len(load_ele) > 0:
                     load_ele[0].wait.deleted(timeout=3, raise_err=False)
+
                 time.sleep(0.5)
 
                 # 获取当前页码
@@ -52,23 +56,17 @@ class AmzoneMatchAction(AmamzonBase):
                     page_pamel = self.tab.eles('xpath://kat-pagination',timeout=5)
                     if len(page_pamel) > 0:
                         current_page = page_pamel[0].sr('xpath:.//ul[@class="pages"]//li[@aria-current="true"]').text
+                        current_page = int(current_page.strip())
                         # 测试
                         # if current_page > 1:
                         #     break
                         # 总页数
                         total_page = page_pamel[0].sr.eles('xpath:.//ul[@class="pages"]//span[@class="page__inner"][last()]')
                         if len(total_page) > 0:
-                            total_page = total_page[-1].text
+                            total_page = int(total_page[-1].text.strip())
-                        else:
-                            total_page = 0
                         self.log(f"【{self.mark_name}】当前页码: {current_page} / 总页数: {total_page}")
-                    get_page_faild = 0
                 except Exception as e:
                     self.log(f"【{self.mark_name}】获取页码失败", e)
-                    get_page_faild+= 1
-                    if get_page_faild > 2:
-                        show_notification(f"【{self.mark_name}】获取页码失败超3次停止任务！")
-                        break
 
                 # 保存当前URL，用于失败重试时访问
                 try:
@@ -81,12 +79,27 @@ class AmzoneMatchAction(AmamzonBase):
 
                 sku_ls = self.tab.eles("xpath://div[@data-sku]",timeout=10)
                 self.log(f"【{self.mark_name}】获取到 {len(sku_ls)}")
+                if len(sku_ls) == 0:
+                    retry_num += 1
+                    self.log(f"没有获取到SKU列表，重试{retry_num}/{max_retry_num}")
+                    self.tab.refresh()
+                    self.tab.wait.doc_loaded(raise_err=False, timeout=120)
+                    if retry_num == max_retry_num:
+                        raise RuntimeError(f"与页面的连接已断开，没有获取到SKU列表,重试{retry_num}/{max_retry_num}")
+                    continue
+
                 # for sku_ele in sku_ls[0:2]:
                 for sku_ele in sku_ls:
                     # solve_problem = sku_ele.eles('xpath:.//kat-link[@label="解决商品信息问题"]')
 
                     asin = sku_ele.ele('xpath:.//div[contains(@class,"JanusSplitBox-module__container")]//div[contains(@class,"JanusSplitBox-module__panel--") and contains(string(.),"ASIN")]/..//div[last()]',timeout=3).text
-                    self.log(f"【{self.mark_name}】ASIN {asin} 找到....")
+                    self.log(f"ASIN {asin} 找到....")
+                    if asin in skip_asin:
+                        self.log(f"ASIN {asin} 跳过....")
+                        yield (asin,"有最低价跳过")
+                        self.already_asin.add(asin)
+                        continue
+
                     if asin in self.already_asin:
                         self.log(f"【{self.mark_name}】{asin} 已经处理过了，跳过")
                         continue
@@ -134,6 +147,12 @@ class AmzoneMatchAction(AmamzonBase):
                 self.tab.refresh()
                 self.tab.wait.doc_loaded(raise_err=False,timeout=120)
 
+        #检查页数是否相等，不相等则继续
+        self.log(f"开始检查页数，当前页数 {current_page} / {total_page}")
+        if total_page != 0 and current_page!= 0 and current_page < total_page:
+            self.log(f"检查到页数还未完成，重启浏览器继续")
+            raise RuntimeError(f"与页面的连接已断开，检查到页数还未完成，重启浏览器继续")
+
 
 class MatchTak(TaskBase):
     task_name = "匹配价格-TASK"
@@ -228,6 +247,9 @@ class MatchTak(TaskBase):
         shop_name = shop_item.get("shopName", "未知店铺")
         company_name = shop_item.get("companyName", "")
 
+        skip_asins_by_country = shop_item.get("skipAsinsByCountry",{})
+        skipAsinDetailsByCountry = shop_item.get("skipAsinDetailsByCountry",{})
+
         if not company_name:
             self.log(f"店铺 {shop_name} 的公司名称为空，跳过", "WARNING")
             return
@@ -252,6 +274,11 @@ class MatchTak(TaskBase):
                 self.log(f"检测到任务 {task_id} 的暂停请求，停止处理国家", "WARNING")
                 break
 
+            skip_asin = skip_asins_by_country.get(country_code,[])
+            if skipAsinDetailsByCountry :
+                skipAsinDetails = {i.get("asin"):i.get("minimumPrice") for i in skipAsinDetailsByCountry.get(country_code)}
+            else:
+                skipAsinDetails = {}
             # 打开店铺
             driver = self.open_shop(cls=AmzoneMatchAction,max_retries=max_retries, company_name=company_name,
                                     shop_name=shop_name, iskill=iskill)
@@ -262,10 +289,11 @@ class MatchTak(TaskBase):
                 return
 
             current_url = None
+            max_retries = 200
             for _ in range(max_retries):
                 try:
                     self.process_country(driver, country_code, task_id, shop_name,risk_listing_filter,limit,
-                                         target_url=current_url)
+                                         target_url=current_url,skip_asin=skip_asin,skipAsinDetails=skipAsinDetails)
                     driver.reset_already_asin()
                     driver.close_store()
                     break
@@ -297,7 +325,8 @@ class MatchTak(TaskBase):
             del runing_shop[shop_name]
             self.log(f"店铺 {shop_name} 已从执行列表中移除")
 
-    def process_country(self, driver, country_code, task_id, shop_name, risk_listing_filter,limit=None,target_url=None):
+    def process_country(self, driver, country_code, task_id, shop_name, risk_listing_filter,limit=None,target_url=None,
+                        skip_asin=[],skipAsinDetails={}):
         """处理单个国家的审批任务
         """
 
@@ -339,7 +368,7 @@ class MatchTak(TaskBase):
 
         result = []
         # 处理所有需要审批的商品（通过yield获取结果）
-        for asin, status in driver.run_page_action():
+        for asin, status in driver.run_page_action(skip_asin):
             # 检查是否收到暂停请求
             if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
                 self.log(f"检测到任务 {task_id} 的暂停请求，停止处理ASIN", "WARNING")
@@ -351,16 +380,19 @@ class MatchTak(TaskBase):
             if task_id in runing_task:
                 runing_task[task_id]["current_asin"] = asin
                 runing_task[task_id]["processed_asins"] += 1
-
                 runing_task[task_id]["failed_count"] += 1
 
+            minimumPrice = ""
+            if "有最低价跳过" in status:
+                minimumPrice = skipAsinDetails.get(asin)
             result.append({
                                 "asin": asin,
                                 "status": status,
+                                "minimumPrice": minimumPrice,
                                 "done": False
                             })
 
-            if len(result) > 20:
+            if len(result) > 10:
                 self.post_result_batch(task_id, shop_name, country_code,result)
                 result = []
 

app/amazon/similar_asin.py

@@ -10,9 +10,10 @@ from datetime import datetime
 from collections import defaultdict
 import requests
 from urllib.parse import quote
-
+import os
 from curl_cffi import requests as requests_frp
 
+from config import base_dir
 
 
 from amazon.tool import show_notification,get_shop_info,remove_special_characters,split_currency_values
@@ -29,9 +30,9 @@ except ImportError:
     CONFIG_PROXY_MODE = 1
 
 # Forbidden 后 3 分钟内统一使用代理：记录代理生效截止时间与当前代理
-_FORBIDDEN_PROXY_UNTIL = 0.0
+_FORBIDDEN_PROXY_UNTIL_Similar = 0.0
-_FORBIDDEN_PROXY_DICT = None
+_FORBIDDEN_PROXY_DICT_Similar = None
-_FORBIDDEN_PROXY_MINUTES = 0.5
+_FORBIDDEN_PROXY_MINUTES_Similar = 0.5
 
 
 class ChromeAmzone(ChromeAmzoneBase):
@@ -76,7 +77,9 @@ class ChromeAmzone(ChromeAmzoneBase):
         将图片链接（URL 或本地文件路径）转换为 Base64 编码的字符串。
         """
         if image_source.startswith(('http://', 'https://')):
-            response = requests.get(image_source, timeout=10)
+            response = requests.get(image_source, timeout=10,headers={
+                "user-agent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36 Edg/142.0.0.0 18444"
+            })
             response.raise_for_status()  # 非 2xx 状态码将抛出异常
             image_data = response.content
         else:
@@ -94,7 +97,7 @@ class ChromeAmzone(ChromeAmzoneBase):
         标题 :
 
         """
-        global _FORBIDDEN_PROXY_UNTIL, _FORBIDDEN_PROXY_DICT
+        global _FORBIDDEN_PROXY_UNTIL_Similar, _FORBIDDEN_PROXY_DICT_Similar
 
         headers = {
             "accept": "application/json, text/plain, */*",
@@ -146,14 +149,14 @@ class ChromeAmzone(ChromeAmzoneBase):
 
         # 若 3 分钟内曾出现 Forbidden，则直接使用当时保存的代理
         now = time.time()
-        if now < _FORBIDDEN_PROXY_UNTIL and _FORBIDDEN_PROXY_DICT:
+        if now < _FORBIDDEN_PROXY_UNTIL_Similar and _FORBIDDEN_PROXY_DICT_Similar:
-            proxies = _FORBIDDEN_PROXY_DICT
+            proxies = _FORBIDDEN_PROXY_DICT_Similar
             print("处于 Forbidden 代理窗口内，直接使用代理", proxies)
 
         # 发送第一次请求
         try:
             response = requests_frp.post(url, headers=headers, params=params, json=data, impersonate="chrome101",
-                                     cookies=cookie, proxies=proxies)
+                                     cookies=cookie, proxies=proxies,verify=False)
             response.encoding = "utf-8"
 
             # 检查响应状态码和数据有效性
@@ -176,6 +179,7 @@ class ChromeAmzone(ChromeAmzoneBase):
                     need_retry = True
 
             # 如果需要重试且配置了代理URL
+            # need_retry = True
             if need_retry and CONFIG_PROXY_URL and not proxies:
                 try:
                     proxy_resp = requests.get(CONFIG_PROXY_URL, timeout=10)
@@ -204,7 +208,7 @@ class ChromeAmzone(ChromeAmzoneBase):
                         if proxy_ip:
                             proxies = {
                                 "http": f"http://{proxy_ip}",
-                                "https": f"http://{proxy_ip}",
+                                "https": f"https://{proxy_ip}",
                             }
 
                     if proxies:
@@ -215,8 +219,8 @@ class ChromeAmzone(ChromeAmzoneBase):
                         print("代理重试结果状态码:", response.status_code)
 
                         # 记录 3 分钟内都使用该代理
-                        _FORBIDDEN_PROXY_UNTIL = now + _FORBIDDEN_PROXY_MINUTES * 60
+                        _FORBIDDEN_PROXY_UNTIL_Similar = now + _FORBIDDEN_PROXY_MINUTES_Similar * 60
-                        _FORBIDDEN_PROXY_DICT = proxies
+                        _FORBIDDEN_PROXY_DICT_Similar = proxies
                         
                 except Exception as e:
                     print("获取代理或重试失败:", e)
@@ -240,6 +244,7 @@ class ChromeAmzone(ChromeAmzoneBase):
             'image_url': "",
             'title': "",
             'category' : '',
+            'sku' : '',
             'success' : True
         }
 
@@ -250,8 +255,42 @@ class ChromeAmzone(ChromeAmzoneBase):
             title = title_ele.text
             data["title"] = title
             imge_ele = self.tab.ele('xpath://div[@id="imgTagWrapperId"]//img',timeout=20)
+
+            image_url = ""
+            min_image_url = ""
+            data_a_dynamic_image = imge_ele.attr("data-a-dynamic-image")
+            if data_a_dynamic_image:
+                dynamic_image_json = json.loads(data_a_dynamic_image)
+                self.log(f"图片信息:{dynamic_image_json}")
+                max_area = 0
+                min_area = 0
+
+                for url, (width, height) in dynamic_image_json.items():
+                    area = width * height
+                    if area > max_area:
+                        max_area = area
+                        image_url = url
+                    if min_area == 0:
+                        min_area = area
+                        min_image_url = url
+                    if area < min_area:
+                        min_area = area
+                        min_image_url = url
+
+            if not image_url:
                 image_url = imge_ele.attr("src")
+
+            if not min_image_url:
+                min_image_url = imge_ele.attr("src")
+
+
             data["image_url"] = image_url
+            data["min_image_url"] = min_image_url
+
+            # sku
+            sku_ele_ls = self.tab.eles('xpath://ul[@class="a-unordered-list a-vertical a-spacing-mini"]',timeout=20)
+            if len(sku_ele_ls) > 0:
+                data["sku"] = sku_ele_ls[0].text
             category = self.tab.eles('xpath://div[@id="wayfinding-breadcrumbs_feature_div"]//span[@class="a-list-item"]//a[@class="a-link-normal a-color-tertiary"]',timeout=20)
             if len(category) > 0:
                 data["category"] = category[0].text
@@ -271,12 +310,13 @@ class ChromeAmzone(ChromeAmzoneBase):
         Returns:
             dict: 包含采集到的数据
         """
+        return_data = {}
         try:
             # 验证国家是否支持
             if country not in self.country_info:
                 error_msg = f"不支持的国家: {country}，支持的国家有: {list(self.country_info.keys())}"
                 print(error_msg)
-                show_notification(error_msg, "error")
+                # show_notification(error_msg, "error")
                 return None
 
             # 获取国家配置
@@ -290,7 +330,7 @@ class ChromeAmzone(ChromeAmzoneBase):
             domain = base_url.split("/dp/")[0]
             # 拼接新的URL
             product_url = f"{domain}/dp/{asin}"
-            print(f"正在访问: {product_url}")
+            self.log(f"正在访问: {product_url}")
 
             # 打开链接
             self.tab.get(product_url)
@@ -299,35 +339,41 @@ class ChromeAmzone(ChromeAmzoneBase):
 
             self.close_init_popup()
             # 2. 切换国家/设置邮编
-            print(f"正在检查并设置邮编: {zip_code},标识： {mark}")
+            self.log(f"正在检查并设置邮编: {zip_code},标识： {mark}")
             self._set_zip_code(zip_code, mark)
 
             # self.tab.wait.doc_loaded(timeout=5, raise_err=False)
 
             # 3. 抓取数据
-            print("正在抓取商品数据...")
+            self.log("正在抓取商品数据...")
             data = self._scrape_data()
 
+            return_data.update(data)
+
             # 判断是否采集标题出错
             new_size = "220,220"
             pattern = r"\._(?:[A-Z]+)?(\d+)_\."
-            replacement = f"._{new_size}_.jpg"
+            # replacement = f"._{new_size}_.jpg"
             # 使用正则替换
-            image_new_url = re.sub(pattern, lambda m: replacement, data["image_url"])
+            # image_new_url = re.sub(pattern, lambda m: replacement, data["image_url"])
+            image_new_url = data["min_image_url"]
+
+            print(image_new_url)
 
             iamge_base64 = self.image_to_base64(image_new_url)
 
             similar_data = []
             # 4、请求获取插件的数据
+            if image_new_url:
                 for page_num in range(total_page):
                     resp_data = self.get_aliprice_data(
-                    page=page_num,
+                        page=page_num+1,
                         title = data["title"],
                         domain=domain,
                         category=data["category"],
                         imageBase64 = iamge_base64
                     )
-                self.log(f"Aliprice 扩展数据获取:{resp_data}")
+                    self.log(f"Aliprice 扩展数据获取:{str(resp_data)[0:200]}")
                     if len(resp_data.get("data",[])) > 0:
                         for i in resp_data.get("data"):
                             similar_data.append(i)
@@ -340,14 +386,17 @@ class ChromeAmzone(ChromeAmzoneBase):
             data['url'] = product_url
             data['timestamp'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
 
-            print(f"数据抓取完成: {json.dumps(data)}")
+            print(f"数据抓取完成: {json.dumps(return_data)}")
-            return data
+            return_data.update(data)
+
+            return return_data
 
         except Exception as e:
             error_msg = f"运行出错: {traceback.format_exc()}"
             print(error_msg)
-            show_notification(f"采集失败: {str(e)}", "error")
+            # show_notification(f"采集失败: {str(e)}", "error")
-            return {}
+            raise RuntimeError(f"{traceback.format_exc()}")
+            return return_data
 
 
 class SimilarAsinTask(TaskBase):
@@ -407,22 +456,6 @@ class SimilarAsinTask(TaskBase):
             return payload.get("data") or {}
         raise RuntimeError(f"获取货源查询解析载荷失败: {payload}")
 
-    @staticmethod
-    def _count_payload_rows(data):
-        rows = data.get("rows") or data.get("items") or []
-        if isinstance(rows, list) and rows:
-            return len(rows)
-
-        groups = data.get("groups") or []
-        if not isinstance(groups, list):
-            return 0
-
-        total = 0
-        for group in groups:
-            if isinstance(group, dict) and isinstance(group.get("items"), list):
-                total += len(group.get("items"))
-        return total
-
     def _merge_parsed_payload(self, data, parsed_payload):
         payload_rows = parsed_payload.get("allItems") or parsed_payload.get("items") or []
         merged = {
@@ -445,28 +478,15 @@ class SimilarAsinTask(TaskBase):
         try:
             data = task_data.get("data", {})
             task_id = data.get("taskId")
-            if task_id:
-                queued_rows = self._count_payload_rows(data)
-                try:
             parsed_payload = self.fetch_parsed_payload(task_id, data.get("user_id") or data.get("userId") or 1)
             data = self._merge_parsed_payload(data, parsed_payload)
-                    self.log(
-                        f"similar asin task {task_id} loaded full parsed payload from Java, queuedRows={queued_rows}, fullRows={self._count_payload_rows(data)}"
-                    )
-                except Exception as e:
-                    if not queued_rows:
-                        raise
-                    self.log(
-                        f"similar asin task {task_id} failed to load full parsed payload, fallback to queued rows={queued_rows}: {e}",
-                        "WARNING"
-                    )
             groups = self.normalize_groups(data)
+            print(groups)
 
             if not task_id:
                 self.log("任务ID为空，跳过", "WARNING")
                 return
 
-
             self.log(f"开始处理爬取任务 {task_id}，{len(groups)} 个任务")
 
             if not groups:
@@ -501,48 +521,51 @@ class SimilarAsinTask(TaskBase):
                 result = []
                 for gp_index,gp in enumerate(groups):
                     items = gp.get("items", [])
-                    return_data = {}
+                    group_item = []
-
                     for index,value in enumerate(items):
-
+                        print(value)
                         return_data = {
                             'image_url': "",
-                            'title': ""
+                            'title': "",
+                            'category' : '',
+                            'sku' : '',
+                            'success' : False,
+                            'similar_data' : []
                         }
                         asin = value.get("asin")
                         country = value.get("country")
-                        return_data = None
                         for _ in range(max_retry):
                             try:
-                                return_data = chrome.run(country, asin) or {}
+                                return_data = chrome.run(country, asin,total_page=1) or {}
                                 self.log(f"抓取结果->{return_data}")
                                 break
                             except Exception as e:
-                                if "与页面的连接已断开" in str(e):
+                                # if "与页面的连接已断开" in str(e):
                                 chrome = ChromeAmzone()
-                        if not isinstance(return_data, dict):
+                                self.log(f"{asin}抓取数据报错,{e}")
-                            return_data = {}
+                        # if not isinstance(return_data, dict):
-                        if return_data.get("image_url"):
+                        #     return_data = {}
-                            break
+                        # if return_data.get("image_url"):
+                        #     break
 
-                    group_item = [
+                        group_item.append({
-                        {
+                                "sourceFileKey": value.get("sourceFileKey",""),
-                            "sourceFileKey": i.get("sourceFileKey"),
+                                "sourceFilename": value.get("sourceFilename",""),
-                            "sourceFilename": i.get("sourceFilename"),
+                                "rowToken": value.get("rowToken",""),
-                            "rowToken": i.get("rowToken"),
+                                "groupKey": value.get("groupKey",""),
-                            "groupKey": i.get("groupKey"),
+                                "id": value.get("values",{}).get("id",""),
-                            "id": i.get("id"),
+                                "asin": value.get("asin",""),
-                            "asin": i.get("asin"),
+                                "sku" : return_data.get("sku",""),
-                            "country": i.get("country"),
+                                "country": value.get("country",""),
-                            "url": return_data.get("image_url"),
+                                "url": return_data.get("image_url",""),
-                            "title": return_data.get("title"),
+                                "title": return_data.get("title",""),
                                 "done": False,
-                            "urls" : [i.get("ori_picture") for i in return_data.get("similar_data")]
+                                "urls" : [i.get("ori_picture") for i in return_data.get("similar_data",[])[:16]]
-                        }
+                            })
-                         for i in items
-                    ]
                         # task_id: int, chunkIndex:int,chunkTotal: int, country_code: str, asin: str, status: dict,error:str="",
                         #                     item_data:dict={},
+
+
                     res = {
                       "sourceFileKey": gp.get("sourceFileKey"),
                       "sourceFilename": gp.get("sourceFilename"),
@@ -552,11 +575,11 @@ class SimilarAsinTask(TaskBase):
                       "items": group_item
                     }
 
-                    print("================")
+                    # print("================")
                     result.append(res)
-                    print("================")
+                    # print("================")
                     is_done = gp_index == len(groups)-1
-                    if len(result) > 20 or is_done:
+                    if len(result) > 5 or is_done:
                         self.post_result(task_id=task_id,chunkIndex=gp_index+1,chunkTotal=len(groups),
                                          asin=asin,item_data=result,is_done=is_done)
                         result = []
@@ -567,10 +590,7 @@ class SimilarAsinTask(TaskBase):
                     self.post_result(task_id=task_id, chunkIndex=len(groups), chunkTotal=len(groups),
                                      asin=asin, item_data=result, is_done=is_done)
 
-                try:
+
-                    chrome.close()
-                except Exception as e:
-                    print("退出浏览器出错",e)
                 # 更新已处理店铺数
                 if task_id in runing_task:
                     runing_task[task_id]["processed_shops"] += 1
@@ -578,6 +598,11 @@ class SimilarAsinTask(TaskBase):
                 self.log(f"处理店铺 {task_id} 失败: {str(e)}", "ERROR")
                 self.log(traceback.format_exc(), "ERROR")
 
+            try:
+                chrome.close()
+            except Exception as e:
+                print("退出浏览器出错", e)
+
             # 更新任务状态
             if task_id in runing_task:
                 if runing_task[task_id].get("stop_requested", False):
@@ -585,6 +610,7 @@ class SimilarAsinTask(TaskBase):
                     self.log(f"任务 {task_id} 已被暂停！")
                 else:
                     runing_task[task_id]["status"] = "completed"
+
             self.log(f"任务 {task_id} 处理完成！")
 
         except Exception as e:
@@ -644,13 +670,13 @@ class SimilarAsinTask(TaskBase):
                 time.sleep(2)
 
         self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR")
-        raise RuntimeError("已达到最大重试次数，结果回传最终失败")
+        # raise RuntimeError("已达到最大重试次数，结果回传最终失败")
 
 if __name__ == '__main__':
     spide = ChromeAmzone()
 
     resp = spide.run(
-        country="英国", asin="B0F79LCB3X", total_page=2
+        country="德国", asin="B0CJ8SNXXV", total_page=2
     )
     print(resp)
 #     task_data = {

app/app.py

@@ -3,9 +3,8 @@
 按功能拆分为蓝图：认证(auth)、主页面(main)、管理员(admin)、图片(image)、品牌(brand)
 """
 import os
-import secrets
 import logging
-from datetime import timedelta, datetime
+from datetime import datetime
 from logging.handlers import RotatingFileHandler
 
 from flask import Flask
@@ -73,12 +72,6 @@ def create_app():
         supports_credentials=True,
         resources={r"/api/*": {"origins": cors_origins}, r"/login": {"origins": cors_origins}},
     )
-    # 生产环境必须通过环境变量注入固定 SECRET_KEY；否则服务重启会使旧会话失效。
-    app.secret_key = os.environ.get('SECRET_KEY', 'dev-secret-key-change-me')
-    app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7)
-    app.config['SESSION_COOKIE_HTTPONLY'] = True
-    app.config['SESSION_COOKIE_SAMESITE'] = os.environ.get('SESSION_COOKIE_SAMESITE', 'Lax')
-    app.config['SESSION_COOKIE_SECURE'] = os.environ.get('SESSION_COOKIE_SECURE', '0') == '1'
 
     # 注册蓝图（不设 url_prefix，保持原有 URL 路径不变，前端无需改动）
     app.register_blueprint(auth_bp)

app/app_common.py

@@ -1,5 +1,5 @@
 """
-公共模块：数据库连接、初始化、会话校验、装饰器、模板渲染
+公共模块：数据库连接、初始化、JWT 鉴权装饰器、模板渲染
 供各蓝图复用
 """
 import os
@@ -8,9 +8,10 @@ from datetime import timedelta
 from functools import wraps
 
 import pymysql
-from flask import request, redirect, url_for, session, jsonify, render_template, render_template_string
+from flask import request, redirect, url_for, jsonify, render_template, render_template_string, g
 
 from config import mysql_host, mysql_user, mysql_password, mysql_database
+from jwt_util import parse_token, get_token_from_request
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 STATIC_DIR = os.path.join(BASE_DIR, 'static')
@@ -72,9 +73,36 @@ def _render_html_new(template_name: str, **context):
 
 
 
+def _resolve_jwt_user():
+    """从 Authorization/cookie 解析 JWT，命中后写入 flask.g 缓存。无效返回 None。"""
+    cached = getattr(g, "_aiimage_user", None)
+    if cached is not None:
+        return cached or None
+    token = get_token_from_request()
+    payload = parse_token(token) if token else None
+    if not payload:
+        g._aiimage_user = False
+        return None
+    g._aiimage_user = payload
+    g.aiimage_user_id = payload.get("user_id")
+    g.aiimage_username = payload.get("username") or ""
+    g.aiimage_device_id = payload.get("device_id") or ""
+    return payload
+
+
+def current_user_id():
+    user = _resolve_jwt_user()
+    return user.get("user_id") if user else None
+
+
+def current_username():
+    user = _resolve_jwt_user()
+    return user.get("username") if user else ""
+
+
 def _is_session_user_valid():
-    """校验 session 中的 user_id 是否在数据库中仍存在；不存在则清除 session 并返回 False"""
+    """兼容旧调用：判断当前 JWT 用户是否仍存在于数据库。"""
-    uid = session.get('user_id')
+    uid = current_user_id()
     if not uid:
         return False
     try:
@@ -83,23 +111,22 @@ def _is_session_user_valid():
             cur.execute("SELECT id FROM users WHERE id = %s", (uid,))
             row = cur.fetchone()
         conn.close()
-        if not row:
+        return bool(row)
-            session.clear()
-            return False
-        return True
     except Exception:
-        session.clear()
         return False
 
 
 def _get_current_admin_role():
     """获取当前登录用户的管理角色：super_admin / admin / None（非管理员）"""
+    uid = current_user_id()
+    if not uid:
+        return None, None
     try:
         conn = get_db()
         with conn.cursor() as cur:
             cur.execute(
                 "SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
-                (session['user_id'],)
+                (uid,)
             )
             row = cur.fetchone()
         conn.close()
@@ -110,13 +137,19 @@ def _get_current_admin_role():
         return None, None
 
 
+def _unauthorized_response():
+    if request.headers.get('X-Requested-With') == 'XMLHttpRequest' \
+            or request.path.startswith('/api/') \
+            or 'application/json' in (request.headers.get('Accept') or ''):
+        return jsonify({'success': False, 'error': '未登录'}), 401
+    return redirect(url_for('auth.login'))
+
+
 def login_required(f):
     @wraps(f)
     def decorated(*args, **kwargs):
-        if not session.get('user_id') or not _is_session_user_valid():
+        if not current_user_id():
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+            return _unauthorized_response()
-                return jsonify({'success': False, 'error': '未登录'}), 401
-            return redirect(url_for('auth.login'))
         return f(*args, **kwargs)
     return decorated
 
@@ -124,22 +157,24 @@ def login_required(f):
 def admin_required(f):
     @wraps(f)
     def decorated(*args, **kwargs):
-        if not session.get('user_id'):
+        uid = current_user_id()
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+        if not uid:
-                return jsonify({'success': False, 'error': '未登录'}), 401
+            return _unauthorized_response()
-            return redirect(url_for('auth.login'))
         try:
             conn = get_db()
             with conn.cursor() as cur:
-                cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (session['user_id'],))
+                cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (uid,))
                 row = cur.fetchone()
             conn.close()
             if not row or not row.get('is_admin'):
-                if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+                if request.headers.get('X-Requested-With') == 'XMLHttpRequest' \
+                        or request.path.startswith('/api/') \
+                        or 'application/json' in (request.headers.get('Accept') or ''):
                     return jsonify({'success': False, 'error': '需要管理员权限'}), 403
                 return redirect(url_for('main.home'))
         except Exception as e:
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+            if request.headers.get('X-Requested-With') == 'XMLHttpRequest' \
+                    or request.path.startswith('/api/'):
                 return jsonify({'success': False, 'error': str(e)}), 500
             return redirect(url_for('main.home'))
         return f(*args, **kwargs)

app/assets/brand-COze15GJ.js

@@ -0,0 +1 @@
+import{bN as r}from"./java-modules-B8c-YG5x.js";const n="";function s(e){return r(`${n}/api/brand/expand-folder-recursive`,{folder:e})}export{s as e};

app/assets/categorized-timers-JPA-olTr.js

@@ -0,0 +1 @@
+const r=new Map;function c(n){let t=r.get(n);return t||(t=new Map,r.set(n,t)),t}function u(n,t){const e=r.get(n);e&&(e.delete(t),e.size||r.delete(n))}function l(n,t,e){const i=window.setTimeout(()=>{u(n,i),t()},e);return c(n).set(i,{id:i,kind:"timeout",category:n}),i}function a(n,t,e){const i=window.setInterval(t,e);return c(n).set(i,{id:i,kind:"interval",category:n}),i}function f(n,t){if(t==null)return;const i=r.get(n)?.get(t);i?.kind==="interval"?window.clearInterval(t):window.clearTimeout(t),i?.cancel?.(),u(n,t)}function s(n){const t=r.get(n);if(t){for(const e of t.values())e.kind==="interval"?window.clearInterval(e.id):window.clearTimeout(e.id),e.cancel?.();r.delete(n)}}function d(n,t){return new Promise(e=>{const i=window.setTimeout(()=>{u(n,i),e()},t);c(n).set(i,{id:i,kind:"timeout",category:n,cancel:e})})}function m(n){const t=`${n}:`;for(const e of Array.from(r.keys()))(e===n||e.startsWith(t))&&s(e)}function w(n){const t=e=>`${n}:${e}`;return{setTimeout(e,i,o){return l(t(e),i,o)},setInterval(e,i,o){return a(t(e),i,o)},clearTimer(e,i){f(t(e),i)},clearCategory(e){s(t(e))},clearScope(){m(n)},sleep(e,i){return d(t(e),i)}}}export{w as c};

app/assets/listingFilters-BpGOU_pJ.js

@@ -0,0 +1 @@
+const e=[{value:"SearchSuppressed",label:"在搜索结果中禁止显示"},{value:"ApprovalRequired",label:"需要批准"},{value:"Active",label:"在售"},{value:"DetailPageRemoved",label:"详情页面已删除"}];export{e as L};

app/blueprints/admin.py

@@ -1,379 +1,17 @@
 """
-管理员蓝图：用户管理（列表/创建/更新/删除）、生成历史、管理页
+管理员蓝图：仅保留 /admin 页面渲染。
+用户管理、栏目权限、生成历史等接口已全部迁移至 Java 端。
 """
-import json
+from flask import Blueprint
-import pymysql
-from flask import Blueprint, request, jsonify
-from werkzeug.security import generate_password_hash
 
-from app_common import get_db, _render_html, _get_current_admin_role, admin_required, login_required
+from app_common import _render_html, admin_required, login_required
-from flask import session
+from config import JAVA_API_BASE
 
 admin_bp = Blueprint('admin', __name__)
 
 
-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 []
-
-
 @admin_bp.route('/admin')
 @login_required
 @admin_required
 def admin_page():
-    return _render_html('admin.html')
+    return _render_html('admin.html', java_api_base=JAVA_API_BASE)
-
-
-@admin_bp.route('/api/admin/users')
-@admin_required
-def admin_list_users():
-    """分页获取用户列表；支持用户名模糊搜索、指定管理员所属普通用户筛选"""
-    role, current_row = _get_current_admin_role()
-    if not role:
-        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
-    page = max(1, int(request.args.get('page', 1)))
-    page_size = min(50, max(5, int(request.args.get('page_size', 15))))
-    offset = (page - 1) * page_size
-    search_username = (request.args.get('username') or request.args.get('search') or '').strip()
-    created_by_id_arg = request.args.get('created_by_id') or request.args.get('admin_id')
-    created_by_id = int(created_by_id_arg) if created_by_id_arg and str(created_by_id_arg).isdigit() else None
-    if role != 'super_admin':
-        created_by_id = None
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            if role == 'super_admin':
-                where_parts = ["1=1"]
-                params = []
-                if search_username:
-                    where_parts.append("u.username LIKE %s")
-                    params.append("%" + search_username + "%")
-                if created_by_id is not None:
-                    where_parts.append("u.created_by_id = %s")
-                    params.append(created_by_id)
-                where_sql = " AND ".join(where_parts)
-                cur.execute(
-                    """SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
-                          creator.username AS creator_username
-                   FROM users u
-                   LEFT JOIN users creator ON creator.id = u.created_by_id
-                   WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
-                    tuple(params) + (page_size, offset),
-                )
-                rows = cur.fetchall()
-                cur.execute("SELECT COUNT(*) as total FROM users u WHERE " + where_sql, tuple(params))
-                total = cur.fetchone()['total']
-                cur.execute("SELECT id, username FROM users WHERE role = 'admin' ORDER BY id")
-                admins = [{'id': r['id'], 'username': r['username']} for r in cur.fetchall()]
-            else:
-                admin_id = current_row['id']
-                where_parts = ["(u.id = %s OR (u.role = 'normal' AND u.created_by_id = %s))"]
-                params = [admin_id, admin_id]
-                if search_username:
-                    where_parts.append("u.username LIKE %s")
-                    params.append("%" + search_username + "%")
-                where_sql = " AND ".join(where_parts)
-                cur.execute(
-                    """SELECT u.id, u.username, u.is_admin, u.role, u.created_at, u.created_by_id,
-                          creator.username AS creator_username
-                   FROM users u
-                   LEFT JOIN users creator ON creator.id = u.created_by_id
-                   WHERE """ + where_sql + """ ORDER BY u.id LIMIT %s OFFSET %s""",
-                    tuple(params) + (page_size, offset),
-                )
-                rows = cur.fetchall()
-                cur.execute(
-                    "SELECT COUNT(*) as total FROM users u WHERE " + where_sql,
-                    tuple(params),
-                )
-                total = cur.fetchone()['total']
-                admins = []
-            items = [
-                {
-                    'id': r['id'],
-                    'username': r['username'],
-                    'is_admin': bool(r.get('is_admin')),
-                    'role': r.get('role') or 'normal',
-                    'created_by_id': r.get('created_by_id'),
-                    'creator_username': r.get('creator_username') or '',
-                    'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
-                }
-                for r in rows
-            ]
-        conn.close()
-        payload = {
-            'success': True,
-            'items': items,
-            'total': total,
-            'page': page,
-            'page_size': page_size,
-            'current_user_role': role,
-            'admins': admins,
-        }
-        return jsonify(payload)
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})
-
-
-@admin_bp.route('/api/admin/user', methods=['POST'])
-@admin_required
-def admin_create_user():
-    data = request.get_json() or {}
-    username = (data.get('username') or '').strip()
-    password = data.get('password') or ''
-    role, current_row = _get_current_admin_role()
-    if not role:
-        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
-    want_role = (data.get('role') or 'normal').strip() or 'normal'
-    if want_role not in ('admin', 'normal'):
-        want_role = 'normal'
-    if role == 'admin' and want_role == 'admin':
-        return jsonify({'success': False, 'error': '仅超级管理员可创建管理员'})
-    if want_role == 'admin':
-        want_created_by = current_row['id']
-    elif role == 'super_admin':
-        want_created_by = data.get('created_by_id')
-    else:
-        want_created_by = current_row['id']
-    if not username or not password:
-        return jsonify({'success': False, 'error': '用户名和密码不能为空'})
-    if len(username) < 2:
-        return jsonify({'success': False, 'error': '用户名至少2个字符'})
-    if len(password) < 6:
-        return jsonify({'success': False, 'error': '密码至少6个字符'})
-    if want_role == 'normal' and role == 'super_admin' and want_created_by is None:
-        try:
-            conn = get_db()
-            with conn.cursor() as cur:
-                cur.execute("SELECT id FROM users WHERE role = 'admin' ORDER BY id LIMIT 1")
-                r = cur.fetchone()
-            conn.close()
-            want_created_by = r['id'] if r else current_row['id']
-        except Exception:
-            want_created_by = current_row['id']
-    if want_role == 'normal' and want_created_by is None:
-        want_created_by = current_row['id']
-    is_admin = 1 if want_role in ('super_admin', 'admin') else 0
-    pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                "INSERT INTO users (username, password_hash, is_admin, role, created_by_id) VALUES (%s, %s, %s, %s, %s)",
-                (username, pwd_hash, is_admin, want_role, want_created_by),
-            )
-        conn.commit()
-        conn.close()
-        return jsonify({'success': True, 'msg': '用户创建成功'})
-    except pymysql.IntegrityError:
-        return jsonify({'success': False, 'error': '用户名已存在'})
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})
-
-
-@admin_bp.route('/api/admin/user/<int:uid>', methods=['PUT'])
-@admin_required
-def admin_update_user(uid):
-    data = request.get_json() or {}
-    password = data.get('password')
-    want_role = (data.get('role') or '').strip() or data.get('role')
-    role, current_row = _get_current_admin_role()
-    if not role:
-        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
-    if want_role is None and not password:
-        return jsonify({'success': False, 'error': '请提供要修改的内容'})
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
-            target = cur.fetchone()
-            if not target:
-                conn.close()
-                return jsonify({'success': False, 'error': '用户不存在'})
-            if role == 'admin':
-                if target['role'] != 'normal' or target.get('created_by_id') != current_row['id']:
-                    conn.close()
-                    return jsonify({'success': False, 'error': '只能编辑自己创建的普通用户'}), 403
-                want_role = None
-            else:
-                if target.get('role') == 'super_admin':
-                    conn.close()
-                    return jsonify({'success': False, 'error': '不能修改超级管理员'})
-                if want_role == 'super_admin':
-                    return jsonify({'success': False, 'error': '不能将用户设为超级管理员'})
-                if want_role not in ('admin', 'normal', None, ''):
-                    want_role = None
-            if password:
-                if len(password) < 6:
-                    conn.close()
-                    return jsonify({'success': False, 'error': '密码至少6个字符'})
-                pwd_hash = generate_password_hash(password, method='pbkdf2:sha256')
-                cur.execute("UPDATE users SET password_hash = %s WHERE id = %s", (pwd_hash, uid))
-            if want_role is not None and want_role != '':
-                is_admin = 1 if want_role == 'admin' else 0
-                cur.execute(
-                    "UPDATE users SET is_admin = %s, role = %s WHERE id = %s",
-                    (is_admin, want_role, uid),
-                )
-        conn.commit()
-        conn.close()
-        return jsonify({'success': True, 'msg': '更新成功'})
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})
-
-
-@admin_bp.route('/api/admin/user/<int:uid>', methods=['DELETE'])
-@admin_required
-def admin_delete_user(uid):
-    from flask import session
-    if session.get('user_id') == uid:
-        return jsonify({'success': False, 'error': '不能删除当前登录账号'})
-    role, current_row = _get_current_admin_role()
-    if not role:
-        return jsonify({'success': False, 'error': '需要管理员权限'}), 403
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute("SELECT id, role, created_by_id FROM users WHERE id = %s", (uid,))
-            target = cur.fetchone()
-            if not target:
-                conn.close()
-                return jsonify({'success': False, 'error': '用户不存在'})
-            if target.get('role') == 'super_admin':
-                conn.close()
-                return jsonify({'success': False, 'error': '不能删除超级管理员'})
-            if role == 'admin':
-                if target.get('role') != 'normal' or target.get('created_by_id') != current_row['id']:
-                    conn.close()
-                    return jsonify({'success': False, 'error': '只能删除自己创建的普通用户'}), 403
-            cur.execute("DELETE FROM users WHERE id = %s", (uid,))
-            affected = cur.rowcount
-        conn.commit()
-        conn.close()
-        if affected == 0:
-            return jsonify({'success': False, 'error': '用户不存在'})
-        return jsonify({'success': True, 'msg': '删除成功'})
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})
-
-
-@admin_bp.route('/api/admin/user/<int:uid>/column-permissions')
-@login_required
-def admin_user_column_permissions(uid):
-    """获取指定用户的栏目权限列表：当前用户只能查自己，管理员可查任意用户。超级管理员返回全部栏目。"""
-    current_uid = session.get('user_id')
-    menu_type = (request.args.get('menu_type') or '').strip().lower()
-    if current_uid != uid:
-        role, _ = _get_current_admin_role()
-        if not role:
-            return jsonify({'success': False, 'error': '无权查看该用户的栏目权限'}), 403
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute("SELECT role FROM users WHERE id = %s", (uid,))
-            user_row = cur.fetchone()
-            valid_menu_type = menu_type if menu_type in ('app', 'admin') else ''
-            if user_row and (user_row.get('role') or '').strip() == 'super_admin':
-                sql = """
-                    SELECT id, name, column_key, menu_type, route_path, sort_order, created_at
-                    FROM columns
-                """
-                params = []
-                if valid_menu_type:
-                    sql += " WHERE menu_type = %s"
-                    params.append(valid_menu_type)
-                sql += " ORDER BY sort_order ASC, id ASC"
-                cur.execute(sql, tuple(params))
-                rows = cur.fetchall()
-            else:
-                sql = """
-                    SELECT c.id, c.name, c.column_key, c.menu_type, c.route_path, c.sort_order, c.created_at
-                    FROM columns c
-                    INNER JOIN user_column_permission ucp ON ucp.column_id = c.id
-                    WHERE ucp.user_id = %s
-                """
-                params = [uid]
-                if valid_menu_type:
-                    sql += " AND c.menu_type = %s"
-                    params.append(valid_menu_type)
-                sql += " ORDER BY c.sort_order ASC, c.id ASC"
-                cur.execute(sql, tuple(params))
-                rows = cur.fetchall()
-        conn.close()
-        items = [
-            {
-                'id': r['id'],
-                'name': r['name'],
-                'column_key': r['column_key'],
-                'menu_type': r.get('menu_type') or 'app',
-                'route_path': r.get('route_path') or '',
-                'sort_order': r.get('sort_order') or 0,
-                'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r.get('created_at') else '',
-            }
-            for r in rows
-        ]
-        return jsonify({'success': True, 'items': items})
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})
-
-
-@admin_bp.route('/api/admin/history')
-@admin_required
-def admin_history():
-    """管理员分页获取所有生成记录，支持按用户和时间筛选"""
-    page = max(1, int(request.args.get('page', 1)))
-    page_size = min(50, max(10, int(request.args.get('page_size', 15))))
-    offset = (page - 1) * page_size
-    user_id = request.args.get('user_id', type=int)
-    time_start = (request.args.get('time_start') or '').strip()
-    time_end = (request.args.get('time_end') or '').strip()
-    conditions, params = [], []
-    if user_id:
-        conditions.append("h.user_id = %s")
-        params.append(user_id)
-    if time_start:
-        conditions.append("h.created_at >= %s")
-        params.append(time_start)
-    if time_end:
-        conditions.append("h.created_at <= %s")
-        params.append(time_end + ' 23:59:59' if len(time_end) <= 10 else time_end)
-    where_clause = " AND ".join(conditions) if conditions else "1=1"
-    params_count = params[:]
-    params.extend([page_size, offset])
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute(
-                """SELECT h.id, h.user_id, h.created_at, h.panel_type, h.original_urls, h.params, h.result_urls,
-                          h.long_image_url, u.username
-                   FROM image_history h
-                   LEFT JOIN users u ON h.user_id = u.id
-                   WHERE """ + where_clause + """ ORDER BY h.created_at DESC LIMIT %s OFFSET %s""",
-                params,
-            )
-            rows = cur.fetchall()
-            cur.execute("SELECT COUNT(*) as total FROM image_history h WHERE " + where_clause, params_count)
-            total = cur.fetchone()['total']
-        conn.close()
-        items = []
-        for r in rows:
-            items.append({
-                'id': r['id'],
-                'user_id': r['user_id'],
-                'username': r.get('username') or '-',
-                'created_at': r['created_at'].strftime('%Y-%m-%d %H:%M') if r['created_at'] else '',
-                'panel_type': r['panel_type'] or '',
-                'original_urls': _parse_json(r['original_urls'], []),
-                'params': _parse_json(r['params'], {}),
-                'result_urls': _parse_json(r['result_urls'], []),
-                'long_image_url': (r.get('long_image_url') or '').strip() or None,
-            })
-        return jsonify({'success': True, 'items': items, 'total': total, 'page': page, 'page_size': page_size})
-    except Exception as e:
-        return jsonify({'success': False, 'error': str(e)})

app/blueprints/auth.py

@@ -1,107 +1,56 @@
 """
-认证蓝图：登录、登出、登录状态校验
+认证蓝图：登录页（仅 GET 渲染模板）+ 登出（清 JWT cookie）+ token 同步
+登录表单提交已经直连 Java 后端 /login，Python 这边只负责：
+1. 渲染登录页模板
+2. 把 Java 签发的 JWT 从前端写到 Python 同源 cookie，供后续页面跳转携带
+3. 登出：清 cookie 跳回登录页
 """
-from flask import Blueprint, request, redirect, url_for, session, jsonify
+import sys
-from werkzeug.security import check_password_hash
 
-from app_common import (
+from flask import Blueprint, redirect, url_for, make_response, request, jsonify
-    get_db,
+
-    _render_html,
+from app_common import _render_html, current_user_id
-    _is_session_user_valid,
+from config import JAVA_API_BASE
-    login_required,
+from jwt_util import COOKIE_NAME, parse_token_with_reason
-    BASE_DIR,
-)
-from tool.devices import DeviceIDGenerator
 
 auth_bp = Blueprint('auth', __name__)
 
 
-@auth_bp.route('/login', methods=['GET', 'POST'])
+@auth_bp.route('/login', methods=['GET'])
 def login():
-    if session.get('user_id') and _is_session_user_valid():
+    if current_user_id():
         return redirect(url_for('main.home'))
-    if request.method == 'POST':
+    return _render_html('login.html', java_api_base=JAVA_API_BASE)
-        data = request.get_json() if request.is_json else request.form
+
-        username = (data.get('username') or '').strip()
+
-        password = data.get('password') or ''
+@auth_bp.route('/api/auth/sync', methods=['POST'])
-        if not username or not password:
+def api_auth_sync():
-            if request.is_json:
+    """前端拿到 Java 返回的 JWT 后调用，把 token 写进 Python 同源 cookie。"""
-                return jsonify({'success': False, 'error': '请输入用户名和密码'})
+    data = request.get_json(silent=True) or {}
-            return _render_html('login.html', error='请输入用户名和密码')
+    token = (data.get('token') or '').strip()
-        try:
+
-            conn = get_db()
+    if not token:
-            with conn.cursor() as cur:
+        return jsonify({'success': False, 'error': '缺少 token'}), 400
-                cur.execute(
+    payload, reason = parse_token_with_reason(token)
-                    "SELECT id, password_hash, machine, is_admin FROM users WHERE username = %s",
+    if not payload:
-                    (username,)
+        # 把根因打到服务端日志，并回传给前端，便于现场排查
+        print(f"[auth] /api/auth/sync 校验失败: {reason}", file=sys.stderr)
+        return jsonify({'success': False, 'error': f'token 无效: {reason}'}), 401
+    resp = make_response(jsonify({'success': True}))
+    resp.set_cookie(
+        COOKIE_NAME,
+        token,
+        max_age=7 * 24 * 3600,
+        path='/',
+        httponly=True,
+        samesite='Lax',
     )
-                row = cur.fetchone()
+    return resp
-            if row and check_password_hash(row['password_hash'], password):
-                current_machine = DeviceIDGenerator().get_device_id()
-                stored_machine = (row.get('machine') or '').strip()
-                if not stored_machine:
-                    with conn.cursor() as cur:
-                        cur.execute("UPDATE users SET machine = %s WHERE id = %s", (current_machine, row['id']))
-                    conn.commit()
-                    conn.close()
-                    session.permanent = True
-                    session['user_id'] = row['id']
-                    session['username'] = username
-                    if request.is_json:
-                        return jsonify({'success': True, 'redirect': url_for('main.home')})
-                    return redirect(url_for('main.home'))
-                print("验证设备",stored_machine)
-                print("当前设备",current_machine)
-                if stored_machine != current_machine and row.get("is_admin") != 1:
-                    conn.close()
-                    err_msg = '当前设备与首次登录设备不一致，请在原设备上登录'
-                    if request.is_json:
-                        return jsonify({'success': False, 'error': err_msg})
-                    return _render_html('login.html', error=err_msg)
-                conn.close()
-                session.permanent = True
-                session['user_id'] = row['id']
-                session['username'] = username
-                if request.is_json:
-                    return jsonify({'success': True, 'redirect': url_for('main.home')})
-                return redirect(url_for('main.home'))
-            conn.close()
-        except Exception as e:
-            if request.is_json:
-                return jsonify({'success': False, 'error': str(e)})
-            return _render_html('login.html', error='登录失败，请稍后重试')
-        if request.is_json:
-            return jsonify({'success': False, 'error': '用户名或密码错误'})
-        return _render_html('login.html', error='用户名或密码错误')
-    return _render_html('login.html')
-
-
-@auth_bp.route('/api/auth/check')
-@login_required
-def api_auth_check():
-    """校验登录状态，用于页面加载时判断是否已登录；同时校验机器码是否与首次登录设备一致"""
-    if not session.get('user_id'):
-        return jsonify({'logged_in': False})
-    try:
-        conn = get_db()
-        with conn.cursor() as cur:
-            cur.execute("SELECT machine, is_admin FROM users WHERE id = %s", (session['user_id'],))
-            row = cur.fetchone()
-        conn.close()
-        if not row:
-            return jsonify({'logged_in': False})
-        stored_machine = (row.get('machine') or '').strip()
-        if stored_machine:
-            current_machine = DeviceIDGenerator().get_device_id()
-            if stored_machine != current_machine and row.get("is_admin") != 1:
-                session.clear()
-                return jsonify({'logged_in': False, 'error': '当前设备与首次登录设备不一致'})
-    except Exception:
-        return jsonify({'logged_in': False})
-    return jsonify({'logged_in': True, 'redirect': url_for('main.home')})
 
 
 @auth_bp.route('/logout')
 def logout():
-    session.clear()
+    resp = make_response(redirect(url_for('auth.login')))
-    return redirect(url_for('auth.login'))
+    resp.delete_cookie(COOKIE_NAME, path='/')
+    return resp
+

app/blueprints/brand.py

@@ -13,9 +13,9 @@ 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 flask import Blueprint, request, jsonify, send_file, redirect, Response
 
-from app_common import get_db, login_required, BASE_DIR
+from app_common import get_db, login_required, current_user_id, BASE_DIR
 from config import bucket_path,JAVA_API_BASE
 from brand_spider.main import single_file_handle, TaskCancelledError
 
@@ -70,12 +70,12 @@ def _get_uid_from_request_headers():
 
 def _resolve_user_id():
     """
-    优先使用请求头 uid；没有请求头 uid 时回退到 session['user_id']。
+    优先使用请求头 uid；没有请求头 uid 时回退到 JWT 解析的当前用户。
     若二者同时存在但不一致，则视为无效请求并返回 None。
     """
     req_uid = _get_uid_from_request_headers()
     if req_uid is not None:
-        sess_uid = session.get('user_id')
+        sess_uid = current_user_id()
         if sess_uid is not None:
             try:
                 if int(sess_uid) != int(req_uid):
@@ -84,7 +84,7 @@ def _resolve_user_id():
                 if str(sess_uid) != str(req_uid):
                     return None
         return req_uid
-    return session.get('user_id')
+    return current_user_id()
 
 
 def _get_user_id_or_error():

app/blueprints/brand_备份.py

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

app/blueprints/image.py

@@ -13,10 +13,10 @@ import subprocess
 import requests
 from urllib.parse import urlparse, quote
 
-from flask import Blueprint, request, jsonify, Response, session, send_file
+from flask import Blueprint, request, jsonify, Response, send_file
 from PIL import Image
 
-from app_common import get_db, login_required, BASE_DIR
+from app_common import get_db, login_required, current_user_id, BASE_DIR
 from config import STITCH_WORKFLOW_ID,client_name
 
 image_bp = Blueprint('image', __name__)
@@ -146,11 +146,12 @@ def api_generate():
                     except (TypeError, ValueError):
                         hid = None
                 conn = get_db()
+                _uid = current_user_id()
                 with conn.cursor() as cur:
                     if hid is not None and hid > 0:
                         cur.execute(
                             "SELECT result_urls FROM image_history WHERE id=%s AND user_id=%s",
-                            (hid, session['user_id']),
+                            (hid, _uid),
                         )
                         row = cur.fetchone()
                         existing_urls = []
@@ -182,7 +183,7 @@ def api_generate():
                                 _json.dumps(_sanitize_params_for_history(params)),
                                 _json.dumps(merged_result_urls),
                                 hid,
-                                session['user_id'],
+                                _uid,
                             ),
                         )
                         if cur.rowcount > 0:
@@ -192,7 +193,7 @@ def api_generate():
                             """INSERT INTO image_history (user_id, panel_type, original_urls, params, result_urls, long_image_url)
                                VALUES (%s, %s, %s, %s, %s, %s)""",
                             (
-                                session['user_id'],
+                                _uid,
                                 params.get('panel_type', ''),
                                 _json.dumps(result.get('original_urls') or []),
                                 _json.dumps(_sanitize_params_for_history(params)),
@@ -366,7 +367,7 @@ def api_history():
         conn = get_db()
         with conn.cursor() as cur:
             where_user = "user_id = %s"
-            params_where = [session['user_id']]
+            params_where = [current_user_id()]
             if panel_type:
                 where_user += " AND panel_type = %s"
                 params_where.append(panel_type)

app/blueprints/main.py

@@ -7,14 +7,15 @@ from flask import Blueprint, send_file, render_template_string
 from app_common import (
     get_db,
     _render_html,
-    _is_session_user_valid,
     login_required,
     admin_required,
+    current_user_id,
+    current_username,
     STATIC_DIR,
     BASE_DIR,
     ASSETS_DIR
 )
-from flask import redirect, url_for, session
+from flask import redirect, url_for
 from flask import Flask, request, Response, stream_with_context
 import requests
 
@@ -67,7 +68,7 @@ def _resolve_asset_filename(filename):
 
 @main_bp.route('/')
 def index():
-    if session.get('user_id') and _is_session_user_valid():
+    if current_user_id():
         return redirect(url_for('main.home'))
     return redirect(url_for('auth.login'))
 
@@ -75,15 +76,16 @@ def index():
 @main_bp.route('/home')
 @login_required
 def home():
+    uid = current_user_id()
     try:
         conn = get_db()
         with conn.cursor() as cur:
-            cur.execute("SELECT username, is_admin FROM users WHERE id = %s", (session['user_id'],))
+            cur.execute("SELECT username, is_admin FROM users WHERE id = %s", (uid,))
             row = cur.fetchone()
         conn.close()
-        return _render_html('home.html', username=row.get('username', ''), is_admin=bool(row.get('is_admin')), user_id=session.get('user_id'),baseUrl=base_url,version=version)
+        return _render_html('home.html', username=row.get('username', ''), is_admin=bool(row.get('is_admin')), user_id=uid, baseUrl=base_url, version=version, java_api_base=JAVA_API_BASE)
     except Exception:
-        return _render_html('home.html', username=session.get('username', ''), is_admin=False, user_id=session.get('user_id'),baseUrl=base_url,version=version)
+        return _render_html('home.html', username=current_username(), is_admin=False, user_id=uid, baseUrl=base_url, version=version, java_api_base=JAVA_API_BASE)
 
 
 @main_bp.route('/image')
@@ -112,10 +114,10 @@ def brand_page():
                 if raw[:7] == b'gAAAAAB':
                     raise
                 content = raw.decode('utf-8', errors='replace')
-            return render_template_string(content, user_id=session.get('user_id'))
+            return render_template_string(content, user_id=current_user_id(), java_api_base=JAVA_API_BASE)
         except Exception:
             continue
-    return _render_html('brand.html', user_id=session.get('user_id'))
+    return _render_html('brand.html', user_id=current_user_id(), java_api_base=JAVA_API_BASE)
 
 
 @main_bp.route('/brand/legacy')
@@ -132,7 +134,7 @@ def brand_page_legacy():
         1,
     )
     content = content.replace('height: calc(100vh - 56px);', 'height: 100vh;', 1)
-    return render_template_string(content, user_id=session.get('user_id'))
+    return render_template_string(content, user_id=current_user_id())
 
 
 
@@ -179,7 +181,7 @@ def serve_new_web_source(filename):
     if filename.lower().endswith('.html'):
         with open(file_abs, 'rb') as f:
             content = f.read().decode('utf-8', errors='replace')
-        raw_uid = session.get('user_id')
+        raw_uid = current_user_id()
         uid_value = str(int(raw_uid)) if str(raw_uid or '').isdigit() else '""'
         uid_script = (
             "\n<script>"

app/config.py

@@ -22,7 +22,7 @@ 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")
+JAVA_API_BASE = os.getenv("java_api_base", "http://api.aishufu.top:18080/").rstrip("/")
 
 # 紫鸟浏览器配置
 from urllib.parse import unquote
@@ -32,7 +32,7 @@ ZN_USERNAME = unquote(ZN_USERNAME)
 ZN_PASSWORD = os.getenv("zn_password", "#20zsg25")
 
 # 删除品牌API配置
-DELETE_BRAND_API_BASE = os.getenv("DELETE_BRAND_API_BASE", JAVA_API_BASE)
+DELETE_BRAND_API_BASE = os.getenv("DELETE_BRAND_API_BASE", JAVA_API_BASE).rstrip("/")
 
 
 cache_path = "./user_data"
@@ -50,7 +50,6 @@ file_url_pre = f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/"
 
 os.environ['OSS_ACCESS_KEY_ID'] = accessKeyId
 os.environ['OSS_ACCESS_KEY_SECRET'] = accessKeySecret
-os.environ['SECRET_KEY'] = "ddffc7c1d02121d9554d7b080b2511b6"
 
 
 debug = True

app/jwt_util.py

@@ -0,0 +1,119 @@
+"""
+JWT 工具：解析 Java 端签发的 token、从请求中提取 token。
+与 Java 端 JwtService 共用同一个 HS256 密钥（AIIMAGE_JWT_SECRET）。
+"""
+import os
+import sys
+import time
+from typing import Optional, Tuple
+
+from flask import request
+
+try:
+    import jwt as pyjwt
+    _PYJWT_IMPORT_ERROR = None
+except ImportError as _e:
+    pyjwt = None
+    _PYJWT_IMPORT_ERROR = _e
+    # 启动时立刻给出醒目提示，避免上线后才发现一直 401
+    print(
+        "[auth] WARNING: PyJWT 未安装，所有依赖 JWT 的接口都会返回 401。"
+        " 请执行 `pip install PyJWT==2.10.1` 或 `pip install -r requirements.txt`。",
+        file=sys.stderr,
+    )
+
+
+_DEFAULT_SECRET = "please-change-this-secret-please-rotate-at-least-32-bytes"
+COOKIE_NAME = os.getenv("AIIMAGE_AUTH_COOKIE_NAME", "aiimage_token")
+
+# 桌面端与 Java 服务器时钟可能漂移（笔记本休眠、用户手动改时间、跨时区等），
+# 给 JWT exp/nbf 校验留出容差，避免 /api/auth/sync 因为时间不同步而 401。
+# 默认 5 分钟，可通过 AIIMAGE_JWT_LEEWAY_SECONDS 覆盖。
+try:
+    _JWT_LEEWAY_SECONDS = int(os.getenv("AIIMAGE_JWT_LEEWAY_SECONDS", "300"))
+except ValueError:
+    _JWT_LEEWAY_SECONDS = 300
+
+
+def _signing_key() -> bytes:
+    """与 Java JwtService.signingKey 保持一致：UTF-8 字节，不足 32 字节右侧补 0。"""
+    secret = os.getenv("AIIMAGE_JWT_SECRET", _DEFAULT_SECRET)
+    key_bytes = secret.encode("utf-8")
+    if len(key_bytes) < 32:
+        key_bytes = key_bytes + b"\x00" * (32 - len(key_bytes))
+    return key_bytes
+
+
+def _describe_clock_skew(token: str) -> str:
+    """解析 token 里的 iat/exp，与本地时钟比较，返回 ' iat=.. exp=.. now=.. skew=..s' 字符串。
+    仅供日志使用，不做安全决策；解析失败返回空串，不影响主流程。"""
+    if pyjwt is None:
+        return ""
+    try:
+        unverified = pyjwt.decode(token, options={"verify_signature": False, "verify_exp": False, "verify_iat": False, "verify_nbf": False})
+    except Exception:
+        return ""
+    iat = unverified.get("iat")
+    exp = unverified.get("exp")
+    now = int(time.time())
+    parts = [f"now={now}"]
+    if isinstance(iat, (int, float)):
+        parts.append(f"iat={int(iat)} skew_iat={now - int(iat)}s")
+    if isinstance(exp, (int, float)):
+        parts.append(f"exp={int(exp)} skew_exp={now - int(exp)}s")
+    return ", " + " ".join(parts)
+
+
+def parse_token_with_reason(token: str) -> Tuple[Optional[dict], Optional[str]]:
+    """解析 JWT，返回 (payload, error_reason)。
+    payload 命中时 error_reason 为 None；失败时 payload 为 None，error_reason 描述根因。"""
+    if not token:
+        return None, "token 为空"
+    if pyjwt is None:
+        return None, f"PyJWT 未安装（{_PYJWT_IMPORT_ERROR}）"
+    # 打印 token 头，便于发现 alg 不是 HS256 等情况（不验签，仅 base64 解码 header）
+    try:
+        header = pyjwt.get_unverified_header(token)
+    except Exception as e:
+        return None, f"无法解析 token header: {type(e).__name__}: {e}"
+    alg = header.get("alg") or "?"
+    try:
+        payload = pyjwt.decode(
+            token,
+            _signing_key(),
+            algorithms=["HS256", "HS384", "HS512"],
+            leeway=_JWT_LEEWAY_SECONDS,
+        )
+    except Exception as e:
+        # 常见：ExpiredSignatureError / InvalidSignatureError / DecodeError / InvalidAlgorithmError
+        # 把 token 内 exp/iat 与本地时钟一起打出来，定位"时间漂移"类失败更直接
+        skew_info = _describe_clock_skew(token)
+        return None, f"{type(e).__name__}: {e} (token alg={alg}{skew_info})"
+    sub = payload.get("sub")
+    try:
+        user_id = int(sub) if sub is not None else None
+    except (TypeError, ValueError):
+        return None, f"sub 字段非法: {sub!r}"
+    if user_id is None:
+        return None, "payload 缺少 sub 字段"
+    return {
+        "user_id": user_id,
+        "username": payload.get("username") or "",
+        "device_id": payload.get("deviceId") or "",
+    }, None
+
+
+def parse_token(token: str) -> Optional[dict]:
+    """解析 JWT，返回 {user_id, username, device_id}；失败返回 None。"""
+    payload, _ = parse_token_with_reason(token)
+    return payload
+
+
+def get_token_from_request() -> str:
+    """优先从 Authorization: Bearer 取，其次从 cookie 取。"""
+    auth = request.headers.get("Authorization", "")
+    if auth.startswith("Bearer "):
+        token = auth[len("Bearer "):].strip()
+        if token:
+            return token
+    return (request.cookies.get(COOKIE_NAME) or "").strip()

app/main.py

@@ -4,6 +4,8 @@ import json
 import sys
 import shutil
 import os
+import uuid
+import ctypes
 os.makedirs(cache_path,exist_ok=True)
 if not debug:
     today = datetime.datetime.now().strftime("%Y_%m_%d")
@@ -20,6 +22,7 @@ import subprocess
 from amazon.del_brand import kill_process
 from app import run_app
 from generate_api import generate
+from tool.devices import DeviceIDGenerator
 
 
 with open("version.txt","w",encoding="utf-8") as file:
@@ -185,17 +188,7 @@ class WindowAPI:
         if not result:
             return {'success': False, 'error': '用户取消'}
         path = result[0] if isinstance(result, (list, tuple)) else result
-        try:
+        return self._download_url_to_path(url, path)
-            import requests
-            resp = requests.get(url.strip(), timeout=120, stream=True)
-            resp.raise_for_status()
-            with open(path, 'wb') as f:
-                for chunk in resp.iter_content(chunk_size=65536):
-                    if chunk:
-                        f.write(chunk)
-            return {'success': True, 'path': path}
-        except Exception as e:
-            return {'success': False, 'error': str(e)}
 
     def save_file_from_url_new(self, url, default_filename='download.bin'):
         """弹窗选择保存位置，从 url 下载文件并保存。根据文件后缀动态设置保存类型。"""
@@ -229,18 +222,124 @@ class WindowAPI:
             return {'success': False, 'error': '用户取消'}
 
         path = result[0] if isinstance(result, (list, tuple)) else result
+        return self._download_url_to_path(url, path)
+
+    def save_file_from_url_with_progress(self, url, default_filename='download.bin', download_id=''):
+        """弹窗选择保存位置，从 url 下载文件并通过前端事件上报下载进度。"""
+        print("调用到save_file_from_url_with_progress方法")
+        if not url or not url.strip():
+            return {'success': False, 'error': '下载地址为空'}
+
+        filename = str(default_filename or 'download.bin').strip() or 'download.bin'
+        ext = os.path.splitext(filename)[1].lower()
+        if ext == '.zip':
+            file_types = ('ZIP 压缩包 (*.zip)', '所有文件 (*.*)')
+        elif ext in ('.xlsx', '.xls'):
+            file_types = ('Excel 文件 (*.xlsx;*.xls)', '所有文件 (*.*)')
+        elif ext == '.csv':
+            file_types = ('CSV 文件 (*.csv)', '所有文件 (*.*)')
+        elif ext == '.txt':
+            file_types = ('文本文件 (*.txt)', '所有文件 (*.*)')
+        elif ext == '.json':
+            file_types = ('JSON 文件 (*.json)', '所有文件 (*.*)')
+        else:
+            file_types = ('所有文件 (*.*)',)
+
+        result = self._window.create_file_dialog(
+            webview.SAVE_DIALOG,
+            save_filename=filename,
+            file_types=file_types
+        )
+        if not result:
+            return {'success': False, 'error': '用户取消'}
+
+        path = result[0] if isinstance(result, (list, tuple)) else result
+        progress_id = str(download_id or uuid.uuid4())
+        return self._download_url_to_path(url, path, progress_id)
+
+    def _download_url_to_path(self, url, path, download_id=''):
+        """先下载到临时文件，完整下载后再替换为最终文件，避免用户打开半成品。"""
+        final_path = os.path.abspath(path)
+        final_dir = os.path.dirname(final_path) or os.getcwd()
+        final_name = os.path.basename(final_path)
+        temp_path = os.path.join(final_dir, f".{final_name}.crdownload")
+        last_progress_emit = 0.0
+        last_progress_percent = -1
         try:
-            import requests
+            os.makedirs(final_dir, exist_ok=True)
+            if os.path.exists(temp_path):
+                os.remove(temp_path)
             resp = requests.get(url.strip(), timeout=120, stream=True)
             resp.raise_for_status()
-            with open(path, 'wb') as f:
+            total = int(resp.headers.get('Content-Length') or 0)
+            downloaded = 0
+            self._emit_download_progress(download_id, 'running', final_path, downloaded, total)
+            with open(temp_path, 'wb') as f:
+                self._hide_file(temp_path)
                 for chunk in resp.iter_content(chunk_size=65536):
                     if chunk:
                         f.write(chunk)
-            return {'success': True, 'path': path}
+                        downloaded += len(chunk)
+                        percent = int(downloaded * 100 / total) if total else 0
+                        now_time = time.time()
+                        if percent != last_progress_percent or now_time - last_progress_emit >= 0.2:
+                            self._emit_download_progress(download_id, 'running', final_path, downloaded, total)
+                            last_progress_emit = now_time
+                            last_progress_percent = percent
+                f.flush()
+                os.fsync(f.fileno())
+            self._show_file(temp_path)
+            os.replace(temp_path, final_path)
+            self._emit_download_progress(download_id, 'success', final_path, downloaded, total)
+            return {'success': True, 'path': final_path}
         except Exception as e:
+            try:
+                if os.path.exists(temp_path):
+                    os.remove(temp_path)
+            except Exception:
+                pass
+            self._emit_download_progress(download_id, 'failed', final_path, 0, 0, str(e))
             return {'success': False, 'error': str(e)}
 
+    def _emit_download_progress(self, download_id, status, path, downloaded, total, error=''):
+        if not download_id:
+            return
+        payload = {
+            'id': download_id,
+            'status': status,
+            'path': path,
+            'downloaded': downloaded,
+            'total': total,
+            'percent': round(downloaded * 100 / total, 1) if total else 0,
+            'error': error,
+        }
+        script = (
+            "window.dispatchEvent(new CustomEvent('pywebview-download-progress', "
+            f"{{ detail: {json.dumps(payload, ensure_ascii=False)} }}));"
+        )
+        try:
+            self._window.evaluate_js(script)
+        except Exception:
+            pass
+
+    def _hide_file(self, path):
+        """Windows 下隐藏临时下载文件，减少用户误打开半成品的机会。"""
+        if os.name != 'nt':
+            return
+        try:
+            ctypes.windll.kernel32.SetFileAttributesW(path, 0x02)
+        except Exception:
+            pass
+
+    def _show_file(self, path):
+        """恢复普通文件属性，避免最终保存文件被隐藏。"""
+        if os.name != 'nt':
+            return
+        try:
+            ctypes.windll.kernel32.SetFileAttributesW(path, 0x80)
+        except Exception:
+            pass
+
     def save_template_xlsx(self):
         """弹窗选择保存位置，将品牌文档格式模板 xlsx 保存到用户选择的位置。"""
         template_name = '品牌文档格式_模板.xlsx'
@@ -347,6 +446,13 @@ class WindowAPI:
         except Exception as e:
             return {'success': False, 'error': str(e)}
 
+    def get_device_id(self):
+        """暴露给 HTML 的设备指纹接口：复用桌面端硬件特征生成 64 位 SHA256 ID。"""
+        try:
+            return {'success': True, 'device_id': DeviceIDGenerator().get_device_id()}
+        except Exception as e:
+            return {'success': False, 'error': str(e)}
+
     def get_detail_del(self,task_id):
         """获取正在执行的删除品牌任务详情"""
         task_info = runing_task.get(task_id)
@@ -432,7 +538,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.open_external_url,api.enqueue_json,api.get_detail_del,api.save_file_from_url_new)
+                  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,api.save_file_from_url_new,api.save_file_from_url_with_progress,api.get_device_id)
     webview.start(
         debug=True,
         storage_path=cache_path,

app/new_web_source/convert.html

@@ -0,0 +1,16 @@
+<!doctype html>
+<html lang="zh-CN">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>格式转换 - 数富AI</title>
+    <script type="module" crossorigin src="/assets/convert.js"></script>
+    <link rel="modulepreload" crossorigin href="/assets/java-modules-B8c-YG5x.js">
+    <link rel="modulepreload" crossorigin href="/assets/brand-COze15GJ.js">
+    <link rel="stylesheet" crossorigin href="/assets/java-modules-l5anrOZ2.css">
+    <link rel="stylesheet" crossorigin href="/assets/convert-6oxZNMye.css">
+  </head>
+  <body>
+    <div id="app"></div>
+  </body>
+</html>

app/new_web_source/dedupe.html

@@ -0,0 +1,16 @@
+<!doctype html>
+<html lang="zh-CN">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>数据去重 - 数富AI</title>
+    <script type="module" crossorigin src="/assets/dedupe.js"></script>
+    <link rel="modulepreload" crossorigin href="/assets/java-modules-B8c-YG5x.js">
+    <link rel="modulepreload" crossorigin href="/assets/brand-COze15GJ.js">
+    <link rel="stylesheet" crossorigin href="/assets/java-modules-l5anrOZ2.css">
+    <link rel="stylesheet" crossorigin href="/assets/dedupe-DpfPYQDv.css">
+  </head>
+  <body>
+    <div id="app"></div>
+  </body>
+</html>

app/new_web_source/delete-brand.html

@@ -0,0 +1,17 @@
+<!doctype html>
+<html lang="zh-CN">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>删除品牌 - 数富AI</title>
+    <script type="module" crossorigin src="/assets/delete-brand.js"></script>
+    <link rel="modulepreload" crossorigin href="/assets/java-modules-B8c-YG5x.js">
+    <link rel="modulepreload" crossorigin href="/assets/brand-COze15GJ.js">
+    <link rel="modulepreload" crossorigin href="/assets/categorized-timers-JPA-olTr.js">
+    <link rel="stylesheet" crossorigin href="/assets/java-modules-l5anrOZ2.css">
+    <link rel="stylesheet" crossorigin href="/assets/delete-brand-DZ3x5hOX.css">
+  </head>
+  <body>
+    <div id="app"></div>
+  </body>
+</html>

app/new_web_source/split.html

@@ -0,0 +1,16 @@
+<!doctype html>
+<html lang="zh-CN">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>数据拆分 - 数富AI</title>
+    <script type="module" crossorigin src="/assets/split.js"></script>
+    <link rel="modulepreload" crossorigin href="/assets/java-modules-B8c-YG5x.js">
+    <link rel="modulepreload" crossorigin href="/assets/brand-COze15GJ.js">
+    <link rel="stylesheet" crossorigin href="/assets/java-modules-l5anrOZ2.css">
+    <link rel="stylesheet" crossorigin href="/assets/split-CR1PtDQE.css">
+  </head>
+  <body>
+    <div id="app"></div>
+  </body>
+</html>

app/requirements.txt

@@ -54,6 +54,7 @@ pyinstaller==6.19.0
 pyinstaller-hooks-contrib==2026.3
 PyMsgBox==2.0.1
 PyMySQL==1.1.2
+PyJWT==2.10.1
 pyperclip==1.11.0
 PyQt5==5.15.11
 PyQt5-Qt5==5.15.2