import json import re import time from copy import deepcopy from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional 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 _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") 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_MESSAGE_PARTS = ( re.compile(r"\d+\s*个商品(?:已删除|已经删除)"), re.compile(r"所[作做]更改需要\s*15\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 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]]: rows = self._wait_listing_status_section_rows() parsed_rows = self._parse_listing_status_section_rows(rows) if not parsed_rows: raise RuntimeError(f"未读取到商品状态统计标签: {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, ) -> 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]] = [] total_deleted = 0 stop_reason = "no-candidates" 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, } 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, ) 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, "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) -> 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("商品状态为空，无法执行筛选") 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}") 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 _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) -> 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 = "" 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) -> 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, ) if not confirm_button: raise RuntimeError(f"状态[{status}]未出现删除确认弹窗") confirm_button.wait.displayed(raise_err=False) confirm_button.wait.enabled() confirm_button.click() 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: 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 = [] 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 "") 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 = [] 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]]) -> List[Dict[str, Any]]: 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 candidates.append( { "status": status, "canonicalStatus": canonical_status, "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) @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"): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] [ProductTask] [{level}] {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", []) 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, "stop_requested": False, } 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") 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) 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]], ): """处理单个店铺。""" 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)}") 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, ) 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}" ) 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 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, ) -> Dict[str, Any]: from config import runing_task self.log(f"开始处理国家: {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._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._read_country_cart_ratio_safe(driver, country_name) self.log(f"国家 {country_name} 购物车比例读取完成: {cart_ratio.get('ratio', '')}") return { "success": True, "countrySection": country_section, "cartRatio": cart_ratio, "deletedCount": delete_result.get("totalDeleted", 0), "reopenRequired": False, } 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, 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]: 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: 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 time.sleep(PRODUCT_TASK_RETRY_DELAY) raise RuntimeError(f"切换国家 {country_name} 失败") 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, ): """回传巡店删除结果到 Java API。""" from config import DELETE_BRAND_API_BASE 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 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" max_retries = 3 for retry in range(max_retries): try: self.log(f"尝试回传巡店删除结果 (第 {retry + 1}/{max_retries} 次)") self.log(f"回传URL: {url}") self.log(f"回传数据: {payload}") response = requests.post( url, json=payload, headers={"Content-Type": "application/json"}, timeout=30, verify=False, ) self.log(f"回传结果: {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) if response.status_code == 200 and api_success: self.log(f"巡店删除结果回传成功: {shop_name}") return self.log( f"巡店删除结果回传失败，状态码: {response.status_code}, 响应: {response.text}", "WARNING", ) except Exception as exc: self.log(f"调用巡店删除结果API异常: {str(exc)}", "ERROR") if retry < max_retries - 1: time.sleep(2) 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]] ) -> 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 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": delete_quantity, "processStatus": process_status, } ) if not rows: rows = [ { "status": "全部", "quantity": "", "deleteQuantity": "0" 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 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