From 79b177b34512b52fe8893724c56c3a60cc25c1de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=93=AD=E5=9D=A4?= <10372463+ming-kun@user.noreply.gitee.com> Date: Wed, 13 May 2026 16:51:29 +0800 Subject: [PATCH] modified: amazon/chrome_base.py modified: amazon/detail_spider.py modified: amazon/match_action.py modified: amazon/similar_asin.py deleted: amazon/user_data/chrome_data/BrowserMetrics-spare.pma deleted: amazon/user_data/chrome_data/BrowserMetrics/BrowserMetrics-69FAEF07-558.pma deleted: amazon/user_data/chrome_data/Crashpad/metadata deleted: amazon/user_data/chrome_data/Crashpad/settings.dat --- app/amazon/chrome_base.py | 8 +- app/amazon/detail_spider.py | 12 +- app/amazon/match_action.py | 30 +++- app/amazon/similar_asin.py | 268 ++++++------------------------------ 4 files changed, 77 insertions(+), 241 deletions(-) diff --git a/app/amazon/chrome_base.py b/app/amazon/chrome_base.py index b011245..e0d4241 100644 --- a/app/amazon/chrome_base.py +++ b/app/amazon/chrome_base.py @@ -116,20 +116,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() diff --git a/app/amazon/detail_spider.py b/app/amazon/detail_spider.py index 84fb739..0dd8669 100644 --- a/app/amazon/detail_spider.py +++ b/app/amazon/detail_spider.py @@ -273,18 +273,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 +349,7 @@ class SpiderTask(TaskBase): time.sleep(2) self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR") - raise RuntimeError("已达到最大重试次数，结果回传最终失败") + # raise RuntimeError("已达到最大重试次数，结果回传最终失败") if __name__ == '__main__': # spide = SpiderTask() diff --git a/app/amazon/match_action.py b/app/amazon/match_action.py index 2d8da3f..5304085 100644 --- a/app/amazon/match_action.py +++ b/app/amazon/match_action.py @@ -30,7 +30,7 @@ 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 @@ -86,7 +86,13 @@ class AmzoneMatchAction(AmamzonBase): # 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 @@ -228,6 +234,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 +261,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) @@ -265,7 +279,7 @@ class MatchTak(TaskBase): 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 +311,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 +354,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") @@ -353,10 +368,13 @@ class MatchTak(TaskBase): 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 }) diff --git a/app/amazon/similar_asin.py b/app/amazon/similar_asin.py index 4433ab5..26ea0c6 100644 --- a/app/amazon/similar_asin.py +++ b/app/amazon/similar_asin.py @@ -29,9 +29,9 @@ except ImportError: CONFIG_PROXY_MODE = 1 # Forbidden 后 3 分钟内统一使用代理：记录代理生效截止时间与当前代理 -_FORBIDDEN_PROXY_UNTIL = 0.0 -_FORBIDDEN_PROXY_DICT = None -_FORBIDDEN_PROXY_MINUTES = 0.5 +_FORBIDDEN_PROXY_UNTIL_Similar = 0.0 +_FORBIDDEN_PROXY_DICT_Similar = None +_FORBIDDEN_PROXY_MINUTES_Similar = 0.5 class ChromeAmzone(ChromeAmzoneBase): @@ -92,10 +92,10 @@ class ChromeAmzone(ChromeAmzoneBase): """ "fishkeeper Quick Aquarium Siphon Pump Gravel Cleaner - 256GPH Adjustable Powerful Fish Tank Vacuum Gravel Cleaning Kit for Aquarium Water Changer, Sand Cleaner, Dirt Removal : Amazon.co.uk: Pet Supplies" 标题 : - + """ - global _FORBIDDEN_PROXY_UNTIL, _FORBIDDEN_PROXY_DICT - + global _FORBIDDEN_PROXY_UNTIL_Similar, _FORBIDDEN_PROXY_DICT_Similar + headers = { "accept": "application/json, text/plain, */*", "accept-language": "zh-CN,zh;q=0.9,en;q=0.8", @@ -141,21 +141,21 @@ class ChromeAmzone(ChromeAmzoneBase): data = { "imageBase64" : imageBase64 } - + proxies = None - + # 若 3 分钟内曾出现 Forbidden，则直接使用当时保存的代理 now = time.time() - if now < _FORBIDDEN_PROXY_UNTIL and _FORBIDDEN_PROXY_DICT: - proxies = _FORBIDDEN_PROXY_DICT + if now < _FORBIDDEN_PROXY_UNTIL_Similar and _FORBIDDEN_PROXY_DICT_Similar: + 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) response.encoding = "utf-8" - + # 检查响应状态码和数据有效性 need_retry = False if response.status_code != 200: @@ -174,13 +174,13 @@ class ChromeAmzone(ChromeAmzoneBase): except Exception as e: print(f"解析响应JSON失败: {e}") need_retry = True - + # 如果需要重试且配置了代理URL if need_retry and CONFIG_PROXY_URL and not proxies: try: proxy_resp = requests.get(CONFIG_PROXY_URL, timeout=10) print("代理请求结果->：", proxy_resp.text) - + # 模式 2：账号密码代理，接口返回 JSON if CONFIG_PROXY_MODE == 2: resp_json = proxy_resp.json() @@ -206,17 +206,17 @@ class ChromeAmzone(ChromeAmzoneBase): "http": f"http://{proxy_ip}", "https": f"http://{proxy_ip}", } - + if proxies: print("使用代理重试请求") - response = requests_frp.post(url, headers=headers, params=params, json=data, + response = requests_frp.post(url, headers=headers, params=params, json=data, impersonate="chrome101", cookies=cookie, proxies=proxies) response.encoding = "utf-8" print("代理重试结果状态码:", response.status_code) - + # 记录 3 分钟内都使用该代理 - _FORBIDDEN_PROXY_UNTIL = now + _FORBIDDEN_PROXY_MINUTES * 60 - _FORBIDDEN_PROXY_DICT = proxies + _FORBIDDEN_PROXY_UNTIL_Similar = now + _FORBIDDEN_PROXY_MINUTES_Similar * 60 + _FORBIDDEN_PROXY_DICT_Similar = proxies except Exception as e: print("获取代理或重试失败:", e) @@ -240,6 +240,7 @@ class ChromeAmzone(ChromeAmzoneBase): 'image_url': "", 'title': "", 'category' : '', + 'sku' : '', 'success' : True } @@ -252,6 +253,11 @@ class ChromeAmzone(ChromeAmzoneBase): imge_ele = self.tab.ele('xpath://div[@id="imgTagWrapperId"]//img',timeout=20) image_url = imge_ele.attr("src") data["image_url"] = 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 @@ -290,7 +296,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,13 +305,13 @@ 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() # 判断是否采集标题出错 @@ -407,22 +413,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 +435,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" - ) + 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) 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: @@ -511,10 +488,9 @@ class SimilarAsinTask(TaskBase): } 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=3) or {} self.log(f"抓取结果->{return_data}") break except Exception as e: @@ -533,6 +509,7 @@ class SimilarAsinTask(TaskBase): "groupKey": i.get("groupKey"), "id": i.get("id"), "asin": i.get("asin"), + "sku" : return_data.get("sku"), "country": i.get("country"), "url": return_data.get("image_url"), "title": return_data.get("title"), @@ -556,7 +533,7 @@ class SimilarAsinTask(TaskBase): result.append(res) 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 +544,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 +552,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): @@ -594,168 +573,9 @@ class SimilarAsinTask(TaskBase): runing_task[task_id]["status"] = "failed" runing_task[task_id]["error"] = str(e) - 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): + def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="", + item_data:dict={}, + is_done: bool = False): """回传处理结果到API """ @@ -809,7 +629,7 @@ if __name__ == '__main__': spide = ChromeAmzone() resp = spide.run( - country="英国", asin="B0F79LCB3X", total_page=2 + country="德国", asin="B0CJ8SNXXV", total_page=2 ) print(resp) # task_data = {