chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(collect): persist product status snapshots

Browse Source
This commit is contained in:
chengma
2026-07-18 16:34:37 +08:00
parent b4250f36f8
commit 0cf23db403
11 changed files with 464 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+63
View File
@@ -40,6 +40,9 @@ class DbTests(TempDirMixin, unittest.TestCase):
"image_task_key",
"cover_reset_count",
"cover_reset_at",
"product_status",
"product_status_note",
"product_status_at",
}.issubset(task_columns)
)
finally:
@@ -113,9 +116,13 @@ class DbTests(TempDirMixin, unittest.TestCase):
}
self.assertIn("cover_reset_count", columns)
self.assertIn("cover_reset_at", columns)
self.assertIn("product_status", columns)
self.assertIn("product_status_note", columns)
self.assertIn("product_status_at", columns)
task = db.get_task(1, conn=conn)
self.assertEqual(0, task.cover_reset_count)
self.assertIsNone(task.cover_reset_at)
self.assertIsNone(task.product_status)
finally:
conn.close()
@@ -217,6 +224,62 @@ class DbTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_product_status_snapshot_preserves_collection_lifecycle_and_content(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
batch_id = db.create_batch(["input.xlsx"], path=db_path)
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "Sheet1",
"source_row": 2,
"account_name": "shop",
"alias": "alias",
"item_id": "51100639510",
}
],
path=db_path,
)
task = db.list_tasks(batch_id=batch_id, path=db_path)[0]
db.set_collected(task.id, "旧标题", "old.jpg", db_path)
db.set_product_status(
task.id,
"reviewing",
"审核说明",
"2026-07-18T09:00:00",
path=db_path,
)
status_only = db.get_task(task.id, path=db_path)
self.assertEqual("collected", status_only.stage)
self.assertEqual("success", status_only.status)
self.assertEqual("旧标题", status_only.old_title)
self.assertEqual("old.jpg", status_only.old_cover_path)
self.assertEqual("reviewing", status_only.product_status)
self.assertEqual("审核说明", status_only.product_status_note)
self.assertEqual("2026-07-18T09:00:00", status_only.product_status_at)
db.set_collected(
task.id,
"重新采集标题",
"new-old.jpg",
product_status_value="not-a-status",
product_status_note="x" * 2200,
product_status_at="2026-07-18T10:00:00",
path=db_path,
)
refreshed = db.get_task(task.id, path=db_path)
self.assertEqual("unknown", refreshed.product_status)
self.assertEqual("重新采集标题", refreshed.old_title)
self.assertEqual("new-old.jpg", refreshed.old_cover_path)
self.assertEqual("2026-07-18T10:00:00", refreshed.product_status_at)
self.assertEqual(2000, len(refreshed.product_status_note))
self.assert_removed(temp_dir)
def test_set_generated_cover_preserves_title_and_records_ai_success(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
tests/test_editor_login.py
+62 -1
View File
@@ -39,11 +39,12 @@ class FakeCDP:
class FakeProductCDP:
def __init__(self, ws, ready=True, toasts=None, rects=None, url=""):
def __init__(self, ws, ready=True, toasts=None, rects=None, alerts=None, url=""):
self.ws = ws
self.ready = ready
self.toasts = list(toasts or [])
self.rects = list(rects or [])
self.alerts = list(alerts or [])
self.url = url
self.closed = False
self.sent = []
@@ -63,6 +64,8 @@ class FakeProductCDP:
return json.dumps(self.toasts)
if expr == editor.JS_RECTS:
return json.dumps(self.rects)
if expr == editor.JS_PRODUCT_STATUS_ALERTS:
return json.dumps(self.alerts)
return None
def send(self, method, params=None):
@@ -910,6 +913,64 @@ class EditorLoginTests(unittest.TestCase):
timeout=2.0,
)
def test_read_product_status_uses_warning_alerts_without_vue_scope_selectors(self):
cdp = FakeProductCDP(
"ws-status",
alerts=[
{"title": "无关提示", "description": "保留为未知"},
{"title": " 審核中 ", "description": " 商品等待审核 "},
],
)
result = editor.read_product_status(cdp)
self.assertEqual("reviewing", result["product_status"])
self.assertEqual("标题：審核中；说明：商品等待审核", result["product_status_note"])
self.assertIsNone(result["product_status_error"])
self.assertIn(".eds-alert.eds-alert--warning", editor.JS_PRODUCT_STATUS_ALERTS)
self.assertNotIn("data-v-", editor.JS_PRODUCT_STATUS_ALERTS)
def test_read_product_status_degrades_to_unknown_when_cdp_response_is_invalid(self):
class BrokenStatusCDP:
def val(self, _expr):
raise RuntimeError("状态读取失败")
result = editor.read_product_status(BrokenStatusCDP())
self.assertEqual("unknown", result["product_status"])
self.assertIsNone(result["product_status_note"])
self.assertIn("状态读取失败", result["product_status_error"])
def test_collect_reads_product_status_before_title_and_returns_snapshot(self):
cdp = FakeProductCDP("ws-new", alerts=[])
steps = []
def read_title_after_status(_cdp):
self.assertIn("read_product_status", steps)
return "旧标题"
with mock.patch("app.editor.open_product", return_value=cdp), mock.patch(
"app.editor.read_title",
side_effect=read_title_after_status,
), mock.patch(
"app.editor.read_cover_src",
return_value="https://down-ws-sg.vod.susercontent.com/cover.jpg",
), mock.patch(
"app.editor.download_cover",
return_value="images/main/51100639510_old.jpg",
):
result = editor.collect(
{"debug_port": 9222},
{"item_id": "51100639510", "old_cover_path": "old.jpg"},
on_step=steps.append,
)
self.assertEqual("normal", result["product_status"])
self.assertEqual(
["read_product_status", "read_title", "read_cover", "download_cover"],
steps,
)
def test_collect_keeps_reused_product_tab_open(self):
cdp = FakeProductCDP("ws-existing")
cdp.target_id = "target-existing"
tests/test_gui.py
+4
View File
@@ -8256,6 +8256,8 @@ class GuiTests(TempDirMixin, unittest.TestCase):
return {
"old_title": "旧标题",
"old_cover_path": task["old_cover_path"],
"product_status": "normal",
"product_status_note": None,
}
worker = CollectWorker(
@@ -8308,6 +8310,8 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual("success", by_alias["alias-a"].status)
self.assertEqual("旧标题", by_alias["alias-a"].old_title)
self.assertTrue(by_alias["alias-a"].old_cover_path.endswith(expected_old_cover))
self.assertEqual("normal", by_alias["alias-a"].product_status)
self.assertIsNotNone(by_alias["alias-a"].product_status_at)
self.assertEqual("imported", by_alias["alias-b"].stage)
self.assertEqual("skipped", by_alias["alias-b"].status)
self.assertIn("账号未登录", by_alias["alias-b"].last_error)
tests/test_product_status.py
+80
View File
@@ -0,0 +1,80 @@
import html
import re
import sys
import unittest
from pathlib import Path
from types import SimpleNamespace
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from app import product_status
REPO_ROOT = Path(__file__).resolve().parent.parent
def _sample_alert(filename):
source = (REPO_ROOT / "docs" / "html" / filename).read_text(encoding="utf-8")
title_match = re.search(r'class="eds-alert-title">\s*([^<]+)', source)
desc_match = re.search(r'class="eds-alert-desc"[^>]*>(.*?)</p>', source, re.DOTALL)
if title_match is None or desc_match is None:
raise AssertionError("商品状态样本缺少标题或说明")
description = re.sub(r"<[^>]+>", "", desc_match.group(1))
return {
"title": html.unescape(title_match.group(1)),
"description": html.unescape(description),
}
class ProductStatusTests(unittest.TestCase):
def test_real_warning_samples_are_classified(self):
reviewing = product_status.classify_alerts([_sample_alert("审核中商品提示.html")])
unlisted = product_status.classify_alerts([_sample_alert("未上架商品提示.html")])
self.assertEqual(product_status.STATUS_REVIEWING, reviewing["product_status"])
self.assertEqual(product_status.STATUS_UNLISTED, unlisted["product_status"])
def test_empty_unknown_and_multiple_alerts_follow_contract(self):
self.assertEqual(
product_status.STATUS_NORMAL,
product_status.classify_alerts([])["product_status"],
)
self.assertEqual(
product_status.STATUS_UNKNOWN,
product_status.classify_alerts([{"title": "其他警告", "description": "说明"}])["product_status"],
)
result = product_status.classify_alerts(
[
{"title": "其他警告", "description": "忽略"},
{"title": "您的商品未上架", "description": "已下架"},
{"title": "审核中", "description": "后续提示"},
]
)
self.assertEqual(product_status.STATUS_UNLISTED, result["product_status"])
self.assertIn("您的商品未上架", result["product_status_note"])
def test_invalid_values_are_unknown_and_partitioning_is_consistent(self):
tasks = [
SimpleNamespace(product_status="normal"),
SimpleNamespace(product_status="reviewing"),
SimpleNamespace(product_status=None),
SimpleNamespace(product_status="invalid"),
]
grouped = product_status.partition_tasks(tasks)
self.assertEqual("状态未知", product_status.status_label(None))
self.assertTrue(product_status.is_normal("normal"))
self.assertFalse(product_status.is_normal(None))
self.assertTrue(product_status.is_known_abnormal("unlisted"))
self.assertEqual(1, len(grouped["normal"]))
self.assertEqual(2, len(grouped["unknown"]))
def test_note_is_normalized_and_limited(self):
result = product_status.classify_alerts(
[{"title": " 审核中\n", "description": " " + "x" * 2200}]
)
self.assertEqual(product_status.STATUS_REVIEWING, result["product_status"])
self.assertLessEqual(len(result["product_status_note"]), 2000)
self.assertTrue(result["product_status_note"].startswith("标题：审核中；说明："))