531 lines
20 KiB
Python
531 lines
20 KiB
Python
import json
|
||
import sys
|
||
import os
|
||
import io
|
||
# sys.stdout.reconfigure(encoding='utf-8')
|
||
|
||
import time
|
||
import re
|
||
import traceback
|
||
from datetime import datetime
|
||
from DrissionPage import Chromium, ChromiumOptions
|
||
from collections import defaultdict
|
||
import requests
|
||
|
||
from amazon.del_brand import AmamzonBase, kill_process
|
||
from amazon.tool import show_notification,get_shop_info,remove_special_characters,split_currency_values
|
||
|
||
from config import runing_task, runing_shop,base_dir,DELETE_BRAND_API_BASE
|
||
|
||
|
||
class ChromeAmzone:
|
||
mark_name = "亚马逊详情采集"
|
||
country_info = {
|
||
"英国": {
|
||
"url": "https://www.amazon.co.uk/dp/B0CJ8SNXXV",
|
||
"zip_code": "SW1A 1AA",
|
||
"mark": "SW1A 1"
|
||
},
|
||
"德国": {
|
||
"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("正在关闭现有的chromium浏览器进程...")
|
||
os.system('taskkill /f /t /im chrome.exe')
|
||
time.sleep(2)
|
||
|
||
print("正在启动chromium浏览器...")
|
||
# 配置浏览器选项
|
||
co = ChromiumOptions()
|
||
user_data_path = os.path.join(base_dir, "user_data", "chrome_data")
|
||
if not os.path.exists(user_data_path):
|
||
os.makedirs(user_data_path, exist_ok=True)
|
||
co.set_user_data_path(user_data_path)
|
||
co.set_local_port(port=19897)
|
||
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=5)
|
||
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"]
|
||
mark = country_config.get("mark")
|
||
|
||
# 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)
|
||
|
||
self.close_init_popup()
|
||
# 2. 切换国家/设置邮编
|
||
print(f"正在检查并设置邮编: {zip_code},标识： {mark}")
|
||
self._set_zip_code(zip_code, mark)
|
||
|
||
# self.tab.wait.doc_loaded(timeout=5, raise_err=False)
|
||
|
||
# 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"数据抓取完成: {json.dumps(data)}")
|
||
return data
|
||
|
||
except Exception as e:
|
||
error_msg = f"运行出错: {traceback.format_exc()}"
|
||
print(error_msg)
|
||
show_notification(f"采集失败: {str(e)}", "error")
|
||
return {}
|
||
|
||
def _set_zip_code(self, zip_code, mark=None):
|
||
"""
|
||
设置邮编
|
||
|
||
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 mark is not None and mark in current_text:
|
||
print(f"邮编检测到标识: {mark}，无需修改")
|
||
return True
|
||
# 检查是否已经包含目标邮编
|
||
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()
|
||
|
||
continue_btn = self.tab.eles('xpath://div[@class="a-popover-footer"]//input[@id="GLUXConfirmClose"]',
|
||
timeout=10)
|
||
if len(continue_btn) > 0:
|
||
continue_btn[0].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 = {
|
||
'image_url': "",
|
||
'title': ""
|
||
}
|
||
|
||
try:
|
||
# 等待页面加载
|
||
# time.sleep(3)
|
||
title_ele = self.tab.ele('xpath://h1[@id="title"]',timeout=30)
|
||
title = title_ele.text
|
||
data["title"] = title
|
||
imge_ele = self.tab.ele('xpath://div[@id="imgTagWrapperId"]//img',timeout=20)
|
||
image_url = imge_ele.attr("src")
|
||
data["image_url"] = image_url
|
||
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 SpiderTask:
|
||
mark_name = "亚马逊采集"
|
||
|
||
def __init__(self, user_info: dict = None):
|
||
"""初始化审批任务处理器
|
||
|
||
Args:
|
||
user_info: 用户信息字典，包含 company, username, password
|
||
"""
|
||
self.user_info = user_info or {}
|
||
self.running = True
|
||
|
||
def log(self, message: str, level: str = "INFO"):
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
if level == "ERROR":
|
||
show_notification(message, "error")
|
||
print(f"[{timestamp}] [PriceTask] [{level}] {message}")
|
||
|
||
@staticmethod
|
||
def group_by_id_prefix(data):
|
||
"""
|
||
根据每条数据的 id 字段进行分组：
|
||
- 如果 id 为 "2_1"，则取 "_" 前面的 "2" 作为分组 key
|
||
- 如果 id 为 "1"，则分组 key 就是 "1"
|
||
- 相同 key 的数据归为同一组
|
||
|
||
:param data: 原始列表数据
|
||
:return: 二维列表，按 id 前缀分组
|
||
"""
|
||
grouped = defaultdict(list)
|
||
|
||
for item in data:
|
||
item_id = str(item.get("id", ""))
|
||
group_key = item_id.split("_")[0]
|
||
grouped[group_key].append(item)
|
||
|
||
return list(grouped.values())
|
||
|
||
def process_task(self, task_data: dict):
|
||
"""处理审批任务主入口
|
||
|
||
Args:
|
||
task_data: 任务数据
|
||
"""
|
||
try:
|
||
data = task_data.get("data", {})
|
||
task_id = data.get("taskId")
|
||
groups = data.get("groups")
|
||
items = groups[0].get("items", [])
|
||
|
||
# 用于测试
|
||
limit = data.get("limit", None)
|
||
|
||
if not task_id:
|
||
self.log("任务ID为空，跳过", "WARNING")
|
||
return
|
||
|
||
|
||
self.log(f"开始处理爬取任务 {task_id}，{len(items)} 个任务")
|
||
|
||
from config import runing_task
|
||
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(items) ,
|
||
"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"检测到任务 {task_id} 的暂停请求，停止处理", "WARNING")
|
||
runing_task[task_id]["status"] = "stopped"
|
||
return
|
||
max_retry = 3
|
||
show_notification(f"开始爬取数据", "info")
|
||
chrome = ChromeAmzone()
|
||
try:
|
||
# 数据整理
|
||
new_items = self.group_by_id_prefix(items)
|
||
|
||
result = []
|
||
for index,value_ls in enumerate(new_items):
|
||
|
||
return_data = {
|
||
'image_url': "",
|
||
'title': ""
|
||
}
|
||
for value in value_ls:
|
||
asin = value.get("asin")
|
||
country = value.get("country")
|
||
return_data = None
|
||
for _ in range(max_retry):
|
||
try:
|
||
return_data = chrome.run(country, asin)
|
||
self.log(f"抓取结果->{return_data}")
|
||
break
|
||
except Exception as e:
|
||
if "与页面的连接已断开" in str(e):
|
||
chrome = ChromeAmzone()
|
||
if return_data.get("image_url"):
|
||
break
|
||
|
||
# 提交结果
|
||
# 'image_url': "",
|
||
# 'title': ""
|
||
|
||
group_item = [
|
||
{
|
||
"sourceFileKey": i.get("sourceFileKey"),
|
||
"sourceFilename": i.get("sourceFilename"),
|
||
"rowToken": i.get("rowToken"),
|
||
"groupKey": i.get("groupKey"),
|
||
"id": i.get("id"),
|
||
"asin": i.get("asin"),
|
||
"country": i.get("country"),
|
||
"url": return_data.get("image_url"),
|
||
"title": return_data.get("title"),
|
||
}
|
||
for i in value_ls
|
||
]
|
||
# task_id: int, chunkIndex:int,chunkTotal: int, country_code: str, asin: str, status: dict,error:str="",
|
||
# item_data:dict={},
|
||
result.append({
|
||
"sourceFileKey": groups[0].get("sourceFileKey"),
|
||
"sourceFilename": groups[0].get("sourceFilename"),
|
||
"groupKey":groups[0].get("groupKey"),
|
||
"baseId": groups[0].get("baseId"),
|
||
"displayId": groups[0].get("displayId"),
|
||
"items": group_item
|
||
})
|
||
is_done = index == len(items)-1
|
||
if len(result) > 20 or is_done:
|
||
self.post_result(task_id=task_id,chunkIndex=index,chunkTotal=len(items),
|
||
asin=asin,item_data=result,is_done=is_done)
|
||
result = []
|
||
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:
|
||
import traceback
|
||
self.log(f"处理店铺 {task_id} 失败: {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"任务 {task_id} 已被暂停！")
|
||
else:
|
||
runing_task[task_id]["status"] = "completed"
|
||
self.log(f"任务 {task_id} 处理完成！")
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
self.log(f"任务处理失败: {traceback.format_exc()}", "ERROR")
|
||
if task_id:
|
||
from config import runing_task
|
||
if 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
|
||
"""
|
||
|
||
url = f"{DELETE_BRAND_API_BASE}/api/appearance-patent/tasks/{task_id}/result"
|
||
|
||
submission_id = f"appearance-patent:{task_id}"
|
||
payload ={
|
||
"submissionId": submission_id,
|
||
"chunkIndex": chunkIndex,
|
||
"chunkTotal": chunkTotal,
|
||
"error": error,
|
||
"items": [
|
||
item_data
|
||
],
|
||
"done": is_done
|
||
}
|
||
|
||
max_retries = 3
|
||
for retry in range(max_retries):
|
||
try:
|
||
request_timeout = 300 if is_done else 30
|
||
print("================【详情采集】=====================")
|
||
self.log(f"尝试回传结果 (第 {retry + 1}/{max_retries} 次)")
|
||
self.log(f"回传URL: {url}")
|
||
self.log(f"回传数据: {payload}")
|
||
response = requests.post(
|
||
url,
|
||
json=payload,
|
||
headers={"Content-Type": "application/json"},
|
||
timeout=request_timeout,
|
||
verify=False
|
||
)
|
||
self.log(f"回传结果: {response.text}")
|
||
data = response.json() if response.text else {}
|
||
if response.status_code == 200 and isinstance(data, dict) and data.get("success"):
|
||
self.log(f"结果回传成功: {asin} - {item_data}")
|
||
return
|
||
else:
|
||
self.log(f"结果回传失败，状态码: {response.status_code}", "WARNING")
|
||
print("=====================================")
|
||
|
||
except Exception as e:
|
||
self.log(f"调用API异常: {str(e)}", "ERROR")
|
||
print("=====================================")
|
||
|
||
# 如果还有重试机会，等待后继续
|
||
if retry < max_retries - 1:
|
||
time.sleep(2)
|
||
|
||
self.log(f"已达到最大重试次数，结果回传最终失败", "ERROR")
|
||
raise RuntimeError("已达到最大重试次数，结果回传最终失败")
|
||
|
||
if __name__ == '__main__':
|
||
spide = SpiderTask()
|
||
task_data = {
|
||
"data": {
|
||
"taskId": 1,
|
||
"groups": [
|
||
{
|
||
"sourceFileKey": "",
|
||
"sourceFilename": "",
|
||
"groupKey": "",
|
||
"baseId": "",
|
||
"displayId": "",
|
||
"items": [
|
||
{
|
||
"sourceFileKey": "",
|
||
"sourceFilename": "",
|
||
"rowToken": "",
|
||
"groupKey": "",
|
||
"id": "1",
|
||
"asin": "B0D792ND9V",
|
||
"country": "英国",
|
||
"url": "",
|
||
"title": "低"
|
||
}
|
||
]
|
||
}
|
||
],
|
||
"limit": None
|
||
},
|
||
"code": None
|
||
}
|
||
spide.process_task(task_data)
|