import time import re import traceback from datetime import datetime from DrissionPage import Chromium, ChromiumOptions from amazon.del_brand import AmamzonBase, kill_process from amazon.tool import show_notification,get_shop_info from config import runing_task, runing_shop,base_dir class ChromeAmzone: mark_name = "亚马逊详情采集" country_info = { "英国": { "url": "https://www.amazon.co.uk/dp/B0CJ8SNXXV", "zip_code": "SW1A 1AA" }, "德国": { "url": "https://www.amazon.de/dp/B0CC8CW9G2?th=1", "zip_code": "10115" }, "法国": { "url": "https://www.amazon.fr/dp/B0FRG1MJ8H?th=1", "zip_code": "75001" }, "西班牙": { "url": "https://www.amazon.es/dp/B08ZXVNYNN", "zip_code": "28001" }, "意大利": { "url": "https://www.amazon.it/dp/B0D1P17T2Q", "zip_code": "20121" } } def __init__(self): """ 杀死当前谷歌浏览器进程，并使用 drissionpage 启动谷歌浏览器，使用系统安装的浏览器默认用户文件夹 """ # 杀死现有的Chrome进程 print("正在关闭现有的Chrome浏览器进程...") import os os.system('taskkill /f /t /im chrome.exe') time.sleep(2) print("正在启动Chrome浏览器...") # sellersprite_plug_path = os.path.join(base_dir,"app_resource","sellersprite-extension-mv3") sellersprite_plug_path = os.path.join("D:\\私单交付\\maixiang_AI\\","app_resource","sellersprite-extension-mv3") print(sellersprite_plug_path) # 配置浏览器选项 co = ChromiumOptions() # co.use_system_user_path(on_off=True) co.set_user_data_path(r"D:\私单交付\maixiang_AI\user_data\chrome_data") co.set_local_port(port=19890) co.add_extension(sellersprite_plug_path) # co.set_argument('--disable-features=DisableLoadExtensionCommandLineSwitch') # co.set_argument('--load-extension',sellersprite_plug_path) co.set_browser_path(r'D:\私单交付\maixiang_AI\app_resource\chrome-win\chrome.exe') # 使用系统默认的用户数据目录 # co.set_user_data_path(r'C:\Users\{}\AppData\Local\Google\Chrome\User Data'.format(os.getenv('USERNAME'))) self.browser = Chromium(co) self.tab = self.browser.latest_tab print("Chrome浏览器启动成功") def close_init_popup(self): """ 关闭所有的初始化弹窗 """ self.tab.wait.doc_loaded(timeout=60,raise_err=True) footbar = self.tab.eles('xpath://footer[@class="el-dialog__footer"]',timeout=5) if len(footbar) > 0: do_not_remind = footbar[0].eles('xpath:.//input[@class="el-checkbox__original"]') if len(do_not_remind) > 0: do_not_remind[0].check() resume_immediately = footbar[0].eles('xpath:.//button') if len(resume_immediately) > 0: resume_immediately[0].click() self.tab.wait.doc_loaded(timeout=60,raise_err=True) accept_btn = self.tab.eles('xpath://input[@id="sp-cc-accept"]',timeout=10) if len(accept_btn) > 0: accept_btn[0].click() def run(self, country, asin): """ 运行亚马逊详情采集任务 Args: country: 国家名称（如：英国、德国、法国、西班牙、意大利） asin: 亚马逊商品ASIN码 Returns: dict: 包含采集到的数据 """ try: # 验证国家是否支持 if country not in self.country_info: error_msg = f"不支持的国家: {country}，支持的国家有: {list(self.country_info.keys())}" print(error_msg) show_notification(error_msg, "error") return None # 获取国家配置 country_config = self.country_info[country] zip_code = country_config["zip_code"] # 1. 根据国家和ASIN拼接链接 base_url = country_config["url"] # 提取域名部分 domain = base_url.split("/dp/")[0] # 拼接新的URL product_url = f"{domain}/dp/{asin}" print(f"正在访问: {product_url}") # 打开链接 self.tab.get(product_url) time.sleep(3) # 等待页面初步加载 self.tab.wait.doc_loaded(timeout=30, raise_err=False) # 2. 切换国家/设置邮编 print(f"正在检查并设置邮编: {zip_code}") self._set_zip_code(zip_code) self.close_init_popup() # 3. 抓取数据 print("正在抓取商品数据...") data = self._scrape_data() # 添加基本信息 data['country'] = country data['asin'] = asin data['url'] = product_url data['timestamp'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"数据抓取完成: {data}") return data except Exception as e: error_msg = f"运行出错: {traceback.format_exc()}" print(error_msg) show_notification(f"采集失败: {str(e)}", "error") return None def _set_zip_code(self, zip_code): """ 设置邮编 Args: zip_code: 目标邮编 """ try: # 检查当前邮编 zip_display = self.tab.ele('xpath://div[@id="glow-ingress-block"]', timeout=10) if zip_display: current_text = zip_display.text print(f"当前地址信息: {current_text}") # 检查是否已经包含目标邮编 if zip_code in current_text: print(f"邮编已经设置为: {zip_code}，无需修改") return True # 需要设置邮编 print(f"正在设置邮编为: {zip_code}") # 点击地址选择按钮 location_link = self.tab.ele('xpath://a[@id="nav-global-location-popover-link"]', timeout=10) if not location_link: print("找不到地址设置按钮") return False location_link.click() time.sleep(1) # 等待邮编输入框出现 zip_input = self.tab.ele('xpath://input[@id="GLUXZipUpdateInput"]', timeout=10) if not zip_input: print("找不到邮编输入框") return False # 输入邮编 zip_input.input(zip_code, clear=True) time.sleep(0.5) # 点击提交按钮 submit_btn = self.tab.ele('xpath://input[@aria-labelledby="GLUXZipUpdate-announce"]', timeout=10) if not submit_btn: print("找不到提交按钮") return False submit_btn.click() # 等待提交按钮消失（表示请求已发送） print("等待邮编更新...") time.sleep(2) # 等待页面加载完成 self.tab.wait.doc_loaded(timeout=30, raise_err=False) time.sleep(2) # 验证邮编是否设置成功 zip_display_after = self.tab.ele('xpath://div[@id="glow-ingress-block"]', timeout=10) if zip_display_after: updated_text = zip_display_after.text print(f"更新后的地址信息: {updated_text}") if zip_code in updated_text: print(f"邮编设置成功: {zip_code}") return True else: print(f"邮编设置可能失败，当前显示: {updated_text}") return False return True except Exception as e: print(f"设置邮编时出错: {traceback.format_exc()}") return False def _scrape_data(self): """ 抓取商品数据 Returns: dict: 抓取到的数据 """ data = { 'top_sellers': [], # 前两名卖家信息 'cart_seller': None # 购物车所属卖家 } try: # 等待页面加载 time.sleep(3) # 1. 获取前两名卖家的价格和库存（来自 sellersprite 插件） print("正在抓取卖家排名数据...") surplus_tables = self.tab.eles( 'xpath://div[@id="sellersprite-extension-Inventory-surplus-count"]//div[@class="surplus-table"]', timeout=10 ) if surplus_tables: print(f"找到 {len(surplus_tables)} 个卖家数据") # 只获取前两个 for idx, table in enumerate(surplus_tables[:2]): try: # 提取价格和库存信息 # 注意：需要根据实际的HTML结构调整选择器 text_content = table.text print(f"第{idx+1}名卖家数据: {text_content}") seller_info = { 'rank': idx + 1, 'raw_data': text_content } # 尝试提取更结构化的数据 # 这里需要根据实际HTML结构来解析 # 可能需要查找子元素 try: # 示例：查找价格和库存的具体子节点 price_ele = table.ele('xpath:.//span[contains(@class,"price")]', timeout=2) stock_ele = table.ele('xpath:.//span[contains(@class,"stock")]', timeout=2) if price_ele: seller_info['price'] = price_ele.text.strip() if stock_ele: seller_info['stock'] = stock_ele.text.strip() except: # 如果找不到具体元素，就使用原始文本 pass data['top_sellers'].append(seller_info) except Exception as e: print(f"解析第{idx+1}名卖家数据失败: {str(e)}") else: print("未找到 sellersprite 插件数据，可能插件未启用") # 2. 获取购物车所属卖家 print("正在抓取购物车卖家信息...") cart_seller = self.tab.ele( 'xpath://div[@data-csa-c-content-id="desktop-merchant-info"]//a[@id="sellerProfileTriggerId"]', timeout=10 ) if cart_seller: seller_name = cart_seller.text.strip() print(f"购物车所属卖家: {seller_name}") data['cart_seller'] = seller_name else: print("未找到购物车卖家信息") # 尝试其他可能的选择器 try: alt_seller = self.tab.ele('xpath://a[@id="sellerProfileTriggerId"]', timeout=5) if alt_seller: data['cart_seller'] = alt_seller.text.strip() print(f"购物车所属卖家（备用方式）: {data['cart_seller']}") except: pass return data except Exception as e: print(f"抓取数据时出错: {traceback.format_exc()}") return data def close(self): """关闭浏览器""" try: if self.browser: self.browser.quit() print("浏览器已关闭") except Exception as e: print(f"关闭浏览器时出错: {str(e)}") class AmzonePriceMatch(AmamzonBase): mark_name = "跟价" pass if __name__ == '__main__': # 使用示例 print("=" * 50) print("亚马逊详情采集示例") print("=" * 50) try: # 创建ChromeAmzone实例 chrome = ChromeAmzone() # 示例1：采集英国站点的商品信息 print("\n示例1：采集英国站点商品") result1 = chrome.run(country="英国", asin="B0CJ8SNXXV") if result1: print(f"采集成功: {result1}") # 示例2：采集德国站点的商品信息 # print("\n示例2：采集德国站点商品") # result2 = chrome.run(country="德国", asin="B0CC8CW9G2") # if result2: # print(f"采集成功: {result2}") # # 关闭浏览器 # print("\n正在关闭浏览器...") # chrome.close() except Exception as e: print(f"运行出错: {traceback.format_exc()}")