super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stabilize patrol delete automation against delayed Amazon UI

Browse Source 
The patrol-delete flow now waits for country and listing-status controls before acting, uses the dedicated PatrolDeleteTask entry point, and includes the browser helper scripts required by the bulk-delete path. Tests cover delayed dropdown readiness, payload rejection/retry behavior, complete-draft normalization, filtered bulk deletion, and asset fallback lookup.

Constraint: Amazon listing UI exposes status controls through dynamic KAT components and delayed option rendering

Rejected: Keep row-by-row delete flow | bulk selection is the current implemented path and is covered by the added helper scripts

Confidence: high

Scope-risk: moderate

Directive: Do not remove the patrol_delete JS helper files without checking app/amazon/patrol_delete.py script loading

Tested: uv run --group dev pytest tests\\test_amazon_base.py tests\\test_patrol_delete.py

Tested: uv run python -m py_compile amazon\\base.py amazon\\main.py blueprints\\main.py

Not-tested: Real Amazon Seller Central browser session
This commit is contained in:
koko
2026-04-28 20:46:50 +08:00
parent 8be85fd332
commit 325687d532
14 changed files with 1062 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
app/tests/test_amazon_base.py
View File

@@ -302,6 +302,23 @@ def test_switching_countries_clicks_and_verifies_target(monkeypatch):
assert target.clicks == 1
def test_open_country_dropdown_waits_for_delayed_country_list(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
dropdown = FakeElement()
first_item = FakeElement()
driver = base.AmamzonBase({})
driver.tab = FakeTab(
ele_returns={
'xpath://div[@class="dropdown-account-switcher-header-label"]': [dropdown],
'xpath://div[@class="dropdown-account-switcher-list-item"]': [None, first_item],
}
)
assert driver._open_country_dropdown() is True
assert dropdown.clicks == 2
assert first_item.clicks == 1
def test_login_returns_true_when_otp_submit_succeeds(monkeypatch):
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
password_input = FakeElement()

340
app/tests/test_patrol_delete.py
View File

@@ -1,4 +1,4 @@
from amazon.patrol_delete import InventoryManage, ProductTask
from amazon.patrol_delete import InventoryManage, PatrolDeleteTask
def test_patrol_delete_task_post_result_builds_patrol_delete_payload(monkeypatch):
@@ -8,6 +8,9 @@ def test_patrol_delete_task_post_result_builds_patrol_delete_payload(monkeypatch
status_code = 200
text = '{"success":true}'
def json(self):
return {"success": True}
def fake_post(url, **kwargs):
captured["url"] = url
captured["kwargs"] = kwargs
@@ -24,7 +27,7 @@ def test_patrol_delete_task_post_result_builds_patrol_delete_payload(monkeypatch
]
cart_ratios = [{"country": "德国", "ratio": "25%"}]
ProductTask().post_result(
PatrolDeleteTask().post_result(
task_id=3089,
shop_name="郭亚芳",
country_sections=country_sections,
@@ -50,6 +53,130 @@ def test_patrol_delete_task_post_result_builds_patrol_delete_payload(monkeypatch
assert captured["kwargs"]["json"]["shops"][0]["submissionId"].startswith("patrol-delete:3089:郭亚芳:")
def test_patrol_delete_task_post_result_rejects_empty_success_payload(monkeypatch):
calls = []
def fake_post(url, **kwargs):
calls.append((url, kwargs))
raise AssertionError("empty success payload must not be posted")
monkeypatch.setattr("amazon.patrol_delete.requests.post", fake_post)
try:
PatrolDeleteTask().post_result(task_id=3089, shop_name="郭亚芳", country_sections=[], cart_ratios=[])
except RuntimeError as exc:
assert "缺少国家状态数据" in str(exc)
else:
raise AssertionError("expected empty success payload to fail")
assert calls == []
def test_patrol_delete_task_post_result_retries_business_failure(monkeypatch):
calls = []
class Response:
status_code = 200
text = '{"success":false,"message":"bad payload"}'
def json(self):
return {"success": False, "message": "bad payload"}
def fake_post(url, **kwargs):
calls.append((url, kwargs))
return Response()
monkeypatch.setattr("config.DELETE_BRAND_API_BASE", "http://java.example")
monkeypatch.setattr("amazon.patrol_delete.requests.post", fake_post)
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: None)
try:
PatrolDeleteTask().post_result(
task_id=3089,
shop_name="郭亚芳",
country_sections=[{"country": "德国", "rows": [{"status": "全部", "quantity": "1"}]}],
cart_ratios=[{"country": "德国", "ratio": ""}],
)
except RuntimeError as exc:
assert "最终失败" in str(exc)
else:
raise AssertionError("expected Java business failure to fail")
assert len(calls) == 3
def test_normalize_country_sections_omits_status_fields_for_complete_draft_rows():
rows = PatrolDeleteTask._normalize_country_sections_for_result(
[
{
"country": "意大利",
"rows": [
{
"status": "商品信息草稿",
"quantity": "12",
"deleteQuantity": "3",
"processStatus": "已完成",
},
{
"status": "缺少的信息",
"quantity": "1",
"deleteQuantity": "1",
"processStatus": "已完成",
},
],
}
]
)
assert rows == [
{
"country": "意大利",
"rows": [
{"quantity": "12", "deleteQuantity": "3"},
{"status": "缺少的信息", "quantity": "1", "deleteQuantity": "1", "processStatus": "已完成"},
],
}
]
def test_replace_complete_draft_rows_uses_quick_view_rows():
country_section = {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"status": "商品信息草稿", "quantity": "", "deleteQuantity": "0", "processStatus": ""},
],
}
complete_draft_section = {
"country": "西班牙",
"rows": [
{"type": "completeDraft", "quantity": "1", "deleteQuantity": "1"},
{"type": "completeDraft", "quantity": "0", "deleteQuantity": "0"},
],
}
merged = PatrolDeleteTask._replace_complete_draft_rows(country_section, complete_draft_section)
assert merged == {
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"type": "completeDraft", "quantity": "1", "deleteQuantity": "1"},
{"type": "completeDraft", "quantity": "0", "deleteQuantity": "0"},
],
}
assert PatrolDeleteTask._normalize_country_sections_for_result([merged]) == [
{
"country": "西班牙",
"rows": [
{"status": "全部", "quantity": "6230", "deleteQuantity": "0", "processStatus": "无可删商品"},
{"quantity": "1", "deleteQuantity": "1"},
{"quantity": "0", "deleteQuantity": "0"},
],
}
]
def test_patrol_delete_task_process_task_aggregates_country_results(monkeypatch):
from config import runing_shop, runing_task
@@ -104,16 +231,16 @@ def test_patrol_delete_task_process_task_aggregates_country_results(monkeypatch)
"reopenRequired": True,
}
monkeypatch.setattr(ProductTask, "open_shop", fake_open_shop)
monkeypatch.setattr(ProductTask, "process_country", fake_process_country)
monkeypatch.setattr(ProductTask, "post_result", fake_post_result)
monkeypatch.setattr(PatrolDeleteTask, "open_shop", fake_open_shop)
monkeypatch.setattr(PatrolDeleteTask, "process_country", fake_process_country)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
country_sections = [
{"country": "德国", "rows": [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}]},
{"country": "英国", "rows": [{"status": "全部", "quantity": "", "deleteQuantity": "", "processStatus": ""}]},
]
cart_ratios = [{"country": "德国", "ratio": ""}, {"country": "英国", "ratio": ""}]
ProductTask().process_task(
PatrolDeleteTask().process_task(
{
"type": "patrol-delete-run",
"data": {
@@ -173,10 +300,10 @@ def test_patrol_delete_task_process_task_skips_missing_required_data(monkeypatch
def fake_post_result(self, **kwargs):
calls.append(kwargs)
monkeypatch.setattr(ProductTask, "post_result", fake_post_result)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
ProductTask().process_task({"type": "patrol-delete-run", "data": {"items": [{"shopName": "郭亚芳"}]}})
ProductTask().process_task({"type": "patrol-delete-run", "data": {"taskId": 1, "items": []}})
PatrolDeleteTask().process_task({"type": "patrol-delete-run", "data": {"items": [{"shopName": "郭亚芳"}]}})
PatrolDeleteTask().process_task({"type": "patrol-delete-run", "data": {"taskId": 1, "items": []}})
assert calls == []
@@ -194,10 +321,10 @@ def test_patrol_delete_task_reports_shop_error(monkeypatch):
def fake_open_shop(self, **kwargs):
return None
monkeypatch.setattr(ProductTask, "open_shop", fake_open_shop)
monkeypatch.setattr(ProductTask, "post_result", fake_post_result)
monkeypatch.setattr(PatrolDeleteTask, "open_shop", fake_open_shop)
monkeypatch.setattr(PatrolDeleteTask, "post_result", fake_post_result)
ProductTask().process_task(
PatrolDeleteTask().process_task(
{
"type": "patrol-delete-run",
"data": {
@@ -216,91 +343,121 @@ def test_patrol_delete_task_reports_shop_error(monkeypatch):
assert calls == [{"task_id": 2, "shop_name": "郭亚芳", "error": "店铺 郭亚芳 打开失败", "shop_done": True}]
def test_delete_all_listings_from_non_whitelisted_statuses_loops_until_no_candidates(monkeypatch):
def test_delete_all_listings_from_non_whitelisted_statuses_deletes_each_row_in_status(monkeypatch):
driver = InventoryManage({})
sections = [
status_rows = [
[
{
"country": "德国",
"rows": [
{"status": "商品信息草稿", "quantity": "2", "deleteQuantity": "", "processStatus": ""},
{"status": "全部", "quantity": "9", "deleteQuantity": "", "processStatus": ""},
],
}
{"status": "商品信息草稿", "quantity": "2", "deleteQuantity": "", "processStatus": ""},
{"status": "全部", "quantity": "9", "deleteQuantity": "", "processStatus": ""},
],
[
{
"country": "德国",
"rows": [
{"status": "商品信息草稿", "quantity": "1", "deleteQuantity": "", "processStatus": ""},
{"status": "全部", "quantity": "8", "deleteQuantity": "", "processStatus": ""},
],
}
],
[
{
"country": "德国",
"rows": [
{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""},
],
}
],
[
{
"country": "德国",
"rows": [
{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""},
],
}
{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""},
],
]
state = {"index": 0, "selected": [], "deleted": 0}
def fake_get_listing_status_country_sections(shop_name, country):
current = sections[state["index"]]
def fake_read_listing_status_rows_for_delete(shop_name, country):
current = status_rows[min(state["index"], len(status_rows) - 1)]
state["index"] += 1
return current
def fake_select_listing_status(status):
state["selected"].append(status)
def fake_get_first_inventory_row(status):
return {"sku": f"sku-{state['deleted'] + 1}"}
def fake_delete_selected_filtered_inventory_rows(status):
state["deleted"] += 2
return {
"deletedCount": 2,
"target": {"sku": "bulk-selected", "rawText": "bulk-selected"},
"successMessage": f"{status} 删除成功",
}
def fake_summarize_inventory_row(row):
return {"sku": row["sku"], "rawText": row["sku"]}
def fake_delete_inventory_row(row, status):
state["deleted"] += 1
return f"{status} 删除成功"
monkeypatch.setattr(driver, "get_listing_status_country_sections", fake_get_listing_status_country_sections)
monkeypatch.setattr(driver, "_read_listing_status_rows_for_delete", fake_read_listing_status_rows_for_delete)
monkeypatch.setattr(driver, "select_listing_status", fake_select_listing_status)
monkeypatch.setattr(driver, "_get_first_inventory_row", fake_get_first_inventory_row)
monkeypatch.setattr(driver, "_summarize_inventory_row", fake_summarize_inventory_row)
monkeypatch.setattr(driver, "_delete_inventory_row", fake_delete_inventory_row)
monkeypatch.setattr(driver, "_delete_selected_filtered_inventory_rows", fake_delete_selected_filtered_inventory_rows)
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: None)
result = driver.delete_all_listings_from_non_whitelisted_statuses("郭亚芳", "德国")
assert state["selected"] == ["商品信息草稿", "商品信息草稿"]
assert state["selected"] == ["商品信息草稿"]
assert state["deleted"] == 2
assert result["deleteCounts"] == {"商品信息草稿": 2}
assert result["totalDeleted"] == 2
assert result["stopReason"] == "no-candidates"
assert result["stopReason"] == "processed-candidates"
assert result["statusRows"] == [{"status": "全部", "quantity": "7", "deleteQuantity": "", "processStatus": ""}]
assert [item["target"]["sku"] for item in result["results"]] == ["sku-1", "sku-2"]
assert [item["target"]["sku"] for item in result["results"]] == ["bulk-selected"]
def test_select_listing_status_uses_dropdown_option_not_visible_status(monkeypatch):
driver = InventoryManage({})
calls = []
monkeypatch.setattr(driver, "open_listing_status_dropdown", lambda: calls.append("open") or {"ok": True})
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: None)
monkeypatch.setattr(driver, "_run_js", lambda script, *args: calls.append((script, args)) or {"ok": True})
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: None)
result = driver.select_listing_status("商品信息草稿")
assert result["mode"] == "dropdown-option"
assert calls[0] == "open"
assert calls[-1][1] == ({"status": "商品信息草稿"},)
def test_get_listing_status_dropdown_options_closes_dropdown_after_read(monkeypatch):
driver = InventoryManage({})
calls = []
monkeypatch.setattr(driver, "open_listing_status_dropdown", lambda: calls.append("open") or {"ok": True})
monkeypatch.setattr(
driver,
"_wait_listing_status_option_rows",
lambda: calls.append("read") or [{"raw_text": "详情页面已删除 (1)", "value": "DetailPageRemoved"}],
)
monkeypatch.setattr(driver, "_close_listing_status_dropdown", lambda: calls.append("close"))
rows = driver.get_listing_status_dropdown_options("郭亚芳", "德国")
assert calls == ["open", "read", "close"]
assert rows[0]["status"] == "详情页面已删除"
def test_open_listing_status_dropdown_waits_until_component_enabled(monkeypatch):
driver = InventoryManage({})
calls = []
class Wait:
def doc_loaded(self, raise_err=False):
calls.append("doc_loaded")
class Tab:
wait = Wait()
responses = [
{"ok": False, "reason": "listing status dropdown disabled/loading"},
{"ok": True, "selected": {"text": "全部在售不可售商品信息草稿"}},
]
driver.tab = Tab()
monkeypatch.setattr(driver, "_wait_inventory_loader", lambda timeout=5: calls.append(("loader", timeout)))
monkeypatch.setattr(driver, "_run_js", lambda script: calls.append("js") or responses.pop(0))
monkeypatch.setattr("amazon.patrol_delete.time.sleep", lambda seconds: calls.append(("sleep", seconds)))
result = driver.open_listing_status_dropdown()
assert result["ok"] is True
assert calls.count("js") == 2
def test_extract_country_cart_ratio_prefers_ratio2daysago_only():
ratio = ProductTask._extract_country_cart_ratio(
ratio = PatrolDeleteTask._extract_country_cart_ratio(
[
{"country": "德国", "ratio2DaysAgo": "25%", "ratio30DaysAgo": "19%"},
{"country": "英国", "ratio30DaysAgo": "40%"},
],
"德国",
)
missing_ratio2 = ProductTask._extract_country_cart_ratio(
missing_ratio2 = PatrolDeleteTask._extract_country_cart_ratio(
[{"country": "英国", "ratio30DaysAgo": "40%"}],
"英国",
)
@@ -455,6 +612,20 @@ def test_build_deletable_status_rows_keeps_non_whitelisted_statuses():
]
def test_build_deletable_status_rows_skips_group_labels_without_quantity():
rows = [
{"status": "商品信息草稿", "quantity": ""},
{"status": "详情页面已删除", "quantity": None},
{"status": "自定义异常状态", "quantity": "0"},
{"status": "缺少的信息", "quantity": "1"},
{"status": "全部", "quantity": ""},
]
assert InventoryManage._build_deletable_status_rows(rows) == [
{"status": "缺少的信息", "canonicalStatus": "缺少的信息", "quantity": 1},
]
def test_build_deletable_status_rows_compatibility_map_prevents_whitelist_misdelete():
rows = [
{"status": "搜尋結果中禁止顯示", "quantity": "4"},
@@ -480,20 +651,43 @@ def test_build_deletable_status_rows_keeps_legacy_non_whitelist_statuses_deletab
]
def test_limit_delete_candidates_enforces_single_delete_cap():
candidates = [
{"status": "商品信息草稿", "canonicalStatus": "商品信息草稿", "quantity": 2},
{"status": "自定义异常状态", "canonicalStatus": "自定义异常状态", "quantity": 5},
]
assert InventoryManage._limit_delete_candidates(candidates, max_delete_count=1) == [candidates[0]]
def test_is_delete_success_message_accepts_both_copy_variants():
assert InventoryManage._is_delete_success_message("1 个商品已经删除。所作更改需要15分钟才会显示在商品详情页面上")
assert InventoryManage._is_delete_success_message("所做更改需要 15 分钟才会显示在商品详情页面上。 1 个商品已删除。")
def test_extract_delete_success_count_accepts_success_message():
assert InventoryManage._extract_delete_success_count("所做更改需要 15 分钟才会显示在商品详情页面上。 3 个商品已删除。") == 3
def test_build_country_section_result_sets_zero_delete_quantity_when_no_deletable_products():
section = PatrolDeleteTask._build_country_section_result(
country_name="德国",
status_rows=[{"status": "全部", "quantity": "12"}],
delete_counts={},
default_process_status="无可删商品",
)
assert section == {
"country": "德国",
"rows": [{"status": "全部", "quantity": "12", "deleteQuantity": "0", "processStatus": "无可删商品"}],
}
def test_build_country_section_result_sets_zero_delete_quantity_for_empty_no_deletable_result():
section = PatrolDeleteTask._build_country_section_result(
country_name="德国",
status_rows=[],
delete_counts={},
default_process_status="无可删商品",
)
assert section == {
"country": "德国",
"rows": [{"status": "全部", "quantity": "", "deleteQuantity": "0", "processStatus": "无可删商品"}],
}
def test_parse_label_quantity_text_accepts_complete_draft_sample():
assert InventoryManage._parse_label_quantity_text("未提交的草稿 (0)") == ("未提交的草稿", 0)