1679 lines
70 KiB
Python
1679 lines
70 KiB
Python
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, UserInfo, kill_process
|
||
from amazon.tool import get_shop_info, show_notification
|
||
|
||
|
||
DROPDOWN_OPEN_DELAY = 0.8
|
||
DROPDOWN_OPTIONS_TIMEOUT = 6
|
||
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") / "product"
|
||
|
||
|
||
def _load_product_script(name: str) -> str:
|
||
return (SCRIPT_DIR / name).read_text(encoding="utf-8")
|
||
|
||
|
||
LISTING_STATUS_DROPDOWN_DEBUG_SCRIPT = _load_product_script("listing_status_dropdown_debug.js")
|
||
LISTING_STATUS_DROPDOWN_OPEN_SCRIPT = _load_product_script("listing_status_dropdown_open.js")
|
||
LISTING_STATUS_DROPDOWN_OPTIONS_SCRIPT = _load_product_script("listing_status_dropdown_options.js")
|
||
LISTING_STATUS_CLICK_VISIBLE_STATUS_SCRIPT = _load_product_script("listing_status_click_visible_status.js")
|
||
LISTING_STATUS_SELECT_OPTION_SCRIPT = _load_product_script("listing_status_select_option.js")
|
||
LISTING_STATUS_SECTION_SCRIPT = _load_product_script("listing_status_section.js")
|
||
COMPLETE_DRAFT_QUICK_VIEW_DEBUG_SCRIPT = _load_product_script("complete_draft_quick_view_debug.js")
|
||
COMPLETE_DRAFT_QUICK_VIEW_TAGS_SCRIPT = _load_product_script("complete_draft_quick_view_tags.js")
|
||
COMPLETE_DRAFT_QUICK_VIEW_DIAGNOSTIC_SCRIPT = _load_product_script("complete_draft_quick_view_diagnostic.js")
|
||
RECOMMENDED_OFFER_PERCENTAGE_OPEN_SCRIPT = _load_product_script("recommended_offer_percentage_open.js")
|
||
RECOMMENDED_OFFER_PERCENTAGE_ROWS_SCRIPT = _load_product_script("recommended_offer_percentage_rows.js")
|
||
RECOMMENDED_OFFER_PERCENTAGE_DIAGNOSTIC_SCRIPT = _load_product_script("recommended_offer_percentage_diagnostic.js")
|
||
|
||
LISTING_STATUS_SELECT_DELAY = 0.8
|
||
INVENTORY_ROW_TIMEOUT = 10
|
||
INVENTORY_LOADER_TIMEOUT = 5
|
||
DELETE_SUCCESS_TIMEOUT = 8
|
||
PRODUCT_TASK_MAX_RETRIES = 3
|
||
PRODUCT_TASK_RETRY_DELAY = 2
|
||
PRODUCT_TASK_OPEN_SHOP_DELAY = 3
|
||
PRODUCT_TASK_BROWSER_RESET_DELAY = 2
|
||
PRODUCT_TASK_MAX_UNCHANGED_CANDIDATE_ROUNDS = 3
|
||
DELETE_SUCCESS_MESSAGE_PARTS = (
|
||
re.compile(r"\d+\s*个商品(?:已删除|已经删除)"),
|
||
re.compile(r"所[作做]更改需要\s*15\s*分钟才会显示在商品详情页面上"),
|
||
)
|
||
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"] 出现
|
||
"""
|
||
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)
|
||
|
||
def debug_dump_listing_status_dropdown(self) -> List[Dict[str, Any]]:
|
||
"""打印商品状态下拉框候选 DOM，供现场确认真实页面结构。"""
|
||
self.tab.wait.doc_loaded(raise_err=False)
|
||
candidates = self._run_js(LISTING_STATUS_DROPDOWN_DEBUG_SCRIPT) or []
|
||
logger.info("商品状态下拉框候选 DOM:\n{}", json.dumps(candidates, ensure_ascii=False, indent=2))
|
||
return candidates
|
||
|
||
def open_listing_status_dropdown(self) -> Dict[str, Any]:
|
||
"""只打开实测识别出的商品状态下拉框，不点击任何选项。"""
|
||
self.tab.wait.doc_loaded(raise_err=False)
|
||
result = self._run_js(LISTING_STATUS_DROPDOWN_OPEN_SCRIPT) or {}
|
||
if not result.get("ok"):
|
||
raise RuntimeError(f"未找到商品状态下拉框: {result}")
|
||
time.sleep(DROPDOWN_OPEN_DELAY)
|
||
return result
|
||
|
||
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)
|
||
if not parsed_rows:
|
||
raise RuntimeError(f"商品状态下拉框已打开但未读取到选项: {open_result}")
|
||
return parsed_rows
|
||
|
||
def print_listing_status_json(self, shop_name: str, country: str) -> List[Dict[str, Any]]:
|
||
data = {"countrySections": self.get_listing_status_country_sections(shop_name, country)}
|
||
logger.info("商品状态 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
|
||
return data["countrySections"]
|
||
|
||
def debug_dump_complete_draft_quick_view(self) -> List[Dict[str, Any]]:
|
||
"""打印补全草稿快速查看候选 DOM，供现场确认真实页面结构。"""
|
||
self.tab.wait.doc_loaded(raise_err=False)
|
||
candidates = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_DEBUG_SCRIPT) or []
|
||
logger.info("补全草稿快速查看候选 DOM:\n{}", json.dumps(candidates, ensure_ascii=False, indent=2))
|
||
return candidates
|
||
|
||
def debug_dump_complete_draft_diagnostics(self) -> Dict[str, Any]:
|
||
"""打印补全草稿相关 DOM 和资源 URL，避免输出整页商品列表。"""
|
||
self.tab.wait.doc_loaded(raise_err=False)
|
||
diagnostics = self._run_js(COMPLETE_DRAFT_QUICK_VIEW_DIAGNOSTIC_SCRIPT) or {}
|
||
logger.info("补全草稿快速查看诊断信息:\n{}", json.dumps(diagnostics, ensure_ascii=False, indent=2))
|
||
return diagnostics
|
||
|
||
def open_complete_draft_quick_view(self) -> Dict[str, Any]:
|
||
"""直接打开补全草稿页面。"""
|
||
if self.tab is None:
|
||
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 print_complete_draft_quick_view_json(self, shop_name: str, country: str) -> List[Dict[str, Any]]:
|
||
data = self.get_complete_draft_quick_view_tags(shop_name, country)
|
||
logger.info("补全草稿快速查看标签 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
|
||
return data
|
||
|
||
def open_recommended_offer_percentage_detail(self) -> Dict[str, Any]:
|
||
"""打开 Seller Central 首页中的推荐报价百分比明细。"""
|
||
if self.tab is None:
|
||
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 print_recommended_offer_cart_ratios_json(self) -> Dict[str, List[Dict[str, str]]]:
|
||
data = self.get_recommended_offer_cart_ratios()
|
||
logger.info("推荐报价百分比明细 JSON:\n{}", json.dumps(data, ensure_ascii=False, indent=2))
|
||
return data
|
||
|
||
def debug_dump_recommended_offer_percentage_diagnostics(self) -> Dict[str, Any]:
|
||
diagnostics = self._run_js(RECOMMENDED_OFFER_PERCENTAGE_DIAGNOSTIC_SCRIPT) or {}
|
||
compact = {
|
||
"url": diagnostics.get("url"),
|
||
"title": diagnostics.get("title"),
|
||
"buyBoxFound": diagnostics.get("buyBoxFound"),
|
||
"buyBoxLeafMatches": diagnostics.get("buyBoxLeafMatches") or [],
|
||
}
|
||
logger.info("推荐报价百分比诊断信息:\n{}", json.dumps(compact, ensure_ascii=False, indent=2))
|
||
return diagnostics
|
||
|
||
def _wait_listing_status_option_rows(self) -> List[Dict[str, Any]]:
|
||
deadline = time.time() + DROPDOWN_OPTIONS_TIMEOUT
|
||
rows = []
|
||
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_one_listing_from_non_whitelisted_statuses(
|
||
self, shop_name: str, country: str, max_delete_count: int = 1
|
||
) -> Dict[str, Any]:
|
||
country_sections = self.get_listing_status_country_sections(shop_name, country)
|
||
status_rows = country_sections[0].get("rows") if country_sections else []
|
||
delete_candidates = self._build_deletable_status_rows(status_rows)
|
||
if not delete_candidates:
|
||
raise RuntimeError(f"商品状态中未找到可删除分类: {json.dumps(status_rows, ensure_ascii=False)}")
|
||
|
||
delete_plan = self._limit_delete_candidates(delete_candidates, max_delete_count=max_delete_count)
|
||
if not delete_plan:
|
||
raise RuntimeError(f"删除计划为空: max_delete_count={max_delete_count}")
|
||
|
||
results = []
|
||
for candidate in delete_plan:
|
||
self.select_listing_status(candidate["status"])
|
||
row = self._get_first_inventory_row(candidate["status"])
|
||
target = self._summarize_inventory_row(row)
|
||
success_message = self._delete_inventory_row(row, candidate["status"])
|
||
results.append(
|
||
{
|
||
"status": candidate["status"],
|
||
"canonicalStatus": candidate["canonicalStatus"],
|
||
"statusQuantity": candidate["quantity"],
|
||
"target": target,
|
||
"successMessage": success_message,
|
||
}
|
||
)
|
||
|
||
return {
|
||
"shopName": shop_name,
|
||
"country": country,
|
||
"statusRows": status_rows,
|
||
"deleteCandidates": delete_candidates,
|
||
"deletePlan": delete_plan,
|
||
"results": results,
|
||
}
|
||
|
||
def delete_all_listings_from_non_whitelisted_statuses(
|
||
self,
|
||
shop_name: str,
|
||
country: str,
|
||
max_delete_operations: Optional[int] = None,
|
||
) -> Dict[str, Any]:
|
||
"""循环删除非白名单分类商品，直到当前国家没有可删项。"""
|
||
all_results: List[Dict[str, Any]] = []
|
||
delete_counts: Dict[str, int] = {}
|
||
latest_status_rows: List[Dict[str, str]] = []
|
||
latest_delete_candidates: List[Dict[str, Any]] = []
|
||
previous_signature: Optional[tuple] = None
|
||
unchanged_candidate_rounds = 0
|
||
total_deleted = 0
|
||
stop_reason = "no-candidates"
|
||
|
||
while True:
|
||
country_sections = self.get_listing_status_country_sections(shop_name, country)
|
||
latest_status_rows = deepcopy(country_sections[0].get("rows") if country_sections else [])
|
||
latest_delete_candidates = self._build_deletable_status_rows(latest_status_rows)
|
||
if not latest_delete_candidates:
|
||
stop_reason = "no-candidates"
|
||
break
|
||
|
||
candidate_signature = tuple(
|
||
(item["canonicalStatus"], item["quantity"]) for item in latest_delete_candidates
|
||
)
|
||
if candidate_signature == previous_signature:
|
||
unchanged_candidate_rounds += 1
|
||
if unchanged_candidate_rounds >= PRODUCT_TASK_MAX_UNCHANGED_CANDIDATE_ROUNDS:
|
||
raise RuntimeError(
|
||
f"国家[{country}]删除后商品状态统计未变化，停止重试: "
|
||
f"{json.dumps(latest_delete_candidates, ensure_ascii=False)}"
|
||
)
|
||
else:
|
||
unchanged_candidate_rounds = 0
|
||
previous_signature = candidate_signature
|
||
|
||
candidate = latest_delete_candidates[0]
|
||
self.select_listing_status(candidate["status"])
|
||
row = self._get_first_inventory_row(candidate["status"])
|
||
target = self._summarize_inventory_row(row)
|
||
success_message = self._delete_inventory_row(row, candidate["status"])
|
||
|
||
result = {
|
||
"status": candidate["status"],
|
||
"canonicalStatus": candidate["canonicalStatus"],
|
||
"statusQuantity": candidate["quantity"],
|
||
"target": target,
|
||
"successMessage": success_message,
|
||
}
|
||
all_results.append(result)
|
||
delete_counts[candidate["canonicalStatus"]] = delete_counts.get(candidate["canonicalStatus"], 0) + 1
|
||
total_deleted += 1
|
||
|
||
if max_delete_operations is not None and total_deleted >= max_delete_operations:
|
||
stop_reason = "max-delete-operations"
|
||
break
|
||
|
||
self._wait_inventory_loader()
|
||
|
||
final_country_sections = self.get_listing_status_country_sections(shop_name, country)
|
||
final_status_rows = deepcopy(final_country_sections[0].get("rows") if final_country_sections else latest_status_rows)
|
||
|
||
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 select_listing_status(self, status: str) -> Dict[str, Any]:
|
||
normalized_status = self._normalize_option_text(status)
|
||
if not normalized_status:
|
||
raise RuntimeError("商品状态为空，无法执行筛选")
|
||
|
||
direct_result = self._click_visible_listing_status(normalized_status)
|
||
if direct_result.get("ok"):
|
||
time.sleep(LISTING_STATUS_SELECT_DELAY)
|
||
self._wait_inventory_loader()
|
||
return {"mode": "visible-status", **direct_result}
|
||
|
||
self.open_listing_status_dropdown()
|
||
script = f"const __crawlerPayload = {json.dumps({'status': normalized_status}, ensure_ascii=False)};\n{LISTING_STATUS_SELECT_OPTION_SCRIPT}"
|
||
result = self._run_js(script) or {}
|
||
if not result.get("ok"):
|
||
raise RuntimeError(f"未找到商品状态选项[{normalized_status}]: {result}")
|
||
time.sleep(LISTING_STATUS_SELECT_DELAY)
|
||
self._wait_inventory_loader()
|
||
return {"mode": "dropdown-option", **result}
|
||
|
||
def _click_visible_listing_status(self, status: str) -> Dict[str, Any]:
|
||
script = (
|
||
f"const __crawlerPayload = {json.dumps({'status': status}, ensure_ascii=False)};\n"
|
||
f"{LISTING_STATUS_CLICK_VISIBLE_STATUS_SCRIPT}"
|
||
)
|
||
return self._run_js(script) or {}
|
||
|
||
def _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}]筛选后未找到商品行")
|
||
|
||
@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 _delete_inventory_row(self, row, status: str) -> str:
|
||
try:
|
||
dropdown = row.ele('xpath:.//kat-dropdown-button[@variant="secondary"]', timeout=5)
|
||
except Exception:
|
||
dropdown = None
|
||
if not dropdown:
|
||
raise RuntimeError(f"状态[{status}]商品行缺少三点菜单")
|
||
|
||
dropdown.click()
|
||
time.sleep(1)
|
||
|
||
try:
|
||
delete_button = dropdown.sr("xpath:.//button[@role='menuitem' and @data-action='DeleteListing']")
|
||
except Exception:
|
||
delete_button = None
|
||
if not delete_button:
|
||
raise RuntimeError(f"状态[{status}]三点菜单中未找到“删除商品信息”按钮")
|
||
|
||
delete_button.wait.displayed(raise_err=False)
|
||
delete_button.click()
|
||
|
||
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()
|
||
return self._wait_delete_success_message(status)
|
||
|
||
def _wait_delete_success_message(self, status: str) -> str:
|
||
deadline = time.time() + DELETE_SUCCESS_TIMEOUT
|
||
seen_messages = []
|
||
while time.time() < deadline:
|
||
alerts = self.tab.eles("xpath://kat-alert[@variant='success' and not(@dismissed)]", timeout=2)
|
||
for alert in alerts:
|
||
text = self._extract_alert_text(alert)
|
||
if not text:
|
||
continue
|
||
if text not in seen_messages:
|
||
seen_messages.append(text)
|
||
if self._is_delete_success_message(text):
|
||
return text
|
||
time.sleep(0.3)
|
||
raise RuntimeError(f"状态[{status}]删除后未读取到成功提示: {seen_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)
|
||
|
||
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):
|
||
if self.tab is None:
|
||
raise RuntimeError("浏览器页面不存在，请先打开店铺并进入管理所有库存页面")
|
||
if hasattr(self.tab, "run_js"):
|
||
return self.tab.run_js(script)
|
||
if hasattr(self.tab, "run_js_loaded"):
|
||
return self.tab.run_js_loaded(script)
|
||
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 quantity <= 0 or canonical_status in LISTING_STATUS_DELETE_WHITELIST:
|
||
continue
|
||
|
||
candidates.append(
|
||
{
|
||
"status": status,
|
||
"canonicalStatus": canonical_status,
|
||
"quantity": quantity,
|
||
}
|
||
)
|
||
return candidates
|
||
|
||
@staticmethod
|
||
def _limit_delete_candidates(candidates: Iterable[Dict[str, Any]], max_delete_count: int = 1) -> List[Dict[str, Any]]:
|
||
if max_delete_count <= 0:
|
||
return []
|
||
return list(candidates)[:max_delete_count]
|
||
|
||
@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 ProductTask:
|
||
"""巡店删除任务处理类。"""
|
||
|
||
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)} 个店铺")
|
||
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)
|
||
|
||
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"]
|
||
|
||
if driver is not None:
|
||
self._safe_close_store(driver)
|
||
|
||
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
|
||
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)
|
||
driver.switch_to_manage_inventory()
|
||
|
||
delete_result = driver.delete_all_listings_from_non_whitelisted_statuses(shop_name, country_name)
|
||
latest_status_rows = delete_result.get("statusRows") or []
|
||
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 "",
|
||
)
|
||
|
||
cart_ratio = self._empty_cart_ratio(country_name)
|
||
try:
|
||
cart_ratio = self._extract_country_cart_ratio(
|
||
driver.get_recommended_offer_cart_ratios().get("cartRatios") or [],
|
||
country_name,
|
||
)
|
||
except Exception as cart_exc:
|
||
self.log(f"读取国家 {country_name} 购物车比例失败: {str(cart_exc)}", "WARNING")
|
||
|
||
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")
|
||
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": self._empty_cart_ratio(country_name),
|
||
"deletedCount": 0,
|
||
"reopenRequired": self._should_reopen_browser(str(exc)),
|
||
}
|
||
|
||
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):
|
||
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"] = country_sections or []
|
||
shop_payload["cartRatios"] = cart_ratios or []
|
||
|
||
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}")
|
||
if response.status_code == 200:
|
||
self.log(f"巡店删除结果回传成功: {shop_name}")
|
||
return
|
||
self.log(f"巡店删除结果回传失败，状态码: {response.status_code}", "WARNING")
|
||
except Exception as exc:
|
||
self.log(f"调用巡店删除结果API异常: {str(exc)}", "ERROR")
|
||
|
||
if retry < max_retries - 1:
|
||
time.sleep(2)
|
||
|
||
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
|
||
rows.append(
|
||
{
|
||
"status": status,
|
||
"quantity": InventoryManage._normalize_quantity_text(row.get("quantity")),
|
||
"deleteQuantity": str(delete_count) if delete_count > 0 else "",
|
||
"processStatus": process_status,
|
||
}
|
||
)
|
||
|
||
if not rows:
|
||
rows = [
|
||
{
|
||
"status": "全部",
|
||
"quantity": "",
|
||
"deleteQuantity": "",
|
||
"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
|
||
|
||
|
||
if __name__ == "__main__":
|
||
user_info: UserInfo = {"company": "rongchuang123", "username": "自动化_Robot", "password": "#20zsg25"}
|
||
driver = InventoryManage(user_info)
|
||
|
||
shop_name = "郭亚芳"
|
||
country = "德国"
|
||
|
||
driver.open_shop(shop_name)
|
||
driver.switch_to_country(country)
|
||
driver.print_recommended_offer_cart_ratios_json()
|