super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: amazon/chrome_base.py

Browse Source 
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
This commit is contained in:
铭坤
2026-05-13 16:51:29 +08:00
parent e4e01c1686
commit 79b177b345
4 changed files with 77 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

268
app/amazon/similar_asin.py
View File

@@ -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 = {