修复BUG，完善匹配店铺

2026-04-11 00:57:34 +08:00
parent 9a59dac6d0
commit b400b494ad
21 changed files with 1057 additions and 1663 deletions
@@ -226,6 +226,9 @@ class MatchTak:
            items = data.get("items", [])
            country_codes = data.get("country_codes", [])
            risk_listing_filter = data.get("risk_listing_filter", "All")
+            user_id = data.get("user_id")
+            stage_index = data.get("stage_index")
+            final_stage = bool(data.get("final_stage", True))

            if not task_id:
                self.log("任务ID为空，跳过", "WARNING")
@@ -269,7 +272,7 @@ class MatchTak:
                show_notification(f"开始处理店铺: {shop_name}", "info")
                
                try:
-                    self.process_shop(shop_item, country_codes, task_id,risk_listing_filter)
+                    self.process_shop(shop_item, country_codes, task_id, risk_listing_filter, user_id, stage_index, final_stage)
                    # 更新已处理店铺数
                    if task_id in runing_task:
                        runing_task[task_id]["processed_shops"] += 1
@@ -296,7 +299,7 @@ class MatchTak:
                    runing_task[task_id]["status"] = "failed"
                    runing_task[task_id]["error"] = str(e)
    
-    def process_shop(self, shop_item: dict, country_codes: list, task_id: int, risk_listing_filter: str):
+    def process_shop(self, shop_item: dict, country_codes: list, task_id: int, risk_listing_filter: str, user_id=None, stage_index=None, final_stage: bool = True):
        """处理单个店铺
        
        Args:
@@ -406,9 +409,11 @@ class MatchTak:
                    import traceback
                    self.log(f"处理国家 {country_code} 失败: {str(e)}", "ERROR")
                    self.log(traceback.format_exc(), "ERROR")
-            # 最后回传，标记完成
            try:
-                self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
+                if final_stage:
+                    self.post_result(task_id, shop_name, country_code, "", "", is_done=True)
+                else:
+                    self.post_stage_finished(task_id, user_id, stage_index)
            except Exception as e:
                self.log(f"回传结果失败: {str(e)}", "ERROR")
        finally:
@@ -560,6 +565,44 @@ class MatchTak:
        
        self.log(f"国家 {country_name} 处理完成")
    
+    def post_stage_finished(self, task_id: int, user_id, stage_index):
+        import requests
+        from config import DELETE_BRAND_API_BASE
+
+        if user_id in (None, "", 0):
+            raise ValueError("user_id is required for stage completion callback")
+        if stage_index is None:
+            raise ValueError("stage_index is required for stage completion callback")
+
+        url = f"{DELETE_BRAND_API_BASE}/api/shop-match/tasks/{task_id}/stage-finished"
+        payload = {"stage_index": stage_index}
+        params = {"user_id": user_id}
+
+        max_retries = 3
+        for retry in range(max_retries):
+            try:
+                self.log(f"Attempting stage completion callback ({retry + 1}/{max_retries})")
+                response = requests.post(
+                    url,
+                    params=params,
+                    json=payload,
+                    headers={"Content-Type": "application/json"},
+                    timeout=30,
+                    verify=False,
+                )
+                self.log(f"Stage completion callback response: {response.text}")
+                data = response.json() if response.text else {}
+                if response.status_code == 200 and isinstance(data, dict) and data.get("success"):
+                    self.log(f"Stage completion callback succeeded: task={task_id}, stage={stage_index}")
+                    return
+                self.log(f"Stage completion callback failed, status={response.status_code}", "WARNING")
+            except Exception as e:
+                self.log(f"Stage completion callback exception: {str(e)}", "ERROR")
+            if retry < max_retries - 1:
+                time.sleep(2)
+
+        raise RuntimeError(f"Stage completion callback failed after retries: task={task_id}, stage={stage_index}")
+
    def post_result(self, task_id: int, shop_name: str, country_code: str, asin: str, status: str,is_done: bool = False):
        """回传处理结果到API
        
@@ -596,8 +639,8 @@ class MatchTak:
        max_retries = 3
        for retry in range(max_retries):
            try:
-                print("================【匹配】=====================")
-                self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)")
+                print("================回传处理结果====================")
+                self.log(f"尝试回传结果 (第{retry + 1}/{max_retries} 次)")
                self.log(f"回传URL: {url}")
                self.log(f"回传数据: {payload}")
                response = requests.post(
@@ -608,12 +651,11 @@ class MatchTak:
                    verify=False
                )
                self.log(f"回传结果: {response.text}")
-                
-                if response.status_code == 200:
-                    self.log(f"结果回传成功: {asin} - {status}")
+                data = response.json() if response.text else {}
+                if response.status_code == 200 and isinstance(data, dict) and data.get("success"):
+                    self.log(f"回传结果成功: {asin} - {status}")
                    return
-                else:
-                    self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING")
+                self.log(f"回传结果失败，状态码: {response.status_code}", "WARNING")
                print("=====================================")
                    
            except Exception as e:
@@ -624,9 +666,7 @@ class MatchTak:
            if retry < max_retries - 1:
                time.sleep(2)
        
-        self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR")
-        
-
+        raise RuntimeError("已达到最大重试次数，结果回传最终失败")

 if __name__ == "__main__":
    user_info = {