fix(status): ignore promotion edit warnings

This commit is contained in:
chengma
2026-07-20 10:52:32 +08:00
parent b3d9857b3e
commit 9c4224286a
10 changed files with 253 additions and 14 deletions
+38
View File
@@ -18,6 +18,7 @@ from .config import make_slug
DEFAULT_BUSY_TIMEOUT_MS = 5000
PRODUCT_STATUS_FEATURE_INTRODUCED_AT = "2026-07-18T16:34:37"
LEGACY_PRODUCT_STATUS_DEFAULT_NOTE = "历史批次默认按架上商品处理(未实时检测)"
LEGACY_PROMOTION_STATUS_REPAIR_NOTE = "历史商品状态修正:促销编辑提示不属于商品状态异常"
VALID_BATCH_FIELDS = {
"source_files_json",
"status",
@@ -501,6 +502,7 @@ def init_db(path=None, conn=None) -> None:
_ensure_task_product_status_columns(database)
_ensure_task_delete_columns(database)
_migrate_legacy_product_status_defaults(database)
_repair_legacy_promotion_status_defaults(database)
_ensure_image_studio_project_suite_columns(database)
_ensure_image_studio_project_draft_columns(database)
_ensure_image_studio_job_recovery_columns(database)
@@ -578,6 +580,42 @@ def _migrate_legacy_product_status_defaults(database):
)
def _repair_legacy_promotion_status_defaults(database):
"""Repair only legacy unknown snapshots caused by a known benign warning."""
rows = database.execute(
"""
SELECT tasks.id, tasks.product_status_note
FROM tasks
INNER JOIN batches ON batches.id = tasks.batch_id
WHERE tasks.product_status = 'unknown'
AND tasks.deleted_at IS NULL
AND batches.deleted_at IS NULL
AND batches.created_at < ?
""",
(PRODUCT_STATUS_FEATURE_INTRODUCED_AT,),
).fetchall()
task_ids = [
int(row["id"])
for row in rows
if product_status.is_promotion_edit_restriction_note(row["product_status_note"])
]
if not task_ids:
return
placeholders = ", ".join("?" for _task_id in task_ids)
database.execute(
f"""
UPDATE tasks
SET product_status = 'normal',
product_status_note = ?,
updated_at = ?
WHERE id IN ({placeholders})
""",
(LEGACY_PROMOTION_STATUS_REPAIR_NOTE, _now(), *task_ids),
)
def _ensure_image_studio_project_suite_columns(database):
columns = {
row["name"]
+1 -1
View File
@@ -669,7 +669,7 @@ class ApplyTab(QWidget):
if status_text:
lines.append("当前筛选结果没有状态正常且可更新的商品。")
lines.append(status_text)
lines.append("请先回到①导入采集重新确认商品状态。")
lines.append("未上架、审核中或状态未知的商品为安全起见不会更新。")
content_error = self._update_content_error(update_plan, update_mode)
if content_error:
if lines:
+28 -4
View File
@@ -56,6 +56,10 @@ COLLECT_SCOPE_ALL = SCOPE_ALL
_WHITESPACE_RE = re.compile(r"\s+")
_NOTE_LIMIT = 2000
_PROMOTION_EDIT_RESTRICTION_MARKERS = (
("促銷", "無法進行編輯"),
("促销", "无法进行编辑"),
)
def normalize_status(value) -> str:
@@ -97,6 +101,22 @@ def normalize_text(value) -> str:
return _WHITESPACE_RE.sub(" ", str(value or "").strip())
def is_promotion_edit_restriction_alert(title, description="") -> bool:
"""Return whether an alert only describes a promotion editing limitation."""
text = normalize_text(f"{title or ''} {description or ''}")
return any(
promotion_marker in text and edit_marker in text
for promotion_marker, edit_marker in _PROMOTION_EDIT_RESTRICTION_MARKERS
)
def is_promotion_edit_restriction_note(note) -> bool:
"""Recognize the persisted summary generated from a known benign alert."""
return is_promotion_edit_restriction_alert(note)
def classify_alerts(alerts) -> dict:
"""Classify normalized EDS warning-alert data without retaining page HTML."""
@@ -105,8 +125,8 @@ def classify_alerts(alerts) -> dict:
if not alerts:
return _snapshot(STATUS_NORMAL, "")
first_note = ""
valid_alert_seen = False
unknown_alert_note = ""
for alert in alerts:
if not isinstance(alert, dict):
continue
@@ -114,16 +134,20 @@ def classify_alerts(alerts) -> dict:
title = normalize_text(alert.get("title"))
description = normalize_text(alert.get("description"))
note = _alert_note(title, description)
if not first_note:
first_note = note
if title in {"審核中", "审核中"}:
return _snapshot(STATUS_REVIEWING, note)
if "您的商品未上架" in title:
return _snapshot(STATUS_UNLISTED, note)
if is_promotion_edit_restriction_alert(title, description):
continue
if not unknown_alert_note:
unknown_alert_note = note
if not valid_alert_seen:
return _snapshot(STATUS_UNKNOWN, "")
return _snapshot(STATUS_UNKNOWN, first_note)
if not unknown_alert_note:
return _snapshot(STATUS_NORMAL, "")
return _snapshot(STATUS_UNKNOWN, unknown_alert_note)
def partition_tasks(tasks) -> dict: