import json import re import sys import time from copy import deepcopy from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Set import requests from loguru import logger 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 COMPLETE_DRAFT_PAGE_URL = "https://sellercentral.amazon.co.uk/myinventory/inventory/views/drafts?subview=unsubmitted-drafts" RECOMMENDED_OFFER_HOME_URL = "https://sellercentral.amazon.co.uk/home" RECOMMENDED_OFFER_OPEN_DELAY = 1 RECOMMENDED_OFFER_ROWS_TIMEOUT = 10 RECOMMENDED_OFFER_COUNTRIES = ["欧洲", "英国", "德国", "法国", "西班牙", "意大利"] RECOMMENDED_OFFER_EXPAND_XPATHS = [ 'xpath://div[@data-key="KPI_CARD_BUYBOX"]//casino-button[@data-actionname="expand"]', 'xpath://casino-card[@id="KPI_CARD_BUYBOX"]//casino-button[@data-actionname="expand"]', 'xpath://div[@data-key="KPI_CARD_BUYBOX"]//*[@aria-label="chevron-down"]', ] SCRIPT_DIR = Path(__file__).with_name("scripts") / "patrol_delete" def _new_trace_id(prefix: str) -> str: return f"{prefix}-{int(time.time() * 1000) % 1000000:06d}" def _safe_log_text(value: Any, limit: int = 3000) -> str: text = "" if value is None else str(value) text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", " ", text) if len(text) > limit: text = text[: limit - 3] + "..." encoding = getattr(getattr(sys, "stderr", None), "encoding", None) or "utf-8" return text.encode(encoding, errors="backslashreplace").decode(encoding, errors="replace") def _load_patrol_delete_script(name: str) -> str: return (SCRIPT_DIR / name).read_text(encoding="utf-8") 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_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") 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") COMPLETE_DRAFT_SELECT_TAG_SCRIPT = _load_patrol_delete_script("complete_draft_select_tag.js") COMPLETE_DRAFT_SELECT_ALL_SCRIPT = _load_patrol_delete_script("complete_draft_select_all.js") COMPLETE_DRAFT_DELETE_SELECTED_SCRIPT = _load_patrol_delete_script("complete_draft_delete_selected.js") RECOMMENDED_OFFER_PERCENTAGE_OPEN_SCRIPT = _load_patrol_delete_script("recommended_offer_percentage_open.js") RECOMMENDED_OFFER_PERCENTAGE_ROWS_SCRIPT = _load_patrol_delete_script("recommended_offer_percentage_rows.js") RECOMMENDED_OFFER_PERCENTAGE_DIAGNOSTIC_SCRIPT = _load_patrol_delete_script("recommended_offer_percentage_diagnostic.js") LISTING_STATUS_SELECT_DELAY = 0.8 INVENTORY_ROW_TIMEOUT = 10 INVENTORY_LOADER_TIMEOUT = 5 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 DELETE_SUCCESS_COUNT_PATTERNS = ( re.compile(r"(\d+)\s*个商品(?:已删除|已经删除)"), re.compile(r"(?:已删除|已经删除)\s*(\d+)\s*个商品"), ) DELETE_SUCCESS_TEXT_PATTERNS = ( re.compile(r"一个或多个商品(?:已删除|已经删除)"), ) COMPLETE_DRAFT_DELETE_SUCCESS_PATTERNS = ( re.compile(r"商品信息草稿(?:已删除|已经删除)"), re.compile(r"已删除\s*\d*\s*个?\s*商品信息草稿"), ) LISTING_STATUS_DELETE_WHITELIST = { "全部", "定价问题", "需要批准", "在搜索结果中禁止显示", "配送问题", "缺少报价", "已停售", "不可售", "在售", } LISTING_STATUS_COMPATIBILITY_MAP = { "搜尋結果中禁止顯示": "在搜索结果中禁止显示", "搜索结果中禁止显示": "在搜索结果中禁止显示", "在搜索中禁止显示结果": "在搜索结果中禁止显示", "配送問題": "配送问题", "缺少報價": "缺少报价", "订单页面已删除": "详情页面已删除", } class InventoryManage(AmamzonBase): mark_name = "管理所有库存" def switch_to_manage_inventory(self): """ 切换至管理所有库存页面 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( 'xpath://a[@data-page-id="ezdpc-gui-inventory-mons"]' ) page_btn.wait.displayed(raise_err=False) page_btn.click(timeout=5) self.tab.wait.doc_loaded() search_region = self.tab.ele('xpath://div[@id="searchBoxContainer"]//kat-input-group') search_region.wait.displayed(raise_err=False) 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 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 open_complete_draft_quick_view(self) -> Dict[str, Any]: """直接打开补全草稿页面。""" if self.tab is None: raise RuntimeError("浏览器页面不存在，请先打开店铺") self.tab.get(COMPLETE_DRAFT_PAGE_URL) self.tab.wait.doc_loaded(raise_err=False) time.sleep(QUICK_VIEW_OPEN_DELAY) return {"ok": True, "url": COMPLETE_DRAFT_PAGE_URL, "title": getattr(self.tab, "title", "")} def get_complete_draft_quick_view_tags(self, shop_name: str, country: str) -> List[Dict[str, Any]]: """读取补全草稿快速查看区域中的全部“标签 + 数量”项。""" open_result = self.open_complete_draft_quick_view() tag_rows = self._wait_complete_draft_quick_view_tag_rows() parsed_rows = self._parse_quick_view_tag_rows(tag_rows, shop_name, country) if not parsed_rows: raise RuntimeError(f"补全草稿快速查看已打开但未读取到标签: {open_result}") return parsed_rows def delete_complete_drafts_from_non_whitelisted_tags( self, shop_name: str, country: str, max_delete_operations: Optional[int] = None, trace_id: str = "", delete_conditions: Optional[Iterable[Any]] = None, allowed_statuses: Optional[Iterable[str]] = None, ) -> Dict[str, Any]: """按补全草稿快速查看的准确标签数量，删除非白名单标签下方商品。""" trace_id = trace_id or _new_trace_id("draft") effective_conditions = delete_conditions if delete_conditions is not None else allowed_statuses logger.info( f"trace_id={trace_id} 开始删除非白名单补全草稿: " f"shop={shop_name}, country={country}, max_delete_operations={max_delete_operations}, " f"delete_conditions={self._format_delete_conditions(effective_conditions)}" ) initial_tags = self.get_complete_draft_quick_view_tags(shop_name, country) delete_candidates = self._build_deletable_complete_draft_tag_rows(initial_tags, effective_conditions) results: List[Dict[str, Any]] = [] delete_counts: Dict[str, int] = {} total_deleted = 0 stop_reason = "no-candidates" if not delete_candidates: logger.info(f"trace_id={trace_id} 补全草稿没有可删除标签: shop={shop_name}, country={country}") return { "shopName": shop_name, "country": country, "draftRows": initial_tags, "deleteCandidates": delete_candidates, "results": results, "deleteCounts": delete_counts, "totalDeleted": total_deleted, "stopReason": stop_reason, } stop_reason = "processed-candidates" for candidate in delete_candidates: tag = candidate["tag"] quantity = candidate["quantity"] canonical_tag = candidate["canonicalTag"] logger.info( f"trace_id={trace_id} 开始删除补全草稿标签下方商品: " f"shop={shop_name}, country={country}, tag={tag}, quantity={quantity}" ) self.select_complete_draft_tag(tag, trace_id=trace_id) deleted_for_tag = 0 while quantity <= 0 or deleted_for_tag < quantity: remaining_limit = None if max_delete_operations is not None: remaining_limit = max(0, max_delete_operations - total_deleted) if remaining_limit <= 0: stop_reason = "max-delete-operations" break select_result = self._select_all_complete_draft_rows( tag, trace_id=trace_id, max_select_count=remaining_limit, ) selected_count = int(select_result.get("selectedCount") or select_result.get("rowCount") or 0) if select_result.get("empty"): logger.info( f"trace_id={trace_id} 补全草稿标签下方已无商品，跳过删除: " f"shop={shop_name}, country={country}, tag={tag}" ) break if selected_count <= 0: logger.info( f"trace_id={trace_id} 补全草稿标签下方没有可删除商品: " f"shop={shop_name}, country={country}, tag={tag}" ) break delete_result = self._click_complete_draft_delete_selected(tag, trace_id=trace_id) self._confirm_delete_modal(tag, trace_id=trace_id) self._wait_complete_draft_rows_after_delete(tag, select_result, trace_id=trace_id) deleted_count = min(selected_count, quantity - deleted_for_tag) if quantity > 0 else selected_count deleted_for_tag += deleted_count total_deleted += deleted_count delete_counts[canonical_tag] = delete_counts.get(canonical_tag, 0) + deleted_count results.append( { "tag": tag, "canonicalTag": canonical_tag, "tagQuantity": quantity, "deletedCount": deleted_count, "selectResult": select_result, "deleteResult": delete_result, } ) logger.info( f"trace_id={trace_id} 补全草稿标签删除成功: " f"shop={shop_name}, country={country}, tag={tag}, " f"deleted_count={deleted_count}, tag_deleted={deleted_for_tag}, total_deleted={total_deleted}" ) if max_delete_operations is not None and total_deleted >= max_delete_operations: stop_reason = "max-delete-operations" break if quantity > 0 and deleted_for_tag >= quantity: break if stop_reason == "max-delete-operations": break final_tags = self.get_complete_draft_quick_view_tags(shop_name, country) return { "shopName": shop_name, "country": country, "draftRows": final_tags, "deleteCandidates": delete_candidates, "results": results, "deleteCounts": delete_counts, "totalDeleted": total_deleted, "stopReason": stop_reason, } def select_complete_draft_tag(self, tag: str, trace_id: str = "") -> Dict[str, Any]: normalized_tag = self._normalize_option_text(tag) if not normalized_tag: raise RuntimeError("补全草稿标签为空，无法执行筛选") trace_id = trace_id or _new_trace_id("draft") result = self._run_js(COMPLETE_DRAFT_SELECT_TAG_SCRIPT, {"tag": normalized_tag}) or {} if not result.get("ok"): logger.info( f"trace_id={trace_id} 补全草稿快速查看标签未命中: " f"tag={normalized_tag}, result={_safe_log_text(result)}" ) raise RuntimeError(f"未找到补全草稿标签[{normalized_tag}]: {result}") time.sleep(LISTING_STATUS_SELECT_DELAY) return result def _select_all_complete_draft_rows( self, tag: str, trace_id: str = "", max_select_count: Optional[int] = None, ) -> Dict[str, Any]: trace_id = trace_id or _new_trace_id("draft") deadline = time.time() + INVENTORY_ROW_TIMEOUT result: Dict[str, Any] = {} saw_empty = False while time.time() < deadline: payload = {"maxSelectCount": max_select_count} if max_select_count is not None else None result = self._run_js(COMPLETE_DRAFT_SELECT_ALL_SCRIPT, payload) if payload else self._run_js(COMPLETE_DRAFT_SELECT_ALL_SCRIPT) result = result or {} if result.get("ok"): time.sleep(0.5) logger.info(f"trace_id={trace_id} 补全草稿已全选: tag={tag}, result={_safe_log_text(result)}") return result if result.get("reason") == "no complete draft rows": saw_empty = True time.sleep(0.5) continue if result.get("reason") and result.get("reason") != "no complete draft rows": break time.sleep(0.5) if not result.get("ok"): if saw_empty and result.get("reason") == "no complete draft rows": logger.info(f"trace_id={trace_id} 补全草稿标签下方无商品行: tag={tag}, result={_safe_log_text(result)}") return {**result, "ok": True, "empty": True, "selectedCount": 0, "rowCount": 0} logger.info( f"trace_id={trace_id} 补全草稿全选失败: tag={tag}, result={_safe_log_text(result)}" ) raise RuntimeError(f"补全草稿标签[{tag}]全选失败: {result}") return result def _click_complete_draft_delete_selected(self, tag: str, trace_id: str = "") -> Dict[str, Any]: trace_id = trace_id or _new_trace_id("draft") result: Dict[str, Any] = {} for attempt in range(8): result = self._run_js(COMPLETE_DRAFT_DELETE_SELECTED_SCRIPT) or {} if result.get("ok"): break if result.get("menuOpened"): logger.info( f"trace_id={trace_id} 补全草稿选择组操作菜单已打开，等待删除选项出现: " f"tag={tag}, attempt={attempt + 1}, result={_safe_log_text(result)}" ) time.sleep(0.8) continue break if not result.get("ok"): logger.info( f"trace_id={trace_id} 补全草稿点击删除失败: tag={tag}, result={_safe_log_text(result)}" ) raise RuntimeError(f"补全草稿标签[{tag}]点击删除失败: {result}") time.sleep(0.5) logger.info(f"trace_id={trace_id} 补全草稿删除按钮点击成功: tag={tag}, result={_safe_log_text(result)}") return result def _wait_complete_draft_rows_after_delete( self, tag: str, before_result: Dict[str, Any], trace_id: str = "", ) -> None: trace_id = trace_id or _new_trace_id("draft") before_count = int(before_result.get("rowCount") or 0) deadline = time.time() + DELETE_SUCCESS_TIMEOUT last_result: Dict[str, Any] = {} while time.time() < deadline: time.sleep(0.5) result = self._run_js(COMPLETE_DRAFT_SELECT_ALL_SCRIPT, {"probeOnly": True}) or {} last_result = result row_count = int(result.get("rowCount") or 0) if row_count < before_count or row_count == 0: logger.info( f"trace_id={trace_id} 补全草稿删除后行数已变化: " f"tag={tag}, before={before_count}, after={row_count}" ) return for text in self._collect_delete_success_messages(): if self._is_complete_draft_delete_success_message(text) or self._is_delete_success_message(text): logger.info(f"trace_id={trace_id} 补全草稿删除读取到成功提示: tag={tag}, message={_safe_log_text(text)}") return logger.info( f"trace_id={trace_id} 补全草稿删除后未确认行数变化: " f"tag={tag}, before={before_count}, last_result={_safe_log_text(last_result)}" ) raise RuntimeError("补全草稿删除后未确认行数变化") def open_recommended_offer_percentage_detail(self) -> Dict[str, Any]: """打开 Seller Central 首页中的推荐报价百分比明细。""" if self.tab is None: raise RuntimeError("浏览器页面不存在，请先打开店铺") self.tab.get(RECOMMENDED_OFFER_HOME_URL) self.tab.wait.doc_loaded(raise_err=False) result = self._open_recommended_offer_percentage_detail_with_driver() if not result.get("ok"): result = self._run_js(RECOMMENDED_OFFER_PERCENTAGE_OPEN_SCRIPT) or {} if not result.get("ok"): raise RuntimeError(f"未找到推荐报价百分比入口: {result}") time.sleep(RECOMMENDED_OFFER_OPEN_DELAY) return result def get_recommended_offer_cart_ratios(self) -> Dict[str, List[Dict[str, str]]]: """打开一次推荐报价百分比下拉，读取全部目标商城的明细。""" open_result = self.open_recommended_offer_percentage_detail() rows = self._wait_recommended_offer_percentage_rows() cart_ratios = self._parse_all_cart_ratio_rows(rows) incomplete = [item["country"] for item in cart_ratios if not item["ratio2DaysAgo"] or not item["ratio30DaysAgo"]] if incomplete: logger.info( "推荐报价百分比未完整读取: countries={}, open_result={}, rows={}", ",".join(incomplete), json.dumps(open_result, ensure_ascii=False), json.dumps(rows, ensure_ascii=False), ) return {"cartRatios": cart_ratios} def _wait_listing_status_option_rows(self) -> List[Dict[str, Any]]: deadline = time.time() + DROPDOWN_OPTIONS_TIMEOUT rows = [] while time.time() < deadline: rows = self._run_js(LISTING_STATUS_DROPDOWN_OPTIONS_SCRIPT) or [] if rows: return rows time.sleep(0.3) return rows def get_listing_status_country_sections(self, shop_name: str, country: str) -> List[Dict[str, Any]]: trace_id = _new_trace_id("status") logger.info(f"trace_id={trace_id} 开始读取商品状态统计: shop={shop_name}, country={country}") rows = self._wait_listing_status_section_rows() parsed_rows = self._parse_listing_status_section_rows(rows) if not parsed_rows: logger.info( f"trace_id={trace_id} 商品状态统计读取为空: " f"raw_rows={_safe_log_text(json.dumps(rows, ensure_ascii=False))}" ) raise RuntimeError(f"未读取到商品状态统计标签: {rows}") logger.info( f"trace_id={trace_id} 商品状态统计读取完成: " f"shop={shop_name}, country={country}, row_count={len(parsed_rows)}" ) return [{"country": country, "rows": self._build_country_section_rows(parsed_rows)}] def delete_all_listings_from_non_whitelisted_statuses( self, shop_name: str, country: str, max_delete_operations: Optional[int] = None, trace_id: str = "", delete_conditions: Optional[Iterable[Any]] = None, allowed_statuses: Optional[Iterable[str]] = None, ) -> Dict[str, Any]: """循环删除非白名单分类商品，直到当前国家没有可删项。""" trace_id = trace_id or _new_trace_id("del") effective_conditions = delete_conditions if delete_conditions is not None else allowed_statuses logger.info( f"trace_id={trace_id} 开始批量删除非白名单商品: " f"shop={shop_name}, country={country}, max_delete_operations={max_delete_operations}, " f"delete_conditions={self._format_delete_conditions(effective_conditions)}" ) 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]] = [] total_deleted = 0 stop_reason = "no-candidates" latest_status_rows = self._read_listing_status_rows_for_delete(shop_name, country, trace_id=trace_id) latest_delete_candidates = self._build_deletable_status_rows( latest_status_rows, delete_conditions=effective_conditions, ) if not latest_delete_candidates: logger.info(f"trace_id={trace_id} 未找到可删除商品分类: shop={shop_name}, country={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, } logger.info( f"trace_id={trace_id} 可删除商品分类已生成: shop={shop_name}, country={country}, " f"candidates={self._format_deletable_candidates(latest_delete_candidates)}" ) stop_reason = "processed-candidates" for candidate in latest_delete_candidates: logger.info( f"trace_id={trace_id} 开始处理可删除分类: shop={shop_name}, country={country}, " f"status={candidate['status']}, quantity={candidate['quantity']}" ) self.select_listing_status(candidate["status"], trace_id=trace_id) 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"], trace_id=trace_id) deleted_count = int(bulk_result.get("deletedCount") or 0) if deleted_count <= 0: logger.info( f"trace_id={trace_id} 当前分类未删除到商品，结束分类处理: " f"shop={shop_name}, country={country}, status={candidate['status']}, " f"deleted_for_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( f"trace_id={trace_id} 批量删除成功: shop={shop_name}, country={country}, " f"status={candidate['status']}, deleted_count={deleted_count}, " f"status_deleted={deleted_for_status}, total_deleted={total_deleted}, " f"success_message={bulk_result.get('successMessage', '')}" ) if max_delete_operations is not None and total_deleted >= max_delete_operations: stop_reason = "max-delete-operations" logger.info( f"trace_id={trace_id} 达到最大删除次数，准备停止: " f"shop={shop_name}, country={country}, total_deleted={total_deleted}, " f"limit={max_delete_operations}" ) break self._wait_inventory_loader() if stop_reason == "max-delete-operations": break logger.info( f"trace_id={trace_id} 可删除分类处理完成: shop={shop_name}, country={country}, " f"status={candidate['status']}, deleted_for_status={deleted_for_status}" ) final_status_rows = self._read_listing_status_rows_for_delete(shop_name, country, trace_id=trace_id) logger.info( f"trace_id={trace_id} 批量删除非白名单商品完成: " f"shop={shop_name}, country={country}, total_deleted={total_deleted}, stop_reason={stop_reason}" ) 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, } def _read_listing_status_rows_for_delete( self, shop_name: str, country: str, trace_id: str = "", ) -> List[Dict[str, str]]: trace_id = trace_id or _new_trace_id("rows") logger.info(f"trace_id={trace_id} 开始读取商品状态行: shop={shop_name}, country={country}") option_rows = self.get_listing_status_dropdown_options(shop_name, country) rows = self._build_country_section_rows(option_rows) logger.info(f"trace_id={trace_id} 商品状态行读取完成: shop={shop_name}, country={country}, row_count={len(rows)}") return rows def select_listing_status(self, status: str, trace_id: str = "") -> Dict[str, Any]: normalized_status = self._normalize_option_text(status) if not normalized_status: raise RuntimeError("商品状态为空，无法执行筛选") trace_id = trace_id or _new_trace_id("status") logger.info(f"trace_id={trace_id} 开始筛选商品状态: status={normalized_status}") open_result = self.open_listing_status_dropdown() option_rows = self._wait_listing_status_option_rows() if not option_rows: logger.info( f"trace_id={trace_id} 商品状态下拉选项为空: " f"open_result={_safe_log_text(json.dumps(open_result, ensure_ascii=False))}" ) raise RuntimeError(f"商品状态下拉框已打开但未读取到选项: {open_result}") result = self._run_js(LISTING_STATUS_SELECT_OPTION_SCRIPT, {"status": normalized_status}) or {} if not result.get("ok"): logger.info( f"trace_id={trace_id} 商品状态选项未命中: " f"status={normalized_status}, result={_safe_log_text(result)}" ) raise RuntimeError(f"未找到商品状态选项[{normalized_status}]: {result}") time.sleep(LISTING_STATUS_SELECT_DELAY) self._wait_inventory_loader() logger.info(f"trace_id={trace_id} 商品状态筛选完成: status={normalized_status}, result={result}") return {"mode": "dropdown-option", **result} 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: loader = self.tab.ele("xpath://div[contains(@class,'Loader-module__loader')]", timeout=1) except Exception: loader = None if loader: loader.wait.deleted(timeout=timeout, raise_err=False) time.sleep(0.5) def _get_first_inventory_row(self, status: str): deadline = time.time() + INVENTORY_ROW_TIMEOUT while time.time() < deadline: self._wait_inventory_loader(timeout=2) rows = self.tab.eles("xpath://div[@data-sku]", timeout=2) if rows: return rows[0] 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, trace_id: str = "") -> Dict[str, Any]: trace_id = trace_id or _new_trace_id("del") logger.info(f"trace_id={trace_id} 开始批量删除当前筛选结果: status={status}") row = self._get_first_inventory_row_or_none(status) if row is None: logger.info(f"trace_id={trace_id} 当前筛选结果没有可删除商品: 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"): logger.info(f"trace_id={trace_id} 全选商品失败: status={status}, result={_safe_log_text(select_result)}") raise RuntimeError(f"状态[{status}]全选商品失败: {select_result}") logger.info( f"trace_id={trace_id} 当前筛选结果已全选: status={status}, " f"row_count={select_result.get('rowCount')}, first_sku={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( f"trace_id={trace_id} 状态[{status}]已点击选择组操作，等待删除菜单展开: " f"{_safe_log_text(delete_result)}" ) time.sleep(1) continue logger.info( f"trace_id={trace_id} 状态[{status}]第{attempt + 1}次点击批量删除未命中: " f"{_safe_log_text(delete_result)}" ) if not delete_result.get("ok"): logger.info(f"trace_id={trace_id} 点击批量删除失败: status={status}, result={_safe_log_text(delete_result)}") raise RuntimeError(f"状态[{status}]点击批量删除失败: {delete_result}") logger.info(f"trace_id={trace_id} 批量删除按钮点击成功: status={status}, result={_safe_log_text(delete_result)}") self._confirm_delete_modal(status, trace_id=trace_id) success_message = self._wait_delete_success_message(status, trace_id=trace_id) deleted_count = self._extract_delete_success_count(success_message) or int(select_result.get("rowCount") or 1) logger.info( f"trace_id={trace_id} 批量删除确认完成: status={status}, " f"deleted_count={deleted_count}, success_message={_safe_log_text(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 = "" text = "" try: sku = row.attr("data-sku") or "" except Exception: sku = "" try: text = cls._normalize_option_text(getattr(row, "text", "") or "") except Exception: text = "" return {"sku": sku, "rawText": text[:400]} def _confirm_delete_modal(self, status: str, trace_id: str = "") -> None: trace_id = trace_id or _new_trace_id("modal") js_result = self._click_confirm_delete_modal_with_js() if js_result.get("ok"): logger.info(f"trace_id={trace_id} 已通过 JS 确认删除弹窗: status={status}, result={_safe_log_text(js_result)}") return logger.info( f"trace_id={trace_id} JS 确认删除弹窗未命中，尝试驱动点击: " f"status={status}, result={_safe_log_text(js_result)}" ) confirm_button = self.tab.ele( 'xpath://kat-modal[@data-testid="action-modal"]//kat-button[@variant="primary"]', timeout=5, ) if not confirm_button: logger.info(f"trace_id={trace_id} 删除确认弹窗未出现: status={status}") raise RuntimeError(f"状态[{status}]未出现删除确认弹窗") confirm_button.wait.displayed(raise_err=False) confirm_button.wait.enabled() confirm_button.click() logger.info(f"trace_id={trace_id} 已通过驱动确认删除弹窗: 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, trace_id: str = "") -> str: trace_id = trace_id or _new_trace_id("toast") deadline = time.time() + DELETE_SUCCESS_TIMEOUT seen_messages = [] while time.time() < deadline: 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(f"trace_id={trace_id} 读取到删除成功提示: status={status}, message={_safe_log_text(text)}") return text time.sleep(0.3) logger.info( f"trace_id={trace_id} 删除后未读取到成功提示: " f"status={status}, seen_messages={_safe_log_text(seen_messages)}" ) raise RuntimeError("删除后未读取到成功提示") 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 = [] try: text = cls._normalize_option_text(getattr(alert, "text", "") or "") if text: parts.append(text) except Exception: pass for attr_name in ("description", "text", "label", "header"): try: value = cls._normalize_option_text(alert.attr(attr_name) or "") except Exception: value = "" if value and value not in parts: parts.append(value) return " ".join(parts).strip() @staticmethod def _is_delete_success_message(text: str) -> bool: normalized = re.sub(r"\s+", "", text or "") if InventoryManage._extract_delete_success_count(normalized) > 0: return True return any(pattern.search(normalized) for pattern in DELETE_SUCCESS_TEXT_PATTERNS) @staticmethod def _is_complete_draft_delete_success_message(text: str) -> bool: normalized = re.sub(r"\s+", "", text or "") return any(pattern.search(normalized) for pattern in COMPLETE_DRAFT_DELETE_SUCCESS_PATTERNS) @staticmethod def _extract_delete_success_count(text: str) -> int: normalized = re.sub(r"\s+", "", text or "") for pattern in DELETE_SUCCESS_COUNT_PATTERNS: match = pattern.search(normalized) if match: return int(match.group(1)) return 0 def _wait_complete_draft_quick_view_tag_rows(self) -> List[Dict[str, Any]]: deadline = time.time() + QUICK_VIEW_TAGS_TIMEOUT rows = [] last_result: Dict[str, Any] = {} while time.time() < deadline: result = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_TAGS_SCRIPT) or {} last_result = result rows = result.get("rows") or [] if rows: return rows time.sleep(0.3) diagnostics = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_DIAGNOSTIC_SCRIPT) or {} logger.info("补全草稿快速查看未读取到标签，最后一次 DOM 探测:\n{}", json.dumps(last_result, ensure_ascii=False, indent=2)) logger.info("补全草稿快速查看诊断信息:\n{}", json.dumps(diagnostics, ensure_ascii=False, indent=2)) return rows def _wait_recommended_offer_percentage_rows(self) -> List[Dict[str, Any]]: deadline = time.time() + RECOMMENDED_OFFER_ROWS_TIMEOUT rows = [] last_result: Dict[str, Any] = {} while time.time() < deadline: result = self._run_js(RECOMMENDED_OFFER_PERCENTAGE_ROWS_SCRIPT) or {} last_result = result rows = list(result.get("rows") or []) if self._has_complete_cart_ratio_rows(rows): return rows time.sleep(0.3) logger.info("推荐报价百分比明细未读取到目标日期，最后一次 DOM 探测:\n{}", json.dumps(last_result, ensure_ascii=False, indent=2)) 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 rows 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, *args) if hasattr(self.tab, "run_js_loaded"): 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]: if self.tab is None: return {"ok": False, "reason": "tab missing"} for xpath in RECOMMENDED_OFFER_EXPAND_XPATHS: try: button = self.tab.ele(xpath, timeout=3) except Exception as exc: logger.info("定位推荐报价展开按钮失败: xpath={}, error={}", xpath, exc) button = None if not button: continue try: button.click() return {"ok": True, "source": "driver", "xpath": xpath} except Exception as exc: logger.info("点击推荐报价展开按钮失败: xpath={}, error={}", xpath, exc) return {"ok": False, "reason": "expand button not found by driver"} def _wait_listing_status_section_rows(self) -> List[Dict[str, Any]]: deadline = time.time() + DROPDOWN_OPTIONS_TIMEOUT rows = [] while time.time() < deadline: result = self._run_js(LISTING_STATUS_SECTION_SCRIPT) or {} rows = result.get("rows") or [] if self._has_listing_status_rows(rows): return rows time.sleep(0.3) return rows @staticmethod def _has_listing_status_rows(rows: Iterable[Dict[str, Any]]) -> bool: parsed_rows = [row for row in rows if (row.get("status") or "").strip()] quantity_count = sum(1 for row in parsed_rows if str(row.get("quantity") or "").strip()) return len(parsed_rows) >= 3 and quantity_count >= 2 @classmethod def _parse_status_option_rows( cls, option_rows: Iterable[Dict[str, Any]], shop_name: str, country: str ) -> List[Dict[str, Any]]: rows = [] for option in option_rows: raw_text = cls._normalize_option_text(option.get("raw_text") or "") value = option.get("value") if not raw_text and not value: continue status, quantity = cls._parse_status_text(raw_text or str(value)) if not status: continue rows.append( { "shop_name": shop_name, "country": country, "status": status, "quantity": quantity, "value": value, "raw_text": raw_text, } ) return rows @classmethod def _parse_listing_status_section_rows(cls, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: parsed_rows = [] seen = set() for row in rows: status = cls._normalize_option_text(row.get("status") or "") quantity = cls._normalize_quantity_text(row.get("quantity")) raw_text = cls._normalize_option_text(row.get("raw_text") or "") if not status and raw_text: for item in cls._extract_listing_status_rows_from_text(raw_text): key = (item["status"], item["quantity"]) if key in seen: continue seen.add(key) parsed_rows.append(item) continue if not status: continue key = (status, quantity) if key in seen: continue seen.add(key) parsed_rows.append({"status": status, "quantity": quantity, "raw_text": raw_text or status}) return parsed_rows @classmethod def _extract_listing_status_rows_from_text(cls, raw_text: str) -> List[Dict[str, Any]]: text = cls._normalize_option_text(raw_text) if not text: return [] pattern = re.compile( r"(全部|在搜索结果中禁止显示|配送问题|缺少报价|已停售|详情页面已删除|订单页面已删除|商品信息草稿|缺少的信息|定价问题|需要批准|不可售|在售)" r"\s*(?:[（(]\s*([\d,]+)\s*[）)])?" ) rows = [] for match in pattern.finditer(text): rows.append( { "status": match.group(1).strip(), "quantity": cls._normalize_quantity_text(match.group(2)), "raw_text": cls._normalize_option_text(match.group(0)), } ) return rows @staticmethod def _normalize_quantity_text(quantity: Any) -> str: if quantity is None: return "" text = re.sub(r"[^\d,]", "", str(quantity)) return text.replace(",", "") @classmethod def _build_country_section_rows(cls, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, str]]: return [ { "status": cls._normalize_option_text(row.get("status") or ""), "quantity": cls._normalize_quantity_text(row.get("quantity")), "deleteQuantity": "", "processStatus": "", } for row in rows if cls._normalize_option_text(row.get("status") or "") ] @classmethod def _build_deletable_status_rows( cls, rows: Iterable[Dict[str, Any]], delete_conditions: Optional[Iterable[Any]] = None, ) -> List[Dict[str, Any]]: condition_texts = cls._normalize_delete_condition_texts(delete_conditions) if not condition_texts: return [] inventory_status_enabled = cls._matches_any_delete_condition( ("库存状态删除", "商品状态删除", "库存状态"), condition_texts, ) candidates = [] for row in rows: status = cls._normalize_option_text(row.get("status") or "") if not status: continue 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 not quantity_text or quantity <= 0 or canonical_status in LISTING_STATUS_DELETE_WHITELIST: continue if not inventory_status_enabled and not cls._matches_delete_conditions(status, condition_texts): continue candidates.append( { "status": status, "canonicalStatus": canonical_status, "quantity": quantity, } ) return candidates @classmethod def _normalize_delete_condition_texts( cls, delete_conditions: Optional[Iterable[Any]], ) -> List[str]: texts = [] if delete_conditions is None: return texts if isinstance(delete_conditions, (str, bytes, dict)): delete_conditions = [delete_conditions] for condition in delete_conditions or []: if isinstance(condition, dict): raw_text = next( ( condition.get(key) for key in ("conditionText", "condition_text", "status", "text", "label", "name") if condition.get(key) ), "", ) else: raw_text = condition for part in re.split(r"[,，;；、\r\n]+", str(raw_text or "")): text = cls._normalize_option_text(part).lower() if text: texts.append(text) return list(dict.fromkeys(texts)) @classmethod def _matches_delete_conditions(cls, status: str, condition_texts: List[str]) -> bool: normalized_status = cls._normalize_option_text(status).lower() canonical_status = cls._canonical_listing_status(status).lower() return any( condition in normalized_status or condition in canonical_status or normalized_status in condition or canonical_status in condition for condition in condition_texts ) @classmethod def _matches_any_delete_condition(cls, statuses: Iterable[str], condition_texts: List[str]) -> bool: return any(cls._matches_delete_conditions(status, condition_texts) for status in statuses) @classmethod def _format_delete_conditions(cls, delete_conditions: Optional[Iterable[Any]]) -> str: texts = cls._normalize_delete_condition_texts(delete_conditions) return ", ".join(texts) if texts else "未指定" @classmethod def _build_deletable_complete_draft_tag_rows( cls, rows: Iterable[Dict[str, Any]], delete_conditions: Optional[Iterable[Any]] = None, ) -> List[Dict[str, Any]]: condition_texts = cls._normalize_delete_condition_texts(delete_conditions) if not condition_texts: return [] complete_draft_enabled = bool( cls._matches_any_delete_condition( ("补全草稿删除", "补全草稿", "商品信息草稿"), condition_texts, ) ) candidates = [] for row in rows: tag = cls._normalize_option_text(row.get("tag") or row.get("status") or "") if not tag: continue quantity_text = cls._normalize_quantity_text(row.get("quantity")) quantity = int(quantity_text) if quantity_text else 0 canonical_tag = cls._canonical_listing_status(tag) if not quantity_text or quantity <= 0 or canonical_tag in LISTING_STATUS_DELETE_WHITELIST: continue if not complete_draft_enabled and not cls._matches_delete_conditions(tag, condition_texts): continue candidates.append( { "tag": tag, "canonicalTag": canonical_tag, "quantity": quantity, } ) return candidates @staticmethod 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: normalized = cls._normalize_option_text(status) return LISTING_STATUS_COMPATIBILITY_MAP.get(normalized, normalized) @classmethod def _normalize_allowed_status_set(cls, statuses: Optional[Iterable[str]]) -> Optional[Set[str]]: if statuses is None: return None allowed = set() for status in statuses: text = cls._normalize_option_text(str(status or "")) if not text: continue for part in re.split(r"[,，;；、\r\n]+", text): normalized = cls._normalize_option_text(part) if normalized: allowed.add(cls._canonical_listing_status(normalized)) return allowed @staticmethod def _normalize_option_text(text: str) -> str: return re.sub(r"\s+", " ", text).strip() @staticmethod def _parse_status_text(raw_text: str) -> tuple[str, Optional[int]]: text = re.sub(r"\s+", " ", raw_text).strip() if not text: return "", None match = re.search(r"^(.*?)\s*[（(]\s*([\d,]+)\s*[）)]\s*$", text) if match: return match.group(1).strip(), int(match.group(2).replace(",", "")) match = re.search(r"^(.*?)\s+([\d,]+)\s*$", text) if match and not re.search(r"\d", match.group(1)): return match.group(1).strip(), int(match.group(2).replace(",", "")) return text, None @classmethod def _parse_quick_view_tag_rows( cls, rows: Iterable[Dict[str, Any]], shop_name: str, country: str ) -> List[Dict[str, Any]]: parsed_rows = [] seen = set() for row in rows: raw_text = cls._normalize_option_text(row.get("raw_text") or "") label_text = cls._normalize_option_text(row.get("label_text") or "") quantity_text = cls._normalize_option_text(str(row.get("quantity_text") or "")) parsed = cls._parse_label_quantity_text(raw_text) if parsed is None and label_text and quantity_text: parsed = cls._parse_label_quantity_text(f"{label_text} ({quantity_text})") if parsed is None: continue tag, quantity = parsed if not cls._is_complete_draft_tag_candidate(tag): continue key = (tag, quantity) if key in seen: continue seen.add(key) parsed_rows.append( { "shop_name": shop_name, "country": country, "tag": tag, "quantity": quantity, "raw_text": raw_text or f"{label_text} ({quantity_text})", } ) return parsed_rows @staticmethod def _is_complete_draft_tag_candidate(tag: str) -> bool: normalized_tag = re.sub(r"\s+", " ", tag).strip() if not normalized_tag: return False if normalized_tag == "补全草稿": return False if "/" in normalized_tag: return False if re.search(r"\b页面\b", normalized_tag): return False return True @staticmethod def _parse_label_quantity_text(raw_text: str) -> Optional[tuple[str, int]]: text = re.sub(r"\s+", " ", raw_text).strip() if not text: return None match = re.search(r"^(.*?)\s*[（(]\s*([\d,]+)\s*[）)]\s*$", text) if not match: return None tag = match.group(1).strip() if not tag or re.fullmatch(r"[\d,]+", tag): return None if re.search(r"[（(]\s*[\d,]+\s*[）)]", tag): return None return tag, int(match.group(2).replace(",", "")) @classmethod def _parse_cart_ratio_rows(cls, rows: Iterable[Dict[str, Any]], country: str) -> Dict[str, str]: ratio = cls._empty_cart_ratio(country) texts = cls._cart_ratio_texts(rows) for text in texts: if not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = cls._extract_ratio_for_days(text, 2) if not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = cls._extract_ratio_for_days(text, 30) if ratio["ratio2DaysAgo"] and ratio["ratio30DaysAgo"]: return ratio combined_text = " ".join(texts) if not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = cls._extract_ratio_for_days(combined_text, 2) if not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = cls._extract_ratio_for_days(combined_text, 30) return ratio @classmethod def _parse_all_cart_ratio_rows(cls, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, str]]: rect_rows = cls._parse_rect_based_cart_ratio_rows(rows) if rect_rows: return rect_rows direct_rows = cls._parse_direct_cart_ratio_rows(rows) if direct_rows: return direct_rows texts = cls._cart_ratio_texts(rows) combined_text = cls._normalize_option_text(" ".join(texts)) return [cls._parse_country_cart_ratio(texts, combined_text, country) for country in RECOMMENDED_OFFER_COUNTRIES] @classmethod def _parse_rect_based_cart_ratio_rows(cls, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, str]]: countries = [] percentages = [] for row in rows: attrs = row.get("attrs") or {} candidate = str( attrs.get("title") or row.get("label_text") or row.get("raw_text") or "" ).strip() if not candidate or len(candidate) > 20: continue left = row.get("rect_left") top = row.get("rect_top") width = row.get("rect_width") height = row.get("rect_height") if not all(isinstance(value, (int, float)) for value in (left, top, width, height)): continue if width <= 0 or height <= 0: continue country_match = next( ( target for target in RECOMMENDED_OFFER_COUNTRIES if any(candidate.lower() == alias.lower() for alias in cls._country_aliases(target)) ), None, ) if country_match: countries.append( { "country": country_match, "x": float(left), "y": float(top) + float(height) / 2, "h": float(height), "area": float(width) * float(height), } ) continue if re.fullmatch(r"\d+(?:\.\d+)?\s*%", candidate): percentages.append( { "value": cls._normalize_ratio_text(candidate), "x": float(left), "y": float(top) + float(height) / 2, "area": float(width) * float(height), } ) if not countries or not percentages: return [] deduped_countries = [] for item in sorted(countries, key=lambda row: (row["y"], row["x"], row["area"])): exists = any( row["country"] == item["country"] and abs(row["y"] - item["y"]) < 8 for row in deduped_countries ) if not exists: deduped_countries.append(item) parsed = [] for row in deduped_countries: row_y_tolerance = max(12.0, min(24.0, row["h"] / 2 + 4.0)) same_row = sorted( ( item for item in percentages if item["x"] > row["x"] + 20 and abs(item["y"] - row["y"]) <= row_y_tolerance ), key=lambda item: (item["x"], item["y"], item["area"]), ) unique = [] for item in same_row: if not any(abs(existing["x"] - item["x"]) < 24 for existing in unique): unique.append(item) parsed.append( { "country": row["country"], "ratio2DaysAgo": unique[0]["value"] if len(unique) >= 1 else "", "ratio30DaysAgo": unique[1]["value"] if len(unique) >= 2 else "", } ) if not any(item["ratio2DaysAgo"] or item["ratio30DaysAgo"] for item in parsed): return [] by_country = {item["country"]: item for item in parsed} return [by_country.get(country, cls._empty_cart_ratio(country)) for country in RECOMMENDED_OFFER_COUNTRIES] @classmethod def _parse_direct_cart_ratio_rows(cls, rows: Iterable[Dict[str, Any]]) -> List[Dict[str, str]]: by_country: Dict[str, Dict[str, str]] = {} for row in rows: country = str(row.get("country") or "").strip() if country not in RECOMMENDED_OFFER_COUNTRIES: continue ratio = by_country.setdefault(country, cls._empty_cart_ratio(country)) ratio2 = cls._normalize_ratio_text(str(row.get("ratio2DaysAgo") or "")) if row.get("ratio2DaysAgo") else "" ratio30 = cls._normalize_ratio_text(str(row.get("ratio30DaysAgo") or "")) if row.get("ratio30DaysAgo") else "" if ratio2 and not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = ratio2 if ratio30 and not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = ratio30 if not by_country: return [] return [by_country.get(country, cls._empty_cart_ratio(country)) for country in RECOMMENDED_OFFER_COUNTRIES] @classmethod def _parse_country_cart_ratio(cls, texts: List[str], combined_text: str, country: str) -> Dict[str, str]: ratio = cls._empty_cart_ratio(country) for text in texts: segment = cls._extract_country_segment(text, country) if not segment: continue if not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = cls._extract_ratio_for_days(segment, 2) if not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = cls._extract_ratio_for_days(segment, 30) if ratio["ratio2DaysAgo"] and ratio["ratio30DaysAgo"]: return ratio segment = cls._extract_country_segment(combined_text, country) if segment: if not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = cls._extract_ratio_for_days(segment, 2) if not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = cls._extract_ratio_for_days(segment, 30) if not ratio["ratio2DaysAgo"] or not ratio["ratio30DaysAgo"]: header_ratio = cls._extract_ratio_pair_from_country_table(segment, combined_text) if not ratio["ratio2DaysAgo"]: ratio["ratio2DaysAgo"] = header_ratio["ratio2DaysAgo"] if not ratio["ratio30DaysAgo"]: ratio["ratio30DaysAgo"] = header_ratio["ratio30DaysAgo"] return ratio @classmethod def _extract_country_segment(cls, text: str, country: str) -> str: normalized = cls._normalize_option_text(text) if not normalized: return "" matches = sorted( ( (alias, match.start(), match.end()) for alias in cls._country_aliases(country) for match in re.finditer(re.escape(alias), normalized, flags=re.IGNORECASE) ), key=lambda item: item[1], ) if not matches: return "" all_boundaries = sorted( ( (match.start(), match.end()) for target_country in RECOMMENDED_OFFER_COUNTRIES for alias in cls._country_aliases(target_country) for match in re.finditer(re.escape(alias), normalized, flags=re.IGNORECASE) ), key=lambda item: item[0], ) for _, start, end in matches: next_boundary = next((boundary_start for boundary_start, _ in all_boundaries if boundary_start > start), len(normalized)) segment = normalized[start:next_boundary].strip() if cls._extract_ratio_for_days(segment, 2) or cls._extract_ratio_for_days(segment, 30): return segment trailing_segment = normalized[end:next_boundary].strip() if cls._extract_ratio_for_days(trailing_segment, 2) or cls._extract_ratio_for_days(trailing_segment, 30): return trailing_segment if re.search(r"\d+(?:\.\d+)?\s*%", segment): return segment if re.search(r"\d+(?:\.\d+)?\s*%", trailing_segment): return trailing_segment return "" @staticmethod def _country_aliases(country: str) -> List[str]: aliases = { "欧洲": ["欧洲", "Europe"], "英国": ["英国", "UK", "United Kingdom"], "德国": ["德国", "Germany"], "法国": ["法国", "France"], "西班牙": ["西班牙", "Spain"], "意大利": ["意大利", "Italy"], } return aliases.get(country, [country]) @classmethod def _has_any_cart_ratio(cls, rows: Iterable[Dict[str, Any]]) -> bool: return any( item["ratio2DaysAgo"] or item["ratio30DaysAgo"] for item in cls._parse_all_cart_ratio_rows(rows) ) @classmethod def _has_complete_cart_ratio_rows(cls, rows: Iterable[Dict[str, Any]]) -> bool: return all( item["ratio2DaysAgo"] and item["ratio30DaysAgo"] for item in cls._parse_all_cart_ratio_rows(rows) ) @classmethod def _extract_ratio_pair_from_country_table(cls, segment: str, full_text: str) -> Dict[str, str]: ratio = {"ratio2DaysAgo": "", "ratio30DaysAgo": ""} if not cls._looks_like_cart_ratio_table(full_text): return ratio percentages = [cls._normalize_ratio_text(match.group(0)) for match in re.finditer(r"\d+(?:\.\d+)?\s*%", segment)] if len(percentages) >= 1: ratio["ratio2DaysAgo"] = percentages[0] if len(percentages) >= 2: ratio["ratio30DaysAgo"] = percentages[1] return ratio @classmethod def _looks_like_cart_ratio_table(cls, text: str) -> bool: normalized = cls._normalize_option_text(text) if not normalized: return False if cls._extract_ratio_for_days(normalized, 2) and cls._extract_ratio_for_days(normalized, 30): return True if not re.search(r"推荐报价|featured offer|buy box", normalized, flags=re.IGNORECASE): return False country_hits = sum( 1 for country in RECOMMENDED_OFFER_COUNTRIES if any(alias.lower() in normalized.lower() for alias in cls._country_aliases(country)) ) if country_hits >= 2: return True date_patterns = [ r"\b\d{1,2}\s+[A-Za-z]{3,9}\b", r"\b[A-Za-z]{3,9}\s+\d{1,2}\b", r"\b\d{1,2}[/-]\d{1,2}(?:[/-]\d{2,4})?\b", r"\b\d{4}[/-]\d{1,2}[/-]\d{1,2}\b", ] return any(re.search(pattern, normalized, flags=re.IGNORECASE) for pattern in date_patterns) @staticmethod def _empty_cart_ratio(country: str) -> Dict[str, str]: return {"country": country, "ratio2DaysAgo": "", "ratio30DaysAgo": ""} @classmethod def _cart_ratio_texts(cls, rows: Iterable[Dict[str, Any]]) -> List[str]: texts = [] seen = set() for row in rows: parts = [ str(row.get("raw_text") or ""), str(row.get("label_text") or ""), str(row.get("percentage_text") or ""), ] text = cls._normalize_option_text(" ".join(part for part in parts if part)) if not text or text in seen: continue seen.add(text) texts.append(text) return texts @classmethod def _extract_ratio_for_days(cls, raw_text: str, days: int) -> str: text = cls._normalize_option_text(raw_text) if not text: return "" day_matches = list(cls._iter_day_matches(text, days)) if not day_matches: return "" percent_matches = list(re.finditer(r"\d+(?:\.\d+)?\s*%", text)) if not percent_matches: return "" all_day_matches = sorted( [match for target_days in (2, 30) for match in cls._iter_day_matches(text, target_days)], key=lambda match: match.start(), ) prefer_before = percent_matches[0].start() < all_day_matches[0].start() if all_day_matches else False for day_match in day_matches: next_day_start = next( (match.start() for match in all_day_matches if match.start() > day_match.start()), len(text), ) previous_day_end = max( [match.end() for match in all_day_matches if match.end() < day_match.start()], default=0, ) before = [match for match in percent_matches if previous_day_end <= match.end() <= day_match.start()] after = [match for match in percent_matches if day_match.end() <= match.start() < next_day_start] nearest_before = before[-1] if before else None nearest_after = after[0] if after else None if nearest_before and nearest_after: if prefer_before: return cls._normalize_ratio_text(nearest_before.group(0)) return cls._normalize_ratio_text(nearest_after.group(0)) if nearest_before: return cls._normalize_ratio_text(nearest_before.group(0)) if nearest_after: return cls._normalize_ratio_text(nearest_after.group(0)) return "" @staticmethod def _iter_day_matches(text: str, days: int) -> Iterable[re.Match[str]]: patterns = [ rf"(? str: return re.sub(r"\s+", "", text) class PatrolDeleteTask: """巡店删除任务处理类。""" mark_name = "巡店删除" def __init__(self, user_info: dict = None): self.user_info = user_info or {} self.running = True def log(self, message: str, level: str = "INFO"): level_name = str(level or "INFO").upper() logger.log(level_name, f"[PatrolDeleteTask] {_safe_log_text(message)}") def process_task(self, task_data: dict): """处理巡店删除任务主入口。""" task_id = None try: data = task_data.get("data", {}) task_id = data.get("taskId") items = data.get("items", []) country_sections = data.get("country_sections", []) cart_ratios = data.get("cart_ratios", []) delete_conditions = self._resolve_delete_condition_texts(data) if not task_id: self.log("任务ID为空，跳过", "WARNING") return if not items: self.log("店铺列表为空，跳过", "WARNING") return from config import runing_task total_countries = sum(len(self._resolve_country_sections(item, country_sections)) for item in items) runing_task[task_id] = { "status": "running", "start_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "total_shops": len(items), "total_countries": total_countries, "processed_shops": 0, "processed_countries": 0, "success_count": 0, "failed_count": 0, "failed_countries": 0, "deleted_listings_count": 0, "delete_conditions": delete_conditions, "stop_requested": False, } self.log( f"开始处理巡店删除任务 {task_id}，共 {len(items)} 个店铺，{total_countries} 个国家，" f"删除条件={delete_conditions if delete_conditions is not None else '未指定'}" ) for index, shop_item in enumerate(items, 1): if runing_task[task_id].get("stop_requested", False): self.log(f"检测到任务 {task_id} 的暂停请求，停止处理", "WARNING") runing_task[task_id]["status"] = "stopped" return shop_name = shop_item.get("shopName", "未知店铺") self.log(f"[{index}/{len(items)}] 开始处理店铺: {shop_name}") self.process_shop(task_id, shop_item, country_sections, cart_ratios, delete_conditions) runing_task[task_id]["processed_shops"] += 1 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 exc: 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(exc) def process_shop( self, task_id: int, shop_item: Dict[str, Any], country_sections: List[Dict[str, Any]], cart_ratios: List[Dict[str, Any]], delete_conditions: Optional[List[str]] = None, ): """处理单个店铺。""" from config import runing_shop, runing_task shop_name = shop_item.get("shopName", "") company_name = shop_item.get("companyName", "") if not shop_name: self.log("店铺名称为空，跳过", "WARNING") return if not company_name: raise RuntimeError(f"店铺 {shop_name} 的公司名称为空") runing_task[task_id]["current_shop"] = shop_name runing_shop[shop_name] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.log(f"店铺 {shop_name} 已添加到执行列表") try: payload_country_sections = self._resolve_country_sections(shop_item, country_sections) payload_cart_ratios = self._resolve_cart_ratios(shop_item, cart_ratios) if not payload_country_sections: raise RuntimeError(f"店铺 {shop_name} 缺少国家模板，无法执行巡店删除") 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)}，" f"删除条件={delete_conditions if delete_conditions is not None else '未指定'}" ) driver: Optional[InventoryManage] = None requires_browser_reset = False country_results = [] for country_name in country_names: if runing_task[task_id].get("stop_requested", False): self.log(f"检测到任务 {task_id} 的暂停请求，停止处理店铺 {shop_name}", "WARNING") break if driver is None or requires_browser_reset: if driver is not None: self._safe_close_store(driver) driver = self.open_shop( max_retries=PRODUCT_TASK_MAX_RETRIES, company_name=company_name, shop_name=shop_name, iskill=requires_browser_reset, ) requires_browser_reset = False if driver is None: raise RuntimeError(f"店铺 {shop_name} 打开失败") country_result = self.process_country( driver=driver, task_id=task_id, shop_name=shop_name, country_name=country_name, delete_conditions=delete_conditions, ) country_results.append(country_result) aggregated_sections[country_name] = country_result["countrySection"] aggregated_cart_ratios[country_name] = country_result["cartRatio"] runing_task[task_id]["processed_countries"] += 1 runing_task[task_id]["deleted_listings_count"] += country_result["deletedCount"] 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}" ) self._post_progress_result_safely( task_id=task_id, shop_name=shop_name, ) 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, country_sections=[aggregated_sections[country] for country in country_names], cart_ratios=[aggregated_cart_ratios[country] for country in country_names], shop_done=True, ) if any(result["success"] for result in country_results): 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 try: self.post_result(task_id=task_id, shop_name=shop_name, error=str(exc), shop_done=True) except Exception as post_exc: self.log(f"回传店铺失败结果失败: {str(post_exc)}", "ERROR") finally: if shop_name in runing_shop: del runing_shop[shop_name] self.log(f"店铺 {shop_name} 已从执行列表中移除") def _post_progress_result_safely( self, task_id: int, shop_name: str, ) -> None: """Submit a heartbeat-only payload so Java does not append duplicate result rows.""" try: self.post_result( task_id=task_id, shop_name=shop_name, shop_done=False, heartbeat_only=True, ) except Exception as exc: self.log(f"店铺 {shop_name} 中间结果回传失败，继续执行: {str(exc)}", "WARNING") def open_shop(self, max_retries: int, company_name: str, shop_name: str, iskill: bool = False) -> Optional[InventoryManage]: error_info = "" driver: Optional[InventoryManage] = None browser = None for retry in range(max_retries): try: self.log(f"尝试打开店铺 {shop_name} (第 {retry + 1}/{max_retries} 次)") if iskill: self.log("重试前先杀掉浏览器进程...") kill_process("v6") kill_process("v5") time.sleep(PRODUCT_TASK_BROWSER_RESET_DELAY) user_info = {**self.user_info, "company": company_name} driver = InventoryManage(user_info) browser = driver.open_shop(shop_name) if not browser or browser == "店铺不存在": self.log(f"打开店铺失败: {browser}", "WARNING") driver = None continue if driver.need_login(): self.log(f"店铺 {shop_name} 需要登录，正在登录...") response = get_shop_info(shop_name) credential_payload = response.json() if response is not None else {} password = str(((credential_payload.get("data") or {}).get("password") or "")).strip() if not password: message = f"获取店铺 {shop_name} 凭证失败" self.log(message, "ERROR") show_notification(message, "error") driver = None continue if not driver.login(password): self.log(f"店铺 {shop_name} 登录失败", "WARNING") driver = None continue browser = driver.open_shop(shop_name) if not browser or browser == "店铺不存在": self.log(f"登录后打开店铺失败: {browser}", "WARNING") driver = None continue self.log(f"成功打开店铺 {shop_name}") return driver except Exception as exc: error_info = str(exc) self.log(f"打开店铺异常: {error_info}", "ERROR") driver = None if retry < max_retries - 1: time.sleep(PRODUCT_TASK_RETRY_DELAY) if retry < max_retries - 1: time.sleep(PRODUCT_TASK_OPEN_SHOP_DELAY) self.log(f"店铺 {shop_name} 打开失败，已重试 {max_retries} 次，错误: {error_info}", "ERROR") return driver def process_country( self, driver: InventoryManage, task_id: int, shop_name: str, country_name: str, delete_conditions: Optional[List[str]] = None, ) -> Dict[str, Any]: from config import runing_task trace_id = _new_trace_id("country") self.log(f"trace_id={trace_id} 开始处理国家: task_id={task_id}, shop={shop_name}, country={country_name}") show_notification(f"开始处理巡店删除国家: {country_name}", "info") if task_id in runing_task: runing_task[task_id]["current_country"] = country_name latest_status_rows: List[Dict[str, Any]] = [] try: self.log(f"trace_id={trace_id} 准备切换国家: country={country_name}") self._switch_country_with_retry(driver, country_name) self.log(f"trace_id={trace_id} 国家切换完成，准备进入管理所有库存页面: country={country_name}") driver.switch_to_manage_inventory() self.log(f"trace_id={trace_id} 已进入管理所有库存页面，开始删除非白名单商品: country={country_name}") delete_result = driver.delete_all_listings_from_non_whitelisted_statuses( shop_name, country_name, trace_id=trace_id, delete_conditions=delete_conditions, ) latest_status_rows = delete_result.get("statusRows") or [] self.log( f"trace_id={trace_id} 国家 {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_delete_result = self._delete_complete_draft_country_section_safe( driver, shop_name, country_name, delete_conditions, ) complete_draft_section = complete_draft_delete_result.get("countrySection") country_section = self._replace_complete_draft_rows(country_section, complete_draft_section) cart_ratio = self._read_country_cart_ratio_safe(driver, country_name) self.log(f"trace_id={trace_id} 国家 {country_name} 购物车比例读取完成: {cart_ratio.get('ratio', '')}") total_deleted = int(delete_result.get("totalDeleted") or 0) + int( complete_draft_delete_result.get("deletedCount") or 0 ) return { "success": True, "countrySection": country_section, "cartRatio": cart_ratio, "deletedCount": total_deleted, "reopenRequired": False, } except Exception as exc: self.log(f"trace_id={trace_id} 处理国家 {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, status_rows=failure_rows, delete_counts={}, default_process_status=f"处理失败: {self._short_error(str(exc))}", ) return { "success": False, "countrySection": country_section, "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]: trace_id = _new_trace_id("cart") try: self.log(f"trace_id={trace_id} 开始读取国家 {country_name} 购物车比例") cart_ratio = self._extract_country_cart_ratio( driver.get_recommended_offer_cart_ratios().get("cartRatios") or [], country_name, ) self.log(f"trace_id={trace_id} 国家 {country_name} 购物车比例读取成功: {cart_ratio.get('ratio', '')}") return cart_ratio except Exception as cart_exc: self.log(f"trace_id={trace_id} 读取国家 {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]]: trace_id = _new_trace_id("draft") try: self.log(f"trace_id={trace_id} 开始读取国家 {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", "status": PatrolDeleteTask._string_result_field(item.get("tag")), "quantity": quantity, "deleteQuantity": "", "processStatus": "", } ) self.log(f"trace_id={trace_id} 国家 {country_name} 补全草稿读取完成: rows={rows}") return {"country": country_name, "rows": rows} if rows else None except Exception as draft_exc: self.log(f"trace_id={trace_id} 读取国家 {country_name} 补全草稿失败: {str(draft_exc)}", "WARNING") return None def _delete_complete_draft_country_section_safe( self, driver: InventoryManage, shop_name: str, country_name: str, delete_conditions: Optional[List[str]] = None, ) -> Dict[str, Any]: trace_id = _new_trace_id("draft") try: self.log(f"trace_id={trace_id} 开始处理国家 {country_name} 补全草稿删除") delete_result = driver.delete_complete_drafts_from_non_whitelisted_tags( shop_name, country_name, trace_id=trace_id, delete_conditions=delete_conditions, ) rows = [] delete_counts = delete_result.get("deleteCounts") or {} for item in delete_result.get("draftRows") or []: status = PatrolDeleteTask._string_result_field(item.get("tag") or item.get("status")) canonical_status = InventoryManage._canonical_listing_status(status) deleted_count = int(delete_counts.get(canonical_status) or 0) rows.append( { "type": "completeDraft", "status": status, "quantity": InventoryManage._normalize_quantity_text(item.get("quantity")), "deleteQuantity": str(deleted_count), "processStatus": "已完成" if deleted_count > 0 else "", } ) section = {"country": country_name, "rows": rows} if rows else None self.log( f"trace_id={trace_id} 国家 {country_name} 补全草稿删除处理完成: " f"deleted={delete_result.get('totalDeleted', 0)}, rows={rows}" ) return { "countrySection": section, "deletedCount": int(delete_result.get("totalDeleted") or 0), "deleteResult": delete_result, } except Exception as draft_exc: self.log(f"trace_id={trace_id} 删除国家 {country_name} 补全草稿失败: {str(draft_exc)}", "WARNING") fallback_section = self._read_complete_draft_country_section_safe(driver, shop_name, country_name) return {"countrySection": fallback_section, "deletedCount": 0, "error": str(draft_exc)} def _switch_country_with_retry(self, driver: InventoryManage, country_name: str) -> None: trace_id = _new_trace_id("switch") for retry in range(PRODUCT_TASK_MAX_RETRIES): try: self.log( f"trace_id={trace_id} 尝试切换到国家 {country_name} " f"(第 {retry + 1}/{PRODUCT_TASK_MAX_RETRIES} 次)" ) if driver.switch_to_country(country_name): self.log(f"trace_id={trace_id} 成功切换到国家 {country_name}") return if retry < PRODUCT_TASK_MAX_RETRIES - 1: try: driver.tab.refresh() except Exception: pass time.sleep(PRODUCT_TASK_RETRY_DELAY) except Exception as exc: if retry >= PRODUCT_TASK_MAX_RETRIES - 1: raise RuntimeError(f"切换国家 {country_name} 失败: {exc}") from exc self.log(f"trace_id={trace_id} 切换国家 {country_name} 异常，准备重试: {str(exc)}", "WARNING") time.sleep(PRODUCT_TASK_RETRY_DELAY) raise RuntimeError(f"切换国家 {country_name} 失败") @classmethod def _resolve_delete_condition_texts(cls, data: Dict[str, Any]) -> Optional[List[str]]: if "delete_conditions" in data: raw_conditions = data.get("delete_conditions") elif "deleteConditions" in data: raw_conditions = data.get("deleteConditions") else: return [] return cls._parse_delete_condition_texts(raw_conditions) @classmethod def _parse_delete_condition_texts(cls, raw_conditions: Any) -> List[str]: if raw_conditions is None: return [] if not isinstance(raw_conditions, list): raw_conditions = [raw_conditions] texts: List[str] = [] for item in raw_conditions: text = "" if isinstance(item, dict): for key in ("conditionText", "condition_text", "status", "text", "label", "name"): value = item.get(key) if value: text = str(value) break elif item is not None: text = str(item) for part in re.split(r"[,，;；、\r\n]+", text): normalized = InventoryManage._normalize_option_text(part) if normalized and normalized not in texts: texts.append(normalized) return texts def post_result( self, task_id: int, shop_name: str, country_sections: Optional[List[Dict[str, Any]]] = None, cart_ratios: Optional[List[Dict[str, Any]]] = None, error: str = "", shop_done: bool = False, heartbeat_only: bool = False, ): """回传巡店删除结果到 Java API。""" from config import DELETE_BRAND_API_BASE trace_id = _new_trace_id("post") shop_payload: Dict[str, Any] = { "shopName": shop_name, "shopDone": shop_done, "submissionId": self._build_submission_id(task_id, shop_name), "chunkIndex": 1, "chunkTotal": 1, } if error: shop_payload["error"] = error elif heartbeat_only: shop_payload["heartbeatOnly"] = True else: 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" self._log_result_payload_summary(task_id, shop_payload, trace_id=trace_id) max_retries = 3 for retry in range(max_retries): try: self.log(f"trace_id={trace_id} 尝试回传巡店删除结果 (第 {retry + 1}/{max_retries} 次)") self.log(f"trace_id={trace_id} 回传URL: {url}") self.log(f"trace_id={trace_id} 回传数据: {_safe_log_text(payload)}") response = requests.post( url, json=payload, headers={"Content-Type": "application/json"}, timeout=30, verify=False, ) self.log(f"trace_id={trace_id} 回传结果: {_safe_log_text(response.text)}") 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) self.log( f"trace_id={trace_id} 巡店删除结果回传响应: status_code={response.status_code}, " f"api_success={api_success}, response={_safe_log_text(response.text)}" ) if response.status_code == 200 and api_success: self.log(f"trace_id={trace_id} 巡店删除结果回传成功: {shop_name}") return self.log( f"trace_id={trace_id} 巡店删除结果回传失败，状态码: {response.status_code}, " f"响应: {_safe_log_text(response.text)}", "WARNING", ) except Exception as exc: self.log(f"trace_id={trace_id} 调用巡店删除结果API异常: {str(exc)}", "ERROR") if retry < max_retries - 1: time.sleep(2) raise RuntimeError("巡店删除结果回传最终失败") def _log_result_payload_summary(self, task_id: int, shop_payload: Dict[str, Any], trace_id: str = "") -> None: trace_id = trace_id or _new_trace_id("post") country_sections = shop_payload.get("countrySections") or [] cart_ratios = shop_payload.get("cartRatios") or [] countries = [str(section.get("country") or "").strip() for section in country_sections] row_count = sum(len(section.get("rows") or []) for section in country_sections) deleted_total = self._sum_result_delete_quantity(country_sections) self.log( f"trace_id={trace_id} 巡店删除结果回传摘要: " f"task_id={task_id}, shop={shop_payload.get('shopName', '')}, " f"shopDone={shop_payload.get('shopDone', False)}, " f"hasError={bool(shop_payload.get('error'))}, " f"countryCount={len(country_sections)}, countries={','.join(countries)}, " f"rowCount={row_count}, cartRatioCount={len(cart_ratios)}, deletedTotal={deleted_total}" ) @staticmethod def _sum_result_delete_quantity(country_sections: List[Dict[str, Any]]) -> int: total = 0 for section in country_sections or []: for row in section.get("rows") or []: text = str(row.get("deleteQuantity") or "").replace(",", "").strip() if not text: continue match = re.search(r"\d+", text) if match: total += int(match.group(0)) return total @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 []: if cls._is_complete_draft_result_row(row): delete_quantity = cls._string_result_field(row.get("deleteQuantity")) normalized_row = { "status": cls._complete_draft_result_status(row), "quantity": cls._string_result_field(row.get("quantity")), "deleteQuantity": delete_quantity, "processStatus": cls._complete_draft_process_status(row, delete_quantity), } else: normalized_row = { "status": cls._string_result_field(row.get("status")), "quantity": cls._string_result_field(row.get("quantity")), "deleteQuantity": cls._string_result_field(row.get("deleteQuantity")), "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" @staticmethod def _complete_draft_result_status(row: Dict[str, Any]) -> str: status = PatrolDeleteTask._string_result_field(row.get("status") or row.get("tag")).strip() return status or "商品信息草稿" @staticmethod def _complete_draft_process_status(row: Dict[str, Any], delete_quantity: str) -> str: process_status = PatrolDeleteTask._string_result_field(row.get("processStatus")).strip() if process_status: return process_status match = re.search(r"\d+", str(delete_quantity or "")) return "已完成" if match and int(match.group(0)) > 0 else "" @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]] ) -> List[Dict[str, Any]]: return deepcopy(country_sections or shop_item.get("countrySections") or []) @staticmethod def _resolve_cart_ratios(shop_item: Dict[str, Any], cart_ratios: List[Dict[str, Any]]) -> List[Dict[str, Any]]: return deepcopy(cart_ratios or shop_item.get("cartRatios") or []) @staticmethod def _build_submission_id(task_id: int, shop_name: str) -> str: return f"patrol-delete:{task_id}:{shop_name}:{int(time.time() * 1000)}" @staticmethod def _ordered_countries(country_sections: List[Dict[str, Any]]) -> List[str]: ordered = [] for section in country_sections: country = str(section.get("country") or "").strip() if country and country not in ordered: ordered.append(country) return ordered @classmethod def _build_country_section_templates(cls, country_sections: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: templates = {} for section in country_sections: country = str(section.get("country") or "").strip() if not country: continue rows = section.get("rows") or [] templates[country] = { "country": country, "rows": [ { "status": str(row.get("status") or ""), "quantity": str(row.get("quantity") or ""), "deleteQuantity": str(row.get("deleteQuantity") or ""), "processStatus": str(row.get("processStatus") or ""), } for row in rows ], } if not templates[country]["rows"]: templates[country]["rows"] = [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}] return templates @classmethod def _build_cart_ratio_templates( cls, country_names: List[str], cart_ratios: List[Dict[str, Any]], ) -> Dict[str, Dict[str, str]]: ratios_by_country: Dict[str, Dict[str, str]] = {} for item in cart_ratios: country = str(item.get("country") or "").strip() if not country: continue ratios_by_country[country] = { "country": country, "ratio": str(item.get("ratio") or ""), } for country in country_names: ratios_by_country.setdefault(country, cls._empty_cart_ratio(country)) return ratios_by_country @classmethod def _build_country_section_result( cls, country_name: str, status_rows: List[Dict[str, Any]], delete_counts: Dict[str, int], default_process_status: str = "", ) -> Dict[str, Any]: rows = [] default_status_applied = False for row in status_rows: status = InventoryManage._normalize_option_text(str(row.get("status") or "")) if not status: continue canonical_status = InventoryManage._canonical_listing_status(status) delete_count = delete_counts.get(canonical_status, 0) process_status = "已完成" if delete_count > 0 else "" if not process_status and default_process_status and not default_status_applied: process_status = default_process_status default_status_applied = True if canonical_status in LISTING_STATUS_DELETE_WHITELIST: delete_quantity = "无需删除" else: delete_quantity = str(delete_count) if delete_count > 0 else "" if default_process_status == "无可删商品" and canonical_status not in LISTING_STATUS_DELETE_WHITELIST: delete_quantity = "0" rows.append( { "status": status, "quantity": InventoryManage._normalize_quantity_text(row.get("quantity")), "deleteQuantity": delete_quantity, "processStatus": process_status, } ) if not rows: rows = [ { "status": "全部", "quantity": "", "deleteQuantity": "无需删除" if default_process_status == "无可删商品" else "", "processStatus": default_process_status, } ] elif default_process_status and not default_status_applied and not any(row["processStatus"] for row in rows): rows[0]["processStatus"] = default_process_status return {"country": country_name, "rows": rows} @classmethod def _extract_country_cart_ratio( cls, cart_ratios: List[Dict[str, Any]], country_name: str, ) -> Dict[str, str]: for item in cart_ratios: if str(item.get("country") or "").strip() != country_name: continue ratio = str(item.get("ratio2DaysAgo") or item.get("ratio") or "").strip() return {"country": country_name, "ratio": ratio} return cls._empty_cart_ratio(country_name) @staticmethod def _empty_cart_ratio(country_name: str) -> Dict[str, str]: return {"country": country_name, "ratio": ""} @staticmethod def _short_error(message: str, limit: int = 80) -> str: compact = re.sub(r"\s+", " ", message or "").strip() if "删除后未读取到成功提示" in compact: return "删除后未读取到成功提示" if any( keyword in compact for keyword in ("与页面的连接已断开", "target frame detached", "disconnected", "connection disconnected") ): return "浏览器连接中断" if any(keyword in compact for keyword in ("stop_requested", "暂停请求", "用户取消", "任务取消", "取消任务", "interrupted")): return "任务已中断" if "切换国家" in compact and "失败" in compact: return "切换国家失败" simplified_errors = ( ("浏览器页面不存在", "浏览器页面不存在"), ("当前页面对象不支持执行 JavaScript", "当前页面无法执行 JavaScript"), ("未找到可用商品状态下拉框", "未找到商品状态下拉框"), ("商品状态下拉框已打开但未读取到选项", "未读取到商品状态下拉选项"), ("未读取到商品状态统计标签", "未读取到商品状态统计标签"), ("未找到商品状态选项", "未找到商品状态选项"), ("商品状态为空", "商品状态为空"), ("全选商品失败", "全选商品失败"), ("点击批量删除失败", "点击批量删除失败"), ("未出现删除确认弹窗", "未出现删除确认弹窗"), ) for keyword, label in simplified_errors: if keyword in compact: return label if len(compact) <= limit: return compact return compact[: limit - 3] + "..." @staticmethod def _should_reopen_browser(message: str) -> bool: return any(keyword in (message or "") for keyword in ("与页面的连接已断开", "浏览器", "target frame detached")) @staticmethod def _safe_close_store(driver: InventoryManage) -> None: try: driver.close_store() except Exception: pass