diff --git a/app/amazon/__pycache__/del_brand.cpython-39.pyc b/app/amazon/__pycache__/del_brand.cpython-39.pyc index 656d682..950b0df 100644 Binary files a/app/amazon/__pycache__/del_brand.cpython-39.pyc and b/app/amazon/__pycache__/del_brand.cpython-39.pyc differ diff --git a/app/amazon/__pycache__/main.cpython-39.pyc b/app/amazon/__pycache__/main.cpython-39.pyc index d1ed31b..dcbaa46 100644 Binary files a/app/amazon/__pycache__/main.cpython-39.pyc and b/app/amazon/__pycache__/main.cpython-39.pyc differ diff --git a/app/amazon/main.py b/app/amazon/main.py index fd8472c..a238b1a 100644 --- a/app/amazon/main.py +++ b/app/amazon/main.py @@ -215,6 +215,7 @@ class TaskMonitor: try: # 处理每个国家 + chunk_index =1 for country_data in countries: # 检查是否收到暂停请求 if task_id in runing_task and runing_task[task_id].get("stop_requested", False): @@ -222,7 +223,7 @@ class TaskMonitor: break # 跳出循环，进入finally关闭店铺 try: - self.process_country(driver, country_data, task_id, result_id, shop_data) + chunk_index = self.process_country(driver, country_data, task_id, result_id, shop_data,chunk_index) except Exception as e: country_name = country_data.get("country", "未知") self.log(f"处理国家 {country_name} 失败: {str(e)}", "ERROR") @@ -244,7 +245,7 @@ class TaskMonitor: self.log(f"店铺 {shop_name} 已从执行列表中移除") def process_country(self, driver: AmazoneDriver, country_data: Dict[str, Any], - task_id: int, result_id: int, shop_data: Dict[str, Any]): + task_id: int, result_id: int, shop_data: Dict[str, Any],chunk_index:int): """处理单个国家的所有ASIN Args: @@ -294,8 +295,8 @@ class TaskMonitor: # 如果切换失败，回传该国家所有ASIN为失败状态 if not switch_success: self.log(f"切换到国家 {country} 失败，已重试 {max_retries} 次，将所有ASIN标记为失败", "ERROR") - self._report_all_asins_failed(country, items, task_id, shop_data) - return + chunk_index = self._report_all_asins_failed(country, items, task_id, shop_data,chunk_index) + return chunk_index # 切换到库存管理页面 try: @@ -304,8 +305,8 @@ class TaskMonitor: except Exception as e: self.log(f"切换页面失败: {str(e)}", "ERROR") # 切换页面失败也回传所有ASIN为失败 - self._report_all_asins_failed(country, items, task_id, shop_data) - return + chunk_index = self._report_all_asins_failed(country, items, task_id, shop_data,chunk_index) + return chunk_index # 处理每个ASIN file_key = shop_data.get("fileKey", "") @@ -321,15 +322,18 @@ class TaskMonitor: try: self.log(f"[{idx}/{len(items)}] 处理ASIN: {asin_item.get('asin', '')}") self.process_asin(driver, asin_item, country, task_id, result_id, - file_key, source_filename, total_rows) + file_key, source_filename, total_rows,chunk_index) except Exception as e: asin = asin_item.get("asin", "未知") self.log(f"处理ASIN {asin} 失败: {str(e)}", "ERROR") # 继续处理下一个ASIN - + chunk_index += 1 + return chunk_index + + def process_asin(self, driver: AmazoneDriver, asin_item: Dict[str, Any], country: str, task_id: int, result_id: int, file_key: str, - source_filename: str, total_rows: int): + source_filename: str, total_rows: int,chunk_index:int): """处理单个ASIN并回传结果 Args: @@ -445,9 +449,9 @@ class TaskMonitor: "files": [{ "fileKey": file_key, "sourceFilename": source_filename, - "chunkIndex": self.chunk_index, + "chunkIndex": chunk_index, "chunkTotal": total_rows, - "processedRows": self.chunk_index, + "processedRows": chunk_index, "totalRows": total_rows, "currentCountry": country, "currentAsin": asin, @@ -463,13 +467,12 @@ class TaskMonitor: self.post_result(task_id, payload) self.log(f"ASIN {asin} 结果已回传，状态: {status}") - self.chunk_index += 1 except Exception as e: self.log(f"回传结果失败: {str(e)}", "ERROR") def _report_all_asins_failed(self, country: str, items: List[Dict[str, Any]], - task_id: int, shop_data: Dict[str, Any]): + task_id: int, shop_data: Dict[str, Any],chunk_index:int): """将国家下所有ASIN标记为失败并回传 Args: @@ -499,9 +502,9 @@ class TaskMonitor: "files": [{ "fileKey": file_key, "sourceFilename": source_filename, - "chunkIndex": self.chunk_index, + "chunkIndex": chunk_index, "chunkTotal": total_rows, - "processedRows": self.chunk_index, + "processedRows": chunk_index, "totalRows": total_rows, "currentCountry": country, "currentAsin": asin, @@ -516,37 +519,57 @@ class TaskMonitor: } self.post_result(task_id, payload) - self.chunk_index += 1 + chunk_index += 1 self.log(f"ASIN {asin} 失败状态已回传") except Exception as e: self.log(f"回传ASIN {asin} 失败状态时出错: {str(e)}", "ERROR") - + + return chunk_index + def post_result(self, task_id: int, payload: Dict[str, Any]): - """回传结果到API + """回传结果到API（带重试机制） Args: task_id: 任务ID payload: 结果数据 """ url = f"{DELETE_BRAND_API_BASE}/api/delete-brand/tasks/{task_id}/result" + max_retries = 3 # 最多重试3次 - try: - response = requests.post( - url, - json=payload, - headers={"Content-Type": "application/json"}, - timeout=30, - verify=False # 忽略SSL证书验证 - ) - print("【结果提交】：",payload) - if response.status_code == 200: - self.log(f"结果回传成功: {url}") - else: - self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING") + for retry in range(max_retries): + try: + self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)") - except Exception as e: - self.log(f"调用API异常: {str(e)}", "ERROR") + response = requests.post( + url, + json=payload, + headers={"Content-Type": "application/json"}, + timeout=30, + verify=False # 忽略SSL证书验证 + ) + print("【结果提交】：",payload) + + if response.status_code == 200: + self.log(f"结果回传成功: {url}") + return # 成功后直接返回，不再重试 + else: + self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING") + # 如果还有重试机会，继续重试 + if retry < max_retries - 1: + self.log(f"准备重试... ({retry + 1}/{max_retries})") + time.sleep(2) # 等待2秒后重试 + else: + self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR") + + except Exception as e: + self.log(f"调用API异常: {str(e)}", "ERROR") + # 如果还有重试机会，继续重试 + if retry < max_retries - 1: + self.log(f"发生异常，准备重试... ({retry + 1}/{max_retries})") + time.sleep(2) # 等待2秒后重试 + else: + self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR") def update_task_status(self, task_id: int, **kwargs): """更新任务状态 diff --git a/app/main.py b/app/main.py index 955832a..6dbdfa9 100644 --- a/app/main.py +++ b/app/main.py @@ -268,8 +268,8 @@ class WindowAPI: payload = json.loads(data) if not isinstance(payload, (dict, list)): return {'success': False, 'error': '仅支持 JSON 对象或数组'} - if payload.get("type") == "delete-brand-run" and payload.get("data").get("taskId") in runing_task: - return {'success': False, 'error': '已存在正在执行的删除品牌任务'} + # if payload.get("type") == "delete-brand-run" and payload.get("data").get("taskId") in runing_task: + # return {'success': False, 'error': '已存在正在执行的删除品牌任务'} # 判断当前店铺是否正在运行中 if payload.get("data").get("items"): shop_name = payload["data"]["items"][0]["shopName"]