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

Expose patrol delete progress in logs

Browse Source 
Patrol delete can spend a long time switching countries, reading listing states, and confirming deletes, so the flow now emits concise progress logs around each high-value step. The logs avoid full DOM dumps outside existing diagnostics and focus on task, shop, country, status, deletion counts, and callback boundaries.

Constraint: Existing patrol delete runs are browser-driven and need observable progress without changing deletion behavior
Rejected: Log every DOM probe | too noisy for routine task monitoring
Confidence: high
Scope-risk: narrow
Tested: uv run --group dev pytest tests\\test_patrol_delete.py
Tested: uv run --group dev ruff check amazon\\patrol_delete.py
Tested: uv run python -m py_compile amazon\\patrol_delete.py
Not-tested: Live Seller Central browser run
This commit is contained in:
koko
2026-04-28 01:23:58 +08:00
parent 08997e9e20
commit 8be85fd332
1 changed files with 421 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

598
app/amazon/patrol_delete.py
View File

@@ -9,12 +9,13 @@ from typing import Any, Dict, Iterable, List, Optional
import requests
from loguru import logger
from amazon.base import AmamzonBase, UserInfo, kill_process
from amazon.base import AmamzonBase, kill_process
from amazon.tool import get_shop_info, show_notification
DROPDOWN_OPEN_DELAY = 0.8
DROPDOWN_OPTIONS_TIMEOUT = 6
LISTING_STATUS_DROPDOWN_READY_TIMEOUT = 20
QUICK_VIEW_OPEN_DELAY = 1
QUICK_VIEW_TAGS_TIMEOUT = 8
QUICK_VIEW_ENTRY_TIMEOUT = 15
@@ -37,13 +38,15 @@ def _load_patrol_delete_script(name: str) -> str:
return (SCRIPT_DIR / name).read_text(encoding="utf-8")
LISTING_STATUS_DROPDOWN_DEBUG_SCRIPT = _load_patrol_delete_script("listing_status_dropdown_debug.js")
LISTING_STATUS_DROPDOWN_OPEN_SCRIPT = _load_patrol_delete_script("listing_status_dropdown_open.js")
LISTING_STATUS_DROPDOWN_OPTIONS_SCRIPT = _load_patrol_delete_script("listing_status_dropdown_options.js")
LISTING_STATUS_CLICK_VISIBLE_STATUS_SCRIPT = _load_patrol_delete_script("listing_status_click_visible_status.js")
LISTING_STATUS_DROPDOWN_CLOSE_SCRIPT = _load_patrol_delete_script("listing_status_dropdown_close.js")
LISTING_STATUS_SELECT_OPTION_SCRIPT = _load_patrol_delete_script("listing_status_select_option.js")
LISTING_STATUS_SECTION_SCRIPT = _load_patrol_delete_script("listing_status_section.js")
COMPLETE_DRAFT_QUICK_VIEW_DEBUG_SCRIPT = _load_patrol_delete_script("complete_draft_quick_view_debug.js")
INVENTORY_SELECT_ALL_SCRIPT = _load_patrol_delete_script("inventory_select_all.js")
INVENTORY_DELETE_SELECTED_SCRIPT = _load_patrol_delete_script("inventory_delete_selected.js")
DELETE_CONFIRM_MODAL_CLICK_SCRIPT = _load_patrol_delete_script("delete_confirm_modal_click.js")
DELETE_SUCCESS_MESSAGES_SCRIPT = _load_patrol_delete_script("delete_success_messages.js")
COMPLETE_DRAFT_QUICK_VIEW_TAGS_SCRIPT = _load_patrol_delete_script("complete_draft_quick_view_tags.js")
COMPLETE_DRAFT_QUICK_VIEW_DIAGNOSTIC_SCRIPT = _load_patrol_delete_script("complete_draft_quick_view_diagnostic.js")
RECOMMENDED_OFFER_PERCENTAGE_OPEN_SCRIPT = _load_patrol_delete_script("recommended_offer_percentage_open.js")
@@ -53,12 +56,11 @@ RECOMMENDED_OFFER_PERCENTAGE_DIAGNOSTIC_SCRIPT = _load_patrol_delete_script("rec
LISTING_STATUS_SELECT_DELAY = 0.8
INVENTORY_ROW_TIMEOUT = 10
INVENTORY_LOADER_TIMEOUT = 5
DELETE_SUCCESS_TIMEOUT = 8
DELETE_SUCCESS_TIMEOUT = 20
PRODUCT_TASK_MAX_RETRIES = 3
PRODUCT_TASK_RETRY_DELAY = 2
PRODUCT_TASK_OPEN_SHOP_DELAY = 3
PRODUCT_TASK_BROWSER_RESET_DELAY = 2
PRODUCT_TASK_MAX_UNCHANGED_CANDIDATE_ROUNDS = 3
DELETE_SUCCESS_MESSAGE_PARTS = (
re.compile(r"\d+\s*个商品(?:已删除|已经删除)"),
re.compile(r"所[作做]更改需要\s*15\s*分钟才会显示在商品详情页面上"),
@@ -92,6 +94,7 @@ class InventoryManage(AmamzonBase):
切换至管理所有库存页面
1、等待 //navigation-favorites-bar[@class="hydrated"] 出现
"""
logger.info("开始切换到管理所有库存页面")
navigation = self.tab.ele('xpath://navigation-favorites-bar[@class="hydrated"]')
navigation.wait.displayed(raise_err=False)
page_btn = navigation.sr('xpath://internal-fav-bar-links[@data-internal="navigation"]').sr(
@@ -103,51 +106,35 @@ class InventoryManage(AmamzonBase):
self.tab.wait.doc_loaded()
search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group')
search_region.wait.displayed(raise_err=False)
def debug_dump_listing_status_dropdown(self) -> List[Dict[str, Any]]:
"""打印商品状态下拉框候选 DOM，供现场确认真实页面结构。"""
self.tab.wait.doc_loaded(raise_err=False)
candidates = self._run_js(LISTING_STATUS_DROPDOWN_DEBUG_SCRIPT) or []
logger.info("商品状态下拉框候选 DOM:\n{}", json.dumps(candidates, ensure_ascii=False, indent=2))
return candidates
self._wait_inventory_loader(timeout=INVENTORY_LOADER_TIMEOUT)
logger.info("已进入管理所有库存页面")
def open_listing_status_dropdown(self) -> Dict[str, Any]:
"""只打开实测识别出的商品状态下拉框，不点击任何选项。"""
deadline = time.time() + LISTING_STATUS_DROPDOWN_READY_TIMEOUT
result: Dict[str, Any] = {}
while time.time() < deadline:
self.tab.wait.doc_loaded(raise_err=False)
self._wait_inventory_loader(timeout=2)
result = self._run_js(LISTING_STATUS_DROPDOWN_OPEN_SCRIPT) or {}
if not result.get("ok"):
raise RuntimeError(f"未找到商品状态下拉框: {result}")
if result.get("ok"):
time.sleep(DROPDOWN_OPEN_DELAY)
return result
if result.get("reason") != "listing status dropdown disabled/loading":
logger.info("商品状态下拉框暂不可用，继续等待: {}", json.dumps(result, ensure_ascii=False))
time.sleep(0.5)
raise RuntimeError(f"未找到可用商品状态下拉框: {result}")
def get_listing_status_dropdown_options(self, shop_name: str, country: str) -> List[Dict[str, Any]]:
"""读取商品状态下拉框展开区域中的选项。"""
open_result = self.open_listing_status_dropdown()
option_rows = self._wait_listing_status_option_rows()
parsed_rows = self._parse_status_option_rows(option_rows, shop_name, country)
self._close_listing_status_dropdown()
if not parsed_rows:
raise RuntimeError(f"商品状态下拉框已打开但未读取到选项: {open_result}")
return parsed_rows
def print_listing_status_json(self, shop_name: str, country: str) -> List[Dict[str, Any]]:
data = {"countrySections": self.get_listing_status_country_sections(shop_name, country)}
logger.info("商品状态 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
return data["countrySections"]
def debug_dump_complete_draft_quick_view(self) -> List[Dict[str, Any]]:
"""打印补全草稿快速查看候选 DOM，供现场确认真实页面结构。"""
self.tab.wait.doc_loaded(raise_err=False)
candidates = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_DEBUG_SCRIPT) or []
logger.info("补全草稿快速查看候选 DOM:\n{}", json.dumps(candidates, ensure_ascii=False, indent=2))
return candidates
def debug_dump_complete_draft_diagnostics(self) -> Dict[str, Any]:
"""打印补全草稿相关 DOM 和资源 URL，避免输出整页商品列表。"""
self.tab.wait.doc_loaded(raise_err=False)
diagnostics = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_DIAGNOSTIC_SCRIPT) or {}
logger.info("补全草稿快速查看诊断信息:\n{}", json.dumps(diagnostics, ensure_ascii=False, indent=2))
return diagnostics
def open_complete_draft_quick_view(self) -> Dict[str, Any]:
"""直接打开补全草稿页面。"""
if self.tab is None:
@@ -166,11 +153,6 @@ class InventoryManage(AmamzonBase):
raise RuntimeError(f"补全草稿快速查看已打开但未读取到标签: {open_result}")
return parsed_rows
def print_complete_draft_quick_view_json(self, shop_name: str, country: str) -> List[Dict[str, Any]]:
data = self.get_complete_draft_quick_view_tags(shop_name, country)
logger.info("补全草稿快速查看标签 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
return data
def open_recommended_offer_percentage_detail(self) -> Dict[str, Any]:
"""打开 Seller Central 首页中的推荐报价百分比明细。"""
if self.tab is None:
@@ -202,22 +184,6 @@ class InventoryManage(AmamzonBase):
return {"cartRatios": cart_ratios}
def print_recommended_offer_cart_ratios_json(self) -> Dict[str, List[Dict[str, str]]]:
data = self.get_recommended_offer_cart_ratios()
logger.info("推荐报价百分比明细 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
return data
def debug_dump_recommended_offer_percentage_diagnostics(self) -> Dict[str, Any]:
diagnostics = self._run_js(RECOMMENDED_OFFER_PERCENTAGE_DIAGNOSTIC_SCRIPT) or {}
compact = {
"url": diagnostics.get("url"),
"title": diagnostics.get("title"),
"buyBoxFound": diagnostics.get("buyBoxFound"),
"buyBoxLeafMatches": diagnostics.get("buyBoxLeafMatches") or [],
}
logger.info("推荐报价百分比诊断信息:\n{}", json.dumps(compact, ensure_ascii=False, indent=2))
return diagnostics
def _wait_listing_status_option_rows(self) -> List[Dict[str, Any]]:
deadline = time.time() + DROPDOWN_OPTIONS_TIMEOUT
rows = []
@@ -235,44 +201,6 @@ class InventoryManage(AmamzonBase):
raise RuntimeError(f"未读取到商品状态统计标签: {rows}")
return [{"country": country, "rows": self._build_country_section_rows(parsed_rows)}]
def delete_one_listing_from_non_whitelisted_statuses(
self, shop_name: str, country: str, max_delete_count: int = 1
) -> Dict[str, Any]:
country_sections = self.get_listing_status_country_sections(shop_name, country)
status_rows = country_sections[0].get("rows") if country_sections else []
delete_candidates = self._build_deletable_status_rows(status_rows)
if not delete_candidates:
raise RuntimeError(f"商品状态中未找到可删除分类: {json.dumps(status_rows, ensure_ascii=False)}")
delete_plan = self._limit_delete_candidates(delete_candidates, max_delete_count=max_delete_count)
if not delete_plan:
raise RuntimeError(f"删除计划为空: max_delete_count={max_delete_count}")
results = []
for candidate in delete_plan:
self.select_listing_status(candidate["status"])
row = self._get_first_inventory_row(candidate["status"])
target = self._summarize_inventory_row(row)
success_message = self._delete_inventory_row(row, candidate["status"])
results.append(
{
"status": candidate["status"],
"canonicalStatus": candidate["canonicalStatus"],
"statusQuantity": candidate["quantity"],
"target": target,
"successMessage": success_message,
}
)
return {
"shopName": shop_name,
"country": country,
"statusRows": status_rows,
"deleteCandidates": delete_candidates,
"deletePlan": delete_plan,
"results": results,
}
def delete_all_listings_from_non_whitelisted_statuses(
self,
shop_name: str,
@@ -280,62 +208,121 @@ class InventoryManage(AmamzonBase):
max_delete_operations: Optional[int] = None,
) -> Dict[str, Any]:
"""循环删除非白名单分类商品，直到当前国家没有可删项。"""
logger.info(
"开始批量删除非白名单商品: shop={}, country={}, max_delete_operations={}",
shop_name,
country,
max_delete_operations,
)
all_results: List[Dict[str, Any]] = []
delete_counts: Dict[str, int] = {}
latest_status_rows: List[Dict[str, str]] = []
latest_delete_candidates: List[Dict[str, Any]] = []
previous_signature: Optional[tuple] = None
unchanged_candidate_rounds = 0
total_deleted = 0
stop_reason = "no-candidates"
while True:
country_sections = self.get_listing_status_country_sections(shop_name, country)
latest_status_rows = deepcopy(country_sections[0].get("rows") if country_sections else [])
latest_status_rows = self._read_listing_status_rows_for_delete(shop_name, country)
latest_delete_candidates = self._build_deletable_status_rows(latest_status_rows)
if not latest_delete_candidates:
stop_reason = "no-candidates"
break
logger.info("未找到可删除商品分类: shop={}, country={}", shop_name, country)
final_status_rows = latest_status_rows
return {
"shopName": shop_name,
"country": country,
"statusRows": final_status_rows,
"deleteCandidates": latest_delete_candidates,
"results": all_results,
"deleteCounts": delete_counts,
"totalDeleted": total_deleted,
"stopReason": stop_reason,
}
candidate_signature = tuple(
(item["canonicalStatus"], item["quantity"]) for item in latest_delete_candidates
logger.info(
"可删除商品分类已生成: shop={}, country={}, candidates={}",
shop_name,
country,
self._format_deletable_candidates(latest_delete_candidates),
)
if candidate_signature == previous_signature:
unchanged_candidate_rounds += 1
if unchanged_candidate_rounds >= PRODUCT_TASK_MAX_UNCHANGED_CANDIDATE_ROUNDS:
raise RuntimeError(
f"国家[{country}]删除后商品状态统计未变化，停止重试: "
f"{json.dumps(latest_delete_candidates, ensure_ascii=False)}"
stop_reason = "processed-candidates"
for candidate in latest_delete_candidates:
logger.info(
"开始处理可删除分类: shop={}, country={}, status={}, quantity={}",
shop_name,
country,
candidate["status"],
candidate["quantity"],
)
else:
unchanged_candidate_rounds = 0
previous_signature = candidate_signature
candidate = latest_delete_candidates[0]
self.select_listing_status(candidate["status"])
row = self._get_first_inventory_row(candidate["status"])
target = self._summarize_inventory_row(row)
success_message = self._delete_inventory_row(row, candidate["status"])
deleted_for_status = 0
status_quantity = candidate["quantity"]
while status_quantity <= 0 or deleted_for_status < status_quantity:
bulk_result = self._delete_selected_filtered_inventory_rows(candidate["status"])
deleted_count = int(bulk_result.get("deletedCount") or 0)
if deleted_count <= 0:
logger.info(
"当前分类未删除到商品，结束分类处理: shop={}, country={}, status={}, deleted_for_status={}",
shop_name,
country,
candidate["status"],
deleted_for_status,
)
break
result = {
"status": candidate["status"],
"canonicalStatus": candidate["canonicalStatus"],
"statusQuantity": candidate["quantity"],
"target": target,
"successMessage": success_message,
"target": bulk_result.get("target") or {},
"successMessage": bulk_result.get("successMessage", ""),
}
all_results.append(result)
delete_counts[candidate["canonicalStatus"]] = delete_counts.get(candidate["canonicalStatus"], 0) + 1
total_deleted += 1
delete_counts[candidate["canonicalStatus"]] = (
delete_counts.get(candidate["canonicalStatus"], 0) + deleted_count
)
total_deleted += deleted_count
deleted_for_status += deleted_count
logger.info(
"批量删除成功: shop={}, country={}, status={}, deleted_count={}, status_deleted={}, total_deleted={}",
shop_name,
country,
candidate["status"],
deleted_count,
deleted_for_status,
total_deleted,
)
if max_delete_operations is not None and total_deleted >= max_delete_operations:
stop_reason = "max-delete-operations"
logger.info(
"达到最大删除次数，准备停止: shop={}, country={}, total_deleted={}, limit={}",
shop_name,
country,
total_deleted,
max_delete_operations,
)
break
self._wait_inventory_loader()
final_country_sections = self.get_listing_status_country_sections(shop_name, country)
final_status_rows = deepcopy(final_country_sections[0].get("rows") if final_country_sections else latest_status_rows)
if stop_reason == "max-delete-operations":
break
logger.info(
"可删除分类处理完成: shop={}, country={}, status={}, deleted_for_status={}",
shop_name,
country,
candidate["status"],
deleted_for_status,
)
final_status_rows = self._read_listing_status_rows_for_delete(shop_name, country)
logger.info(
"批量删除非白名单商品完成: shop={}, country={}, total_deleted={}, stop_reason={}",
shop_name,
country,
total_deleted,
stop_reason,
)
return {
"shopName": shop_name,
@@ -348,32 +335,38 @@ class InventoryManage(AmamzonBase):
"stopReason": stop_reason,
}
def _read_listing_status_rows_for_delete(self, shop_name: str, country: str) -> List[Dict[str, str]]:
logger.info("开始读取商品状态行: shop={}, country={}", shop_name, country)
option_rows = self.get_listing_status_dropdown_options(shop_name, country)
rows = self._build_country_section_rows(option_rows)
logger.info("商品状态行读取完成: shop={}, country={}, row_count={}", shop_name, country, len(rows))
return rows
def select_listing_status(self, status: str) -> Dict[str, Any]:
normalized_status = self._normalize_option_text(status)
if not normalized_status:
raise RuntimeError("商品状态为空，无法执行筛选")
direct_result = self._click_visible_listing_status(normalized_status)
if direct_result.get("ok"):
time.sleep(LISTING_STATUS_SELECT_DELAY)
self._wait_inventory_loader()
return {"mode": "visible-status", **direct_result}
logger.info("开始筛选商品状态: status={}", normalized_status)
open_result = self.open_listing_status_dropdown()
option_rows = self._wait_listing_status_option_rows()
if not option_rows:
raise RuntimeError(f"商品状态下拉框已打开但未读取到选项: {open_result}")
self.open_listing_status_dropdown()
script = f"const __crawlerPayload = {json.dumps({'status': normalized_status}, ensure_ascii=False)};\n{LISTING_STATUS_SELECT_OPTION_SCRIPT}"
result = self._run_js(script) or {}
result = self._run_js(LISTING_STATUS_SELECT_OPTION_SCRIPT, {"status": normalized_status}) or {}
if not result.get("ok"):
raise RuntimeError(f"未找到商品状态选项[{normalized_status}]: {result}")
time.sleep(LISTING_STATUS_SELECT_DELAY)
self._wait_inventory_loader()
logger.info("商品状态筛选完成: status={}, result={}", normalized_status, result)
return {"mode": "dropdown-option", **result}
def _click_visible_listing_status(self, status: str) -> Dict[str, Any]:
script = (
f"const __crawlerPayload = {json.dumps({'status': status}, ensure_ascii=False)};\n"
f"{LISTING_STATUS_CLICK_VISIBLE_STATUS_SCRIPT}"
)
return self._run_js(script) or {}
def _close_listing_status_dropdown(self) -> None:
try:
self._run_js(LISTING_STATUS_DROPDOWN_CLOSE_SCRIPT)
except Exception as exc:
logger.info("关闭商品状态下拉框失败，继续执行: {}", exc)
time.sleep(0.2)
def _wait_inventory_loader(self, timeout: int = INVENTORY_LOADER_TIMEOUT) -> None:
try:
@@ -394,6 +387,61 @@ class InventoryManage(AmamzonBase):
time.sleep(0.4)
raise RuntimeError(f"状态[{status}]筛选后未找到商品行")
def _get_first_inventory_row_or_none(self, status: str):
try:
return self._get_first_inventory_row(status)
except RuntimeError as exc:
logger.info("状态[{}]未找到可删除商品行: {}", status, exc)
return None
def _delete_selected_filtered_inventory_rows(self, status: str) -> Dict[str, Any]:
logger.info("开始批量删除当前筛选结果: status={}", status)
row = self._get_first_inventory_row_or_none(status)
if row is None:
logger.info("当前筛选结果没有可删除商品: status={}", status)
return {"deletedCount": 0, "target": {}, "successMessage": ""}
target = self._summarize_inventory_row(row)
select_result = self._run_js(INVENTORY_SELECT_ALL_SCRIPT) or {}
if not select_result.get("ok"):
raise RuntimeError(f"状态[{status}]全选商品失败: {select_result}")
logger.info(
"当前筛选结果已全选: status={}, row_count={}, first_sku={}",
status,
select_result.get("rowCount"),
target.get("sku", ""),
)
delete_result: Dict[str, Any] = {}
for attempt in range(8):
time.sleep(0.8 if attempt else 0.3)
delete_result = self._run_js(INVENTORY_DELETE_SELECTED_SCRIPT) or {}
if delete_result.get("ok"):
break
if delete_result.get("menuOpened"):
logger.info("状态[{}]已点击选择组操作，等待删除菜单展开: {}", status, delete_result)
time.sleep(1)
continue
logger.info("状态[{}]第{}次点击批量删除未命中: {}", status, attempt + 1, delete_result)
if not delete_result.get("ok"):
raise RuntimeError(f"状态[{status}]点击批量删除失败: {delete_result}")
logger.info("批量删除按钮点击成功: status={}, result={}", status, delete_result)
self._confirm_delete_modal(status)
success_message = self._wait_delete_success_message(status)
deleted_count = self._extract_delete_success_count(success_message) or int(select_result.get("rowCount") or 1)
logger.info("批量删除确认完成: status={}, deleted_count={}, success_message={}", status, deleted_count, success_message)
return {
"deletedCount": deleted_count,
"target": {
**target,
"selectedRows": select_result.get("rowCount"),
"selectResult": select_result,
"deleteResult": delete_result,
},
"successMessage": success_message,
}
@classmethod
def _summarize_inventory_row(cls, row) -> Dict[str, str]:
sku = ""
@@ -408,27 +456,13 @@ class InventoryManage(AmamzonBase):
text = ""
return {"sku": sku, "rawText": text[:400]}
def _delete_inventory_row(self, row, status: str) -> str:
try:
dropdown = row.ele('xpath:.//kat-dropdown-button[@variant="secondary"]', timeout=5)
except Exception:
dropdown = None
if not dropdown:
raise RuntimeError(f"状态[{status}]商品行缺少三点菜单")
dropdown.click()
time.sleep(1)
try:
delete_button = dropdown.sr("xpath:.//button[@role='menuitem' and @data-action='DeleteListing']")
except Exception:
delete_button = None
if not delete_button:
raise RuntimeError(f"状态[{status}]三点菜单中未找到“删除商品信息”按钮")
delete_button.wait.displayed(raise_err=False)
delete_button.click()
def _confirm_delete_modal(self, status: str) -> None:
js_result = self._click_confirm_delete_modal_with_js()
if js_result.get("ok"):
logger.info("已通过 JS 确认删除弹窗: status={}, result={}", status, js_result)
return
logger.info("JS 确认删除弹窗未命中，尝试驱动点击: status={}, result={}", status, js_result)
confirm_button = self.tab.ele(
'xpath://kat-modal[@data-testid="action-modal"]//kat-button[@variant="primary"]',
timeout=5,
@@ -439,24 +473,52 @@ class InventoryManage(AmamzonBase):
confirm_button.wait.displayed(raise_err=False)
confirm_button.wait.enabled()
confirm_button.click()
return self._wait_delete_success_message(status)
logger.info("已通过驱动确认删除弹窗: status={}", status)
def _click_confirm_delete_modal_with_js(self) -> Dict[str, Any]:
try:
return self._run_js(DELETE_CONFIRM_MODAL_CLICK_SCRIPT) or {}
except Exception as exc:
logger.info("JS 点击删除确认按钮失败，回退到驱动点击: {}", exc)
return {"ok": False, "reason": str(exc)}
def _wait_delete_success_message(self, status: str) -> str:
deadline = time.time() + DELETE_SUCCESS_TIMEOUT
seen_messages = []
while time.time() < deadline:
alerts = self.tab.eles("xpath://kat-alert[@variant='success' and not(@dismissed)]", timeout=2)
for alert in alerts:
text = self._extract_alert_text(alert)
for text in self._collect_delete_success_messages():
if not text:
continue
if text not in seen_messages:
seen_messages.append(text)
if self._is_delete_success_message(text):
logger.info("读取到删除成功提示: status={}, message={}", status, text)
return text
time.sleep(0.3)
raise RuntimeError(f"状态[{status}]删除后未读取到成功提示: {seen_messages}")
def _collect_delete_success_messages(self) -> List[str]:
messages: List[str] = []
try:
alerts = self.tab.eles("xpath://kat-alert[not(@dismissed)]", timeout=1)
for alert in alerts:
text = self._extract_alert_text(alert)
if text and text not in messages:
messages.append(text)
except Exception:
pass
try:
js_messages = self._run_js(DELETE_SUCCESS_MESSAGES_SCRIPT) or []
for text in js_messages:
normalized = self._normalize_option_text(str(text or ""))
if normalized and normalized not in messages:
messages.append(normalized)
except Exception as exc:
logger.info("JS 读取删除提示失败，继续等待: {}", exc)
return messages
@classmethod
def _extract_alert_text(cls, alert) -> str:
parts = []
@@ -482,6 +544,11 @@ class InventoryManage(AmamzonBase):
normalized = re.sub(r"\s+", "", text or "")
return all(pattern.search(normalized) for pattern in DELETE_SUCCESS_MESSAGE_PARTS)
@staticmethod
def _extract_delete_success_count(text: str) -> int:
match = re.search(r"(\d+)\s*个商品(?:已删除|已经删除)", text or "")
return int(match.group(1)) if match else 0
def _wait_complete_draft_quick_view_tag_rows(self) -> List[Dict[str, Any]]:
deadline = time.time() + QUICK_VIEW_TAGS_TIMEOUT
rows = []
@@ -520,13 +587,13 @@ class InventoryManage(AmamzonBase):
logger.info("推荐报价百分比诊断信息:\n{}", json.dumps(compact, ensure_ascii=False, indent=2))
return rows
def _run_js(self, script: str):
def _run_js(self, script: str, *args):
if self.tab is None:
raise RuntimeError("浏览器页面不存在，请先打开店铺并进入管理所有库存页面")
if hasattr(self.tab, "run_js"):
return self.tab.run_js(script)
return self.tab.run_js(script, *args)
if hasattr(self.tab, "run_js_loaded"):
return self.tab.run_js_loaded(script)
return self.tab.run_js_loaded(script, *args)
raise RuntimeError("当前页面对象不支持执行 JavaScript，无法探测 Shadow DOM")
def _open_recommended_offer_percentage_detail_with_driver(self) -> Dict[str, Any]:
@@ -676,7 +743,7 @@ class InventoryManage(AmamzonBase):
quantity_text = cls._normalize_quantity_text(row.get("quantity"))
quantity = int(quantity_text) if quantity_text else 0
canonical_status = cls._canonical_listing_status(status)
if quantity <= 0 or canonical_status in LISTING_STATUS_DELETE_WHITELIST:
if not quantity_text or quantity <= 0 or canonical_status in LISTING_STATUS_DELETE_WHITELIST:
continue
candidates.append(
@@ -689,10 +756,13 @@ class InventoryManage(AmamzonBase):
return candidates
@staticmethod
def _limit_delete_candidates(candidates: Iterable[Dict[str, Any]], max_delete_count: int = 1) -> List[Dict[str, Any]]:
if max_delete_count <= 0:
return []
return list(candidates)[:max_delete_count]
def _format_deletable_candidates(candidates: Iterable[Dict[str, Any]]) -> str:
summary = []
for candidate in candidates:
status = str(candidate.get("status") or "")
quantity = candidate.get("quantity")
summary.append(f"{status}({quantity})")
return ", ".join(summary) if summary else "无"
@classmethod
def _canonical_listing_status(cls, status: str) -> str:
@@ -1157,7 +1227,7 @@ class InventoryManage(AmamzonBase):
return re.sub(r"\s+", "", text)
class ProductTask:
class PatrolDeleteTask:
"""巡店删除任务处理类。"""
mark_name = "巡店删除"
@@ -1204,7 +1274,7 @@ class ProductTask:
"stop_requested": False,
}
self.log(f"开始处理巡店删除任务 {task_id}，共 {len(items)} 个店铺")
self.log(f"开始处理巡店删除任务 {task_id}，共 {len(items)} 个店铺，{total_countries} 个国家")
for index, shop_item in enumerate(items, 1):
if runing_task[task_id].get("stop_requested", False):
self.log(f"检测到任务 {task_id} 的暂停请求，停止处理", "WARNING")
@@ -1264,6 +1334,7 @@ class ProductTask:
country_names = self._ordered_countries(payload_country_sections)
aggregated_sections = self._build_country_section_templates(payload_country_sections)
aggregated_cart_ratios = self._build_cart_ratio_templates(country_names, payload_cart_ratios)
self.log(f"店铺 {shop_name} 待处理国家: {', '.join(country_names)}")
driver: Optional[InventoryManage] = None
requires_browser_reset = False
@@ -1300,10 +1371,16 @@ class ProductTask:
if not country_result["success"]:
runing_task[task_id]["failed_countries"] += 1
requires_browser_reset = country_result["reopenRequired"]
self.log(
f"国家 {country_name} 处理结束，success={country_result['success']}，"
f"deleted={country_result['deletedCount']}，reopenRequired={requires_browser_reset}"
)
if driver is not None:
self.log(f"店铺 {shop_name} 国家处理完成，正在关闭店铺")
self._safe_close_store(driver)
self.log(f"店铺 {shop_name} 处理结果已汇总，准备回传")
self.post_result(
task_id=task_id,
shop_name=shop_name,
@@ -1315,6 +1392,7 @@ class ProductTask:
runing_task[task_id]["success_count"] += 1
else:
runing_task[task_id]["failed_count"] += 1
self.log(f"店铺 {shop_name} 处理完成，成功国家数: {sum(1 for result in country_results if result['success'])}/{len(country_names)}")
except Exception as exc:
self.log(f"处理店铺 {shop_name} 失败: {str(exc)}", "ERROR")
runing_task[task_id]["failed_count"] += 1
@@ -1405,25 +1483,27 @@ class ProductTask:
latest_status_rows: List[Dict[str, Any]] = []
try:
self._switch_country_with_retry(driver, country_name)
self.log(f"国家 {country_name} 切换完成，准备进入管理所有库存页面")
driver.switch_to_manage_inventory()
self.log(f"国家 {country_name} 已进入管理所有库存页面，开始删除非白名单商品")
delete_result = driver.delete_all_listings_from_non_whitelisted_statuses(shop_name, country_name)
latest_status_rows = delete_result.get("statusRows") or []
self.log(
f"国家 {country_name} 删除步骤完成，deleted={delete_result.get('totalDeleted', 0)}，"
f"stopReason={delete_result.get('stopReason', '')}"
)
country_section = self._build_country_section_result(
country_name=country_name,
status_rows=latest_status_rows,
delete_counts=delete_result.get("deleteCounts") or {},
default_process_status="无可删商品" if delete_result.get("totalDeleted", 0) == 0 else "",
)
complete_draft_section = self._read_complete_draft_country_section_safe(driver, shop_name, country_name)
country_section = self._replace_complete_draft_rows(country_section, complete_draft_section)
cart_ratio = self._empty_cart_ratio(country_name)
try:
cart_ratio = self._extract_country_cart_ratio(
driver.get_recommended_offer_cart_ratios().get("cartRatios") or [],
country_name,
)
except Exception as cart_exc:
self.log(f"读取国家 {country_name} 购物车比例失败: {str(cart_exc)}", "WARNING")
cart_ratio = self._read_country_cart_ratio_safe(driver, country_name)
self.log(f"国家 {country_name} 购物车比例读取完成: {cart_ratio.get('ratio', '')}")
return {
"success": True,
@@ -1434,6 +1514,7 @@ class ProductTask:
}
except Exception as exc:
self.log(f"处理国家 {country_name} 失败: {str(exc)}", "ERROR")
cart_ratio = self._read_country_cart_ratio_safe(driver, country_name)
failure_rows = latest_status_rows or [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}]
country_section = self._build_country_section_result(
country_name=country_name,
@@ -1444,16 +1525,53 @@ class ProductTask:
return {
"success": False,
"countrySection": country_section,
"cartRatio": self._empty_cart_ratio(country_name),
"cartRatio": cart_ratio,
"deletedCount": 0,
"reopenRequired": self._should_reopen_browser(str(exc)),
}
def _read_country_cart_ratio_safe(self, driver: InventoryManage, country_name: str) -> Dict[str, str]:
try:
self.log(f"开始读取国家 {country_name} 购物车比例")
return self._extract_country_cart_ratio(
driver.get_recommended_offer_cart_ratios().get("cartRatios") or [],
country_name,
)
except Exception as cart_exc:
self.log(f"读取国家 {country_name} 购物车比例失败: {str(cart_exc)}", "WARNING")
return self._empty_cart_ratio(country_name)
def _read_complete_draft_country_section_safe(
self,
driver: InventoryManage,
shop_name: str,
country_name: str,
) -> Optional[Dict[str, Any]]:
try:
self.log(f"开始读取国家 {country_name} 补全草稿")
tags = driver.get_complete_draft_quick_view_tags(shop_name, country_name)
rows = []
for item in tags:
quantity = InventoryManage._normalize_quantity_text(item.get("quantity"))
rows.append(
{
"type": "completeDraft",
"quantity": quantity,
"deleteQuantity": quantity,
}
)
self.log(f"国家 {country_name} 补全草稿读取完成: rows={rows}")
return {"country": country_name, "rows": rows} if rows else None
except Exception as draft_exc:
self.log(f"读取国家 {country_name} 补全草稿失败: {str(draft_exc)}", "WARNING")
return None
def _switch_country_with_retry(self, driver: InventoryManage, country_name: str) -> None:
for retry in range(PRODUCT_TASK_MAX_RETRIES):
try:
self.log(f"尝试切换到国家 {country_name} (第 {retry + 1}/{PRODUCT_TASK_MAX_RETRIES} 次)")
if driver.switch_to_country(country_name):
self.log(f"成功切换到国家 {country_name}")
return
if retry < PRODUCT_TASK_MAX_RETRIES - 1:
try:
@@ -1489,8 +1607,9 @@ class ProductTask:
if error:
shop_payload["error"] = error
else:
shop_payload["countrySections"] = country_sections or []
shop_payload["cartRatios"] = cart_ratios or []
shop_payload["countrySections"] = self._normalize_country_sections_for_result(country_sections)
shop_payload["cartRatios"] = self._normalize_cart_ratios_for_result(cart_ratios)
self._validate_success_result_payload(shop_payload)
payload = {"shops": [shop_payload]}
url = f"{DELETE_BRAND_API_BASE}/api/patrol-delete/tasks/{task_id}/result"
@@ -1509,10 +1628,18 @@ class ProductTask:
verify=False,
)
self.log(f"回传结果: {response.text}")
if response.status_code == 200:
try:
response_data = response.json()
except (AttributeError, ValueError):
response_data = None
api_success = not (isinstance(response_data, dict) and response_data.get("success") is False)
if response.status_code == 200 and api_success:
self.log(f"巡店删除结果回传成功: {shop_name}")
return
self.log(f"巡店删除结果回传失败，状态码: {response.status_code}", "WARNING")
self.log(
f"巡店删除结果回传失败，状态码: {response.status_code}, 响应: {response.text}",
"WARNING",
)
except Exception as exc:
self.log(f"调用巡店删除结果API异常: {str(exc)}", "ERROR")
@@ -1521,6 +1648,92 @@ class ProductTask:
raise RuntimeError("巡店删除结果回传最终失败")
@classmethod
def _normalize_country_sections_for_result(
cls, country_sections: Optional[List[Dict[str, Any]]]
) -> List[Dict[str, Any]]:
normalized = []
for section in country_sections or []:
country = str(section.get("country") or "").strip()
rows = []
for row in section.get("rows") or []:
normalized_row = {
"quantity": cls._string_result_field(row.get("quantity")),
"deleteQuantity": cls._string_result_field(row.get("deleteQuantity")),
}
if not cls._is_complete_draft_result_row(row):
normalized_row = {
"status": cls._string_result_field(row.get("status")),
**normalized_row,
"processStatus": cls._string_result_field(row.get("processStatus")),
}
rows.append(normalized_row)
if country and rows:
normalized.append({"country": country, "rows": rows})
return normalized
@staticmethod
def _normalize_cart_ratios_for_result(
cart_ratios: Optional[List[Dict[str, Any]]]
) -> List[Dict[str, str]]:
normalized = []
for item in cart_ratios or []:
country = str(item.get("country") or "").strip()
if country:
normalized.append({"country": country, "ratio": PatrolDeleteTask._string_result_field(item.get("ratio"))})
return normalized
@staticmethod
def _string_result_field(value: Any) -> str:
return "" if value is None else str(value)
@staticmethod
def _is_complete_draft_result_row(row: Dict[str, Any]) -> bool:
status = InventoryManage._normalize_option_text(str(row.get("status") or ""))
tag = InventoryManage._normalize_option_text(str(row.get("tag") or ""))
row_type = str(row.get("type") or row.get("kind") or row.get("_kind") or "").strip()
return status in {"补全草稿", "商品信息草稿"} or tag == "补全草稿" or row_type == "completeDraft"
@classmethod
def _replace_complete_draft_rows(
cls,
country_section: Dict[str, Any],
complete_draft_section: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
draft_rows = list((complete_draft_section or {}).get("rows") or [])
if not draft_rows:
return country_section
rows = []
replaced = False
for row in country_section.get("rows") or []:
if cls._is_complete_draft_result_row(row):
if not replaced:
rows.extend(draft_rows)
replaced = True
continue
rows.append(row)
if not replaced:
rows.extend(draft_rows)
return {**country_section, "rows": rows}
@staticmethod
def _validate_success_result_payload(shop_payload: Dict[str, Any]) -> None:
country_sections = shop_payload.get("countrySections") or []
cart_ratios = shop_payload.get("cartRatios") or []
if not country_sections:
raise RuntimeError("巡店删除成功结果缺少国家状态数据，已阻止空回传")
empty_sections = [
str(section.get("country") or "")
for section in country_sections
if not section.get("rows")
]
if empty_sections:
raise RuntimeError(f"巡店删除成功结果存在空国家状态行: {', '.join(empty_sections)}")
if not cart_ratios:
raise RuntimeError("巡店删除成功结果缺少购物车比例数据，已阻止空回传")
@staticmethod
def _resolve_country_sections(
shop_item: Dict[str, Any], country_sections: List[Dict[str, Any]]
@@ -1607,11 +1820,14 @@ class ProductTask:
if not process_status and default_process_status and not default_status_applied:
process_status = default_process_status
default_status_applied = True
delete_quantity = str(delete_count) if delete_count > 0 else ""
if default_process_status == "无可删商品":
delete_quantity = "0"
rows.append(
{
"status": status,
"quantity": InventoryManage._normalize_quantity_text(row.get("quantity")),
"deleteQuantity": str(delete_count) if delete_count > 0 else "",
"deleteQuantity": delete_quantity,
"processStatus": process_status,
}
)
@@ -1621,7 +1837,7 @@ class ProductTask:
{
"status": "全部",
"quantity": "",
"deleteQuantity": "",
"deleteQuantity": "0" if default_process_status == "无可删商品" else "",
"processStatus": default_process_status,
}
]
@@ -1664,15 +1880,3 @@ class ProductTask:
driver.close_store()
except Exception:
pass
if __name__ == "__main__":
user_info: UserInfo = {"company": "rongchuang123", "username": "自动化_Robot", "password": "#20zsg25"}
driver = InventoryManage(user_info)
shop_name = "郭亚芳"
country = "德国"
driver.open_shop(shop_name)
driver.switch_to_country(country)
driver.print_recommended_offer_cart_ratios_json()