365 lines
13 KiB
Python
365 lines
13 KiB
Python
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()}")
|
||
|
||
|
||
|