super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: app/amazon/__pycache__/del_brand.cpython-39.pyc

Browse Source 
modified:   app/amazon/__pycache__/main.cpython-39.pyc
	modified:   app/amazon/main.py
	modified:   app/main.py
This commit is contained in:
铭坤
2026-04-02 23:03:13 +08:00
parent f63ae84853
commit 274d294b2e
4 changed files with 58 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
app/amazon/__pycache__/del_brand.cpython-39.pyc
View File

Binary file not shown.

BIN
app/amazon/__pycache__/main.cpython-39.pyc
View File

Binary file not shown.

89
app/amazon/main.py
View File

@@ -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):
"""更新任务状态