diff --git a/app/amazon/patrol_delete.py b/app/amazon/patrol_delete.py index 3cdfc9a..2c72d4a 100644 --- a/app/amazon/patrol_delete.py +++ b/app/amazon/patrol_delete.py @@ -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]: """只打开实测识别出的商品状态下拉框，不点击任何选项。""" - self.tab.wait.doc_loaded(raise_err=False) - result = self._run_js(LISTING_STATUS_DROPDOWN_OPEN_SCRIPT) or {} - if not result.get("ok"): - raise RuntimeError(f"未找到商品状态下拉框: {result}") - time.sleep(DROPDOWN_OPEN_DELAY) - return result + 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 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_delete_candidates = self._build_deletable_status_rows(latest_status_rows) - if not latest_delete_candidates: - stop_reason = "no-candidates" - break - - candidate_signature = tuple( - (item["canonicalStatus"], item["quantity"]) for item in 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)}" - ) - 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"]) - - result = { - "status": candidate["status"], - "canonicalStatus": candidate["canonicalStatus"], - "statusQuantity": candidate["quantity"], - "target": target, - "successMessage": success_message, + 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: + 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, } - all_results.append(result) - delete_counts[candidate["canonicalStatus"]] = delete_counts.get(candidate["canonicalStatus"], 0) + 1 - total_deleted += 1 - if max_delete_operations is not None and total_deleted >= max_delete_operations: - stop_reason = "max-delete-operations" + logger.info( + "可删除商品分类已生成: shop={}, country={}, candidates={}", + shop_name, + country, + self._format_deletable_candidates(latest_delete_candidates), + ) + stop_reason = "processed-candidates" + for candidate in latest_delete_candidates: + logger.info( + "开始处理可删除分类: shop={}, country={}, status={}, quantity={}", + shop_name, + country, + candidate["status"], + candidate["quantity"], + ) + self.select_listing_status(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": bulk_result.get("target") or {}, + "successMessage": bulk_result.get("successMessage", ""), + } + all_results.append(result) + 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() + + if stop_reason == "max-delete-operations": break + logger.info( + "可删除分类处理完成: shop={}, country={}, status={}, deleted_for_status={}", + shop_name, + country, + candidate["status"], + deleted_for_status, + ) - 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) + 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()