crawler-plugin/app/amazon/patrol_delete.py

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

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")
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"一个或多个商品(?:已删除|已经删除)"),
)
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]]:
        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[List[Dict[str, Any]]] = None,
    ) -> Dict[str, Any]:
        """循环删除非白名单分类商品，直到当前国家没有可删项。"""
        trace_id = trace_id or _new_trace_id("del")
        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(delete_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=delete_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 _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[List[Dict[str, Any]]] = None,
    ) -> List[Dict[str, Any]]:
        condition_texts = cls._normalize_delete_condition_texts(delete_conditions)
        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 condition_texts 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[List[Dict[str, Any]]],
    ) -> List[str]:
        texts = []
        for condition in delete_conditions or []:
            if isinstance(condition, dict):
                raw_text = condition.get("conditionText") or condition.get("condition_text") or condition.get("text")
            else:
                raw_text = condition
            text = cls._normalize_option_text(str(raw_text or "")).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 _format_delete_conditions(cls, delete_conditions: Optional[List[Dict[str, Any]]]) -> str:
        texts = cls._normalize_delete_condition_texts(delete_conditions)
        return ", ".join(texts) if texts else "default"

    @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"(?<!\d){days}\s*天前",
            rf"(?<!\d){days}\s*日前",
            rf"(?<!\d){days}\s*days?\s*ago",
            rf"(?<!\d){days}\s*d\s*ago",
        ]
        for pattern in patterns:
            yield from re.finditer(pattern, text, flags=re.IGNORECASE)

    @staticmethod
    def _normalize_ratio_text(text: str) -> 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 = data.get("delete_conditions") or data.get("deleteConditions") or []

            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"删除条件 {len(delete_conditions)} 条"
            )
            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[Dict[str, Any]]] = 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)}")

            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}"
                )

            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,
        delete_conditions: Optional[List[Dict[str, Any]]] = 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_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"trace_id={trace_id} 国家 {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"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,
                    }
                )
            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 _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} 失败")

    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

        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
        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):
                    normalized_row = {
                        "status": cls._string_result_field(row.get("status")),
                        "quantity": cls._string_result_field(row.get("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"

    @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