后台管理接口修复 后端BUG修复
This commit is contained in:
super
@@ -594,9 +594,168 @@ class SimilarAsinTask(TaskBase):
|
||||
runing_task[task_id]["status"] = "failed"
|
||||
runing_task[task_id]["error"] = str(e)
|
||||
|
||||
def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="",
|
||||
item_data:dict={},
|
||||
is_done: bool = False):
|
||||
def process_task(self, task_data: dict):
|
||||
"""Process similar ASIN task rows one by one."""
|
||||
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)
|
||||
|
||||
if not task_id:
|
||||
self.log("similar asin task_id is empty, skip", "WARNING")
|
||||
return
|
||||
|
||||
self.log(f"similar asin task {task_id} start, groups={len(groups)}")
|
||||
|
||||
if not groups:
|
||||
self.log("similar asin groups/rows is empty, skip", "WARNING")
|
||||
return
|
||||
|
||||
runing_task[task_id] = {
|
||||
"status": "running",
|
||||
"start_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"total_shops": 1,
|
||||
"processed_shops": 0,
|
||||
"total_countries": len(groups),
|
||||
"processed_countries": 0,
|
||||
"total_asins": 0,
|
||||
"processed_asins": 0,
|
||||
"success_count": 0,
|
||||
"failed_count": 0,
|
||||
"stop_requested": False
|
||||
}
|
||||
|
||||
if task_id in runing_task and runing_task[task_id].get("stop_requested", False):
|
||||
self.log(f"similar asin task {task_id} stop requested before processing", "WARNING")
|
||||
runing_task[task_id]["status"] = "stopped"
|
||||
return
|
||||
|
||||
max_retry = 3
|
||||
"""
|
||||
show_notification("开始抓取数据", "info")
|
||||
"""
|
||||
show_notification("Start fetching similar ASIN data", "info")
|
||||
chrome = ChromeAmzone()
|
||||
try:
|
||||
result = []
|
||||
last_asin = ""
|
||||
for gp_index, gp in enumerate(groups):
|
||||
items = gp.get("items", [])
|
||||
group_item = []
|
||||
for value in items:
|
||||
asin = value.get("asin")
|
||||
country = value.get("country")
|
||||
last_asin = asin or last_asin
|
||||
return_data = {
|
||||
"image_url": "",
|
||||
"title": "",
|
||||
"similar_data": []
|
||||
}
|
||||
for _ in range(max_retry):
|
||||
try:
|
||||
return_data = chrome.run(country, asin) or {}
|
||||
self.log(f"similar asin crawl result -> {return_data}")
|
||||
break
|
||||
except Exception as e:
|
||||
if "涓庨〉闈㈢殑杩炴帴宸叉柇寮€" in str(e):
|
||||
chrome = ChromeAmzone()
|
||||
if not isinstance(return_data, dict):
|
||||
return_data = {}
|
||||
similar_data = return_data.get("similar_data")
|
||||
if not isinstance(similar_data, list):
|
||||
similar_data = []
|
||||
group_item.append({
|
||||
"sourceFileKey": value.get("sourceFileKey"),
|
||||
"sourceFilename": value.get("sourceFilename"),
|
||||
"rowToken": value.get("rowToken"),
|
||||
"groupKey": value.get("groupKey"),
|
||||
"id": value.get("id") or value.get("displayId"),
|
||||
"asin": value.get("asin"),
|
||||
"country": value.get("country"),
|
||||
"sku": value.get("sku"),
|
||||
"url": return_data.get("image_url"),
|
||||
"title": return_data.get("title"),
|
||||
"done": False,
|
||||
"urls": [
|
||||
item.get("ori_picture")
|
||||
for item in similar_data
|
||||
if isinstance(item, dict) and item.get("ori_picture")
|
||||
]
|
||||
})
|
||||
|
||||
result.append({
|
||||
"sourceFileKey": gp.get("sourceFileKey"),
|
||||
"sourceFilename": gp.get("sourceFilename"),
|
||||
"groupKey": gp.get("groupKey"),
|
||||
"baseId": gp.get("baseId"),
|
||||
"displayId": gp.get("displayId"),
|
||||
"items": group_item
|
||||
})
|
||||
|
||||
is_done = gp_index == len(groups) - 1
|
||||
if len(result) > 20 or is_done:
|
||||
self.post_result(
|
||||
task_id=task_id,
|
||||
chunkIndex=gp_index + 1,
|
||||
chunkTotal=len(groups),
|
||||
asin=last_asin,
|
||||
item_data=result,
|
||||
is_done=is_done
|
||||
)
|
||||
result = []
|
||||
|
||||
if result:
|
||||
self.post_result(
|
||||
task_id=task_id,
|
||||
chunkIndex=len(groups),
|
||||
chunkTotal=len(groups),
|
||||
asin=last_asin,
|
||||
item_data=result,
|
||||
is_done=True
|
||||
)
|
||||
|
||||
try:
|
||||
chrome.close()
|
||||
except Exception as e:
|
||||
print("close browser failed", e)
|
||||
if task_id in runing_task:
|
||||
runing_task[task_id]["processed_shops"] += 1
|
||||
except Exception as e:
|
||||
self.log(f"similar asin task {task_id} failed: {str(e)}", "ERROR")
|
||||
self.log(traceback.format_exc(), "ERROR")
|
||||
|
||||
if task_id in runing_task:
|
||||
if runing_task[task_id].get("stop_requested", False):
|
||||
runing_task[task_id]["status"] = "stopped"
|
||||
self.log(f"similar asin task {task_id} stopped")
|
||||
else:
|
||||
runing_task[task_id]["status"] = "completed"
|
||||
self.log(f"similar asin task {task_id} completed")
|
||||
|
||||
except Exception as e:
|
||||
self.log(f"similar asin task failed: {traceback.format_exc()}", "ERROR")
|
||||
if task_id and task_id in runing_task:
|
||||
runing_task[task_id]["status"] = "failed"
|
||||
runing_task[task_id]["error"] = str(e)
|
||||
|
||||
def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="",
|
||||
item_data:dict={},
|
||||
is_done: bool = False):
|
||||
"""回传处理结果到API
|
||||
"""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user