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

feat: 完成T-504更新执行增强

Browse Source 
- 新增 dry-run、多账号并行和最大并行账号数设置

- 增加 run_logs/run_log_events 运行日志表及读写接口

- ApplyWorker 支持 dry-run 预览、按账号并行、端口冲突阻断和日志展示

- 补充 DB/GUI 单元测试并同步任务、架构、API、路由和当前状态文档

验证: python -m compileall app main.py tests; python -m unittest discover -s tests
This commit is contained in:
chengma
2026-06-29 10:25:09 +08:00
parent 01e319cad8
commit 0e0ed193b6
17 changed files with 1071 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/appconfig.py
+3
View File
@@ -46,6 +46,9 @@ DEFAULT_CONFIG = {
"allow_cover_update": False,
"max_items_per_run": 1,
"close_success_tab": False,
"dry_run": False,
"parallel_accounts": False,
"max_parallel_accounts": 2,
},
}
app/db.py
+199
View File
@@ -28,6 +28,15 @@ VALID_ACCOUNT_FIELDS = {
"note",
"last_login_at",
}
VALID_RUN_LOG_FIELDS = {
"status",
"done",
"success_count",
"skipped_count",
"failed_count",
"finished_at",
"summary_json",
}
PHASE_ATTEMPT_FIELDS = {
"collect": "collect_attempts",
"collected": "collect_attempts",
@@ -104,6 +113,43 @@ class Task:
updated_at: str
@dataclass(frozen=True)
class RunLog:
id: int
run_type: str
dry_run: int
status: str
total: int
done: int
success_count: int
skipped_count: int
failed_count: int
options_json: Optional[str]
summary_json: Optional[str]
started_at: str
finished_at: Optional[str]
@property
def options(self) -> dict:
return json.loads(self.options_json or "{}")
@property
def summary(self) -> dict:
return json.loads(self.summary_json or "{}")
@dataclass(frozen=True)
class RunLogEvent:
id: int
run_id: int
task_id: Optional[int]
alias: Optional[str]
item_id: Optional[str]
level: str
message: str
created_at: str
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS batches (
id TEXT PRIMARY KEY,
@@ -163,6 +209,36 @@ CREATE INDEX IF NOT EXISTS idx_tasks_batch_stage_status
ON tasks(batch_id, stage, status);
CREATE INDEX IF NOT EXISTS idx_tasks_alias ON tasks(alias);
CREATE INDEX IF NOT EXISTS idx_tasks_item ON tasks(item_id);
CREATE TABLE IF NOT EXISTS run_logs (
id INTEGER PRIMARY KEY,
run_type TEXT NOT NULL,
dry_run INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'running',
total INTEGER NOT NULL DEFAULT 0,
done INTEGER NOT NULL DEFAULT 0,
success_count INTEGER NOT NULL DEFAULT 0,
skipped_count INTEGER NOT NULL DEFAULT 0,
failed_count INTEGER NOT NULL DEFAULT 0,
options_json TEXT,
summary_json TEXT,
started_at TEXT NOT NULL,
finished_at TEXT
);
CREATE TABLE IF NOT EXISTS run_log_events (
id INTEGER PRIMARY KEY,
run_id INTEGER NOT NULL REFERENCES run_logs(id) ON DELETE CASCADE,
task_id INTEGER,
alias TEXT,
item_id TEXT,
level TEXT NOT NULL DEFAULT 'info',
message TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_run_logs_started ON run_logs(started_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_run_log_events_run ON run_log_events(run_id, id);
"""
@@ -561,3 +637,126 @@ def set_applied(task_id, committed, error=None, path=None, conn=None) -> None:
""",
(str(error or "未提交更新"), now, int(task_id)),
)
def create_run_log(
run_type,
dry_run=False,
total=0,
options=None,
path=None,
conn=None,
) -> int:
"""Create a high-level operation log and return its id."""
now = _now()
options_json = json.dumps(
appconfig.sanitize_for_log(options or {}),
ensure_ascii=False,
sort_keys=True,
)
with _connection(conn, path) as database:
with database:
cursor = database.execute(
"""
INSERT INTO run_logs
(run_type, dry_run, status, total, options_json, started_at)
VALUES (?, ?, 'running', ?, ?, ?)
""",
(str(run_type), 1 if dry_run else 0, int(total), options_json, now),
)
return int(cursor.lastrowid)
def finish_run_log(run_id, path=None, conn=None, **fields) -> None:
"""Mark a run log complete/blocked/failed with a sanitized summary."""
_validate_fields(fields, VALID_RUN_LOG_FIELDS)
if "summary_json" in fields and not isinstance(fields["summary_json"], str):
fields["summary_json"] = json.dumps(
appconfig.sanitize_for_log(fields["summary_json"] or {}),
ensure_ascii=False,
sort_keys=True,
)
if "finished_at" not in fields:
fields["finished_at"] = _now()
assignments = ", ".join(f"{field} = ?" for field in fields)
params = list(fields.values()) + [int(run_id)]
with _connection(conn, path) as database:
with database:
database.execute(
f"UPDATE run_logs SET {assignments} WHERE id = ?",
params,
)
def add_run_log_event(
run_id,
message,
task_id=None,
alias=None,
item_id=None,
level="info",
path=None,
conn=None,
) -> int:
"""Append one sanitized event line to a run log."""
sanitized = appconfig.sanitize_for_log(
{
"message": str(message),
"alias": alias,
"item_id": item_id,
"level": level,
}
)
with _connection(conn, path) as database:
with database:
cursor = database.execute(
"""
INSERT INTO run_log_events
(run_id, task_id, alias, item_id, level, message, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
int(run_id),
int(task_id) if task_id is not None else None,
sanitized.get("alias"),
sanitized.get("item_id"),
sanitized.get("level") or "info",
sanitized.get("message") or "",
_now(),
),
)
return int(cursor.lastrowid)
def list_run_logs(limit=50, run_type=None, path=None, conn=None):
clauses = []
params = []
if run_type is not None:
clauses.append("run_type = ?")
params.append(str(run_type))
sql = "SELECT * FROM run_logs"
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY started_at DESC, id DESC LIMIT ?"
params.append(max(1, int(limit)))
with _connection(conn, path) as database:
return _fetch_all(database, sql, params, RunLog)
def list_run_log_events(run_id=None, limit=200, path=None, conn=None):
clauses = []
params = []
if run_id is not None:
clauses.append("run_id = ?")
params.append(int(run_id))
sql = "SELECT * FROM run_log_events"
if clauses:
sql += " WHERE " + " AND ".join(clauses)
sql += " ORDER BY id DESC LIMIT ?"
params.append(max(1, int(limit)))
with _connection(conn, path) as database:
events = _fetch_all(database, sql, params, RunLogEvent)
return list(reversed(events))
app/gui.py
+420 -93
View File
@@ -4,6 +4,8 @@ from __future__ import annotations
import os
import sys
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
try:
from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt
@@ -1012,6 +1014,11 @@ if QT_IMPORT_ERROR is None:
self.task_table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.task_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.task_table.verticalHeader().setVisible(False)
self.run_log_view = QPlainTextEdit()
self.run_log_view.setObjectName("applyRunLogView")
self.run_log_view.setReadOnly(True)
self.run_log_view.setMaximumHeight(128)
self.run_log_view.setPlaceholderText("运行日志")
self.start_update_button = QPushButton("开始更新")
self.stop_update_button = QPushButton("停止")
@@ -1031,6 +1038,8 @@ if QT_IMPORT_ERROR is None:
layout.addWidget(self.risk_label)
layout.addWidget(self.summary_label)
layout.addWidget(self.task_table, 1)
layout.addWidget(QLabel("运行日志"))
layout.addWidget(self.run_log_view)
layout.addLayout(action_layout)
self.batch_filter.currentIndexChanged.connect(self.refresh_tasks)
@@ -1042,6 +1051,7 @@ if QT_IMPORT_ERROR is None:
self.write_back_button.clicked.connect(self.write_back_results)
self.refresh_tasks()
self._load_latest_run_log()
def _set_status(self, message):
if self.status_callback is not None:
@@ -1090,7 +1100,9 @@ if QT_IMPORT_ERROR is None:
if not tasks:
self._set_status("当前筛选结果没有可更新任务")
return
safety_error = self._update_safety_error(tasks)
update_cfg = self._shopee_update_config()
dry_run = bool(update_cfg.get("dry_run", False))
safety_error = self._update_safety_error(tasks, dry_run=dry_run)
if safety_error:
QMessageBox.warning(self, "更新安全开关", safety_error)
self._set_status(safety_error.replace("\n", " "))
@@ -1105,16 +1117,21 @@ if QT_IMPORT_ERROR is None:
if answer != QMessageBox.Yes:
self._set_status("已取消开始更新")
return
update_cfg = self._shopee_update_config()
worker = ApplyWorker(
tasks,
db_path=self.db_path,
config=self.config,
close_success_tab=bool(update_cfg.get("close_success_tab", False)),
dry_run=dry_run,
parallel_accounts=bool(update_cfg.get("parallel_accounts", False)),
max_parallel_accounts=max(
1,
int(update_cfg.get("max_parallel_accounts", 1) or 1),
),
)
worker.progress.connect(self._on_apply_progress)
worker.row_updated.connect(self._on_apply_row_updated)
worker.log.connect(self._set_status)
worker.log.connect(self._on_apply_log)
worker.failed.connect(self._on_apply_failed)
worker.finished.connect(self._on_apply_finished)
worker.cancelled.connect(self._on_apply_cancelled)
@@ -1123,7 +1140,11 @@ if QT_IMPORT_ERROR is None:
self.apply_worker = worker
self.apply_thread = thread
self._set_apply_running(True)
self._set_status(f"开始更新：{len(tasks)} 条")
self.run_log_view.clear()
if dry_run:
self._set_status(f"开始 dry-run 预览：{len(tasks)} 条")
else:
self._set_status(f"开始更新：{len(tasks)} 条")
thread.start()
def stop_update(self, checked=False):
@@ -1222,6 +1243,12 @@ if QT_IMPORT_ERROR is None:
update_cfg = self._shopee_update_config()
cover_text = "允许" if update_cfg.get("allow_cover_update") else "不允许"
close_text = "是" if update_cfg.get("close_success_tab") else "否"
dry_run_text = "开启" if update_cfg.get("dry_run") else "关闭"
parallel_text = (
f"开启，最多 {update_cfg.get('max_parallel_accounts', 1)} 个账号"
if update_cfg.get("parallel_accounts")
else "关闭"
)
return (
"即将按当前筛选结果开始更新 Shopee 线上商品。\n\n"
f"批次：{self._batch_filter_label()}\n"
@@ -1232,12 +1259,20 @@ if QT_IMPORT_ERROR is None:
f"测试商品ID={update_cfg.get('test_item_id') or '未配置'}，"
f"封面更新={cover_text}，"
f"最大条数={update_cfg.get('max_items_per_run', 1)}，"
f"成功后关闭新页={close_text}\n\n"
"确认后后续执行会打开商品编辑页、替换标题/允许时替换封面，并点击「更新」提交线上。"
f"成功后关闭新页={close_text}，"
f"dry-run={dry_run_text}，"
f"多账号并行={parallel_text}\n\n"
+ (
"dry-run 开启时只写运行日志和预览，不打开 Shopee、不点击「更新」、不改任务状态。"
if update_cfg.get("dry_run")
else "确认后后续执行会打开商品编辑页、替换标题/允许时替换封面，并点击「更新」提交线上。"
)
)
def _update_safety_error(self, tasks):
def _update_safety_error(self, tasks, dry_run=False):
update_cfg = self._shopee_update_config()
if dry_run:
return None
if not update_cfg.get("allow_real_submit", False):
return "设置未开启「允许真实提交线上商品」，已阻止本次更新。"
max_items = max(1, int(update_cfg.get("max_items_per_run", 1) or 1))
@@ -1304,6 +1339,10 @@ if QT_IMPORT_ERROR is None:
def _on_apply_progress(self, payload):
self._set_status("更新进度：" + self._apply_progress_text(payload))
def _on_apply_log(self, message):
self._append_run_log(message)
self._set_status(message)
def _on_apply_row_updated(self, task_id, fields):
self.refresh_tasks()
@@ -1317,9 +1356,10 @@ if QT_IMPORT_ERROR is None:
self._show_apply_blocked(payload)
return
self.last_apply_summary = dict(payload)
message = "更新完成：" + self._apply_progress_text(payload)
prefix = "dry-run 预览完成：" if payload.get("dry_run") else "更新完成："
message = prefix + self._apply_progress_text(payload)
batch_ids = payload.get("batch_ids") or self._active_batch_ids()
if payload.get("done", 0) > 0 and batch_ids:
if (not payload.get("dry_run")) and payload.get("done", 0) > 0 and batch_ids:
if self._start_result_write_back(
batch_ids,
auto=True,
@@ -1336,18 +1376,46 @@ if QT_IMPORT_ERROR is None:
self._set_status("更新已停止：" + self._apply_progress_text(payload))
def _apply_progress_text(self, payload):
return "完成{done}/{total}，成功{applied}，略过{skipped}，失败{failed}".format(
success_label = "可更新" if payload.get("dry_run") else "成功"
return "完成{done}/{total}，{success_label}{applied}，略过{skipped}，失败{failed}".format(
done=payload.get("done", 0),
total=payload.get("total", 0),
success_label=success_label,
applied=payload.get("applied", 0),
skipped=payload.get("skipped", 0),
failed=payload.get("failed", 0),
)
def _append_run_log(self, message):
self.run_log_view.appendPlainText(str(message))
def _load_latest_run_log(self):
try:
logs = db.list_run_logs(limit=1, run_type="apply", path=self.db_path)
if not logs:
return
events = db.list_run_log_events(logs[0].id, limit=30, path=self.db_path)
except Exception:
return
lines = [
f"{event.created_at} [{event.level}] {event.message}"
for event in events
]
self.run_log_view.setPlainText("\n".join(lines))
def _show_apply_blocked(self, payload):
lines = ["更新前检查未通过。"]
if payload.get("no_accounts"):
lines.append("当前没有配置账号。")
duplicate_ports = payload.get("duplicate_ports") or []
if duplicate_ports:
for item in duplicate_ports:
lines.append(
"以下账号调试端口冲突：端口 {port} -> {aliases}".format(
port=item.get("debug_port"),
aliases="、".join(item.get("aliases") or []),
)
)
not_running = payload.get("not_running") or []
if not_running:
lines.append(
@@ -1475,14 +1543,18 @@ if QT_IMPORT_ERROR is None:
def _show_apply_summary(self, apply_summary, write_back_payload=None):
QMessageBox.information(
self,
"更新完成",
"dry-run 预览完成" if apply_summary.get("dry_run") else "更新完成",
self._apply_summary_message(apply_summary, write_back_payload),
)
def _apply_summary_message(self, apply_summary, write_back_payload=None, error=None):
dry_run = bool(apply_summary.get("dry_run"))
lines = [
"更新完成。",
"成功：{applied}，失败：{failed}，略过：{skipped}".format(
"dry-run 预览完成，未打开 Shopee、未提交线上、未改任务状态。"
if dry_run
else "更新完成。",
"{success_label}：{applied}，失败：{failed}，略过：{skipped}".format(
success_label="可更新" if dry_run else "成功",
applied=apply_summary.get("applied", 0),
failed=apply_summary.get("failed", 0),
skipped=apply_summary.get("skipped", 0),
@@ -2032,7 +2104,7 @@ if QT_IMPORT_ERROR is None:
class ApplyWorker(BaseWorker):
"""Apply generated title/cover changes to Shopee one task at a time."""
"""Apply generated title/cover changes, optionally previewing or grouping by account."""
def __init__(
self,
@@ -2041,6 +2113,9 @@ if QT_IMPORT_ERROR is None:
config=None,
preflight=True,
close_success_tab=False,
dry_run=False,
parallel_accounts=False,
max_parallel_accounts=1,
):
super().__init__()
self.tasks = list(tasks)
@@ -2048,6 +2123,11 @@ if QT_IMPORT_ERROR is None:
self.config = config
self.preflight = preflight
self.close_success_tab = close_success_tab
self.dry_run = bool(dry_run)
self.parallel_accounts = bool(parallel_accounts)
self.max_parallel_accounts = max(1, int(max_parallel_accounts or 1))
self._progress_lock = threading.Lock()
self._run_id = None
def execute(self):
account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
@@ -2063,90 +2143,62 @@ if QT_IMPORT_ERROR is None:
skipped = 0
failed = 0
done = 0
if self.preflight:
blocked = self._preflight_block(eligible, account_rows, account_by_alias)
if blocked:
blocked.update(
{
"ok": False,
"blocked": True,
"total": total,
"done": 0,
"applied": 0,
"skipped": 0,
"failed": 0,
"batch_ids": batch_ids,
}
)
return blocked
for task in eligible:
if self.should_cancel():
break
account = account_by_alias.get(str(task.alias).strip())
if account is None:
skipped += 1
done += 1
reason = "别名未匹配账号"
db.mark_skipped(task.id, reason, path=self.db_path)
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
self._emit_progress(done, total, applied, skipped, failed)
continue
try:
db.mark_running(task.id, "apply", path=self.db_path)
self.row_updated.emit(task.id, {"status": "running", "last_error": None})
result = editor.apply_task(
account,
task,
close_success_tab=self.close_success_tab,
)
committed = bool(result.get("committed")) and not result.get("error")
error = result.get("error")
if committed:
db.set_applied(task.id, True, path=self.db_path)
applied += 1
self.row_updated.emit(
task.id,
{
"stage": "applied",
"status": "success",
"committed": 1,
"last_error": None,
},
)
else:
failed += 1
error = error or "更新未提交"
db.set_applied(task.id, False, error, path=self.db_path)
self.failed.emit(task.id, str(error))
self.row_updated.emit(
task.id,
{"status": "failed", "last_error": str(error), "committed": 0},
)
except Exception as exc:
failed += 1
error = str(exc) or exc.__class__.__name__
db.set_applied(task.id, False, error, path=self.db_path)
self.failed.emit(task.id, error)
self.row_updated.emit(
task.id,
{"status": "failed", "last_error": error, "committed": 0},
)
finally:
done += 1
self._emit_progress(done, total, applied, skipped, failed)
return {
"ok": failed == 0,
"total": total,
counters = {
"done": done,
"applied": applied,
"skipped": skipped,
"failed": failed,
"batch_ids": batch_ids,
}
self._run_id = self._create_run_log(eligible, batch_ids)
self._log_run_event(
"运行开始：{mode}，任务{total}，{parallel}".format(
mode="dry-run 预览" if self.dry_run else "真实更新",
total=total,
parallel=(
f"多账号并行最多{self.max_parallel_accounts}"
if self.parallel_accounts
else "串行"
),
)
)
if self.preflight and not self.dry_run:
blocked = self._preflight_block(eligible, account_rows, account_by_alias)
if blocked:
summary = self._summary(
ok=False,
total=total,
counters=counters,
batch_ids=batch_ids,
blocked=True,
extra=blocked,
)
self._finish_run_log("blocked", summary)
return summary
if self.dry_run:
for task in eligible:
if self.should_cancel():
break
outcome = self._preview_task(task, account_by_alias)
self._record_outcome(counters, total, outcome)
elif self.parallel_accounts and self.max_parallel_accounts > 1:
self._run_parallel_by_account(eligible, account_by_alias, counters, total)
else:
for task in eligible:
if self.should_cancel():
break
outcome = self._apply_one_task(task, account_by_alias)
self._record_outcome(counters, total, outcome)
summary = self._summary(
ok=counters["failed"] == 0,
total=total,
counters=counters,
batch_ids=batch_ids,
)
self._finish_run_log("cancelled" if self.should_cancel() else "done", summary)
return summary
def _is_actionable_task(self, task):
return (
@@ -2161,6 +2213,12 @@ if QT_IMPORT_ERROR is None:
"reason": "NO_ACCOUNTS",
"no_accounts": True,
}
duplicate_ports = self._duplicate_debug_ports(account_rows, eligible, account_by_alias)
if duplicate_ports:
return {
"reason": "DUPLICATE_DEBUG_PORT",
"duplicate_ports": duplicate_ports,
}
required_accounts = []
seen_aliases = set()
for task in eligible:
@@ -2188,6 +2246,183 @@ if QT_IMPORT_ERROR is None:
}
return None
def _duplicate_debug_ports(self, account_rows, eligible, account_by_alias):
required_aliases = {
str(task.alias).strip()
for task in eligible
if account_by_alias.get(str(task.alias).strip()) is not None
}
by_port = {}
for account in account_rows:
if account.alias not in required_aliases:
continue
by_port.setdefault(int(account.debug_port), []).append(account)
duplicates = []
for port, rows in by_port.items():
if len(rows) > 1:
duplicates.append(
{
"debug_port": port,
"aliases": [row.alias for row in rows],
}
)
return duplicates
def _run_parallel_by_account(self, eligible, account_by_alias, counters, total):
groups = self._group_tasks_by_alias(eligible)
max_workers = min(self.max_parallel_accounts, len(groups))
if max_workers <= 1:
for group_tasks in groups:
self._run_task_group(group_tasks, account_by_alias, counters, total)
return
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [
executor.submit(
self._run_task_group,
group_tasks,
account_by_alias,
counters,
total,
)
for group_tasks in groups
]
for future in as_completed(futures):
future.result()
def _group_tasks_by_alias(self, tasks):
groups = []
index_by_alias = {}
for task in tasks:
alias = str(task.alias).strip()
if alias not in index_by_alias:
index_by_alias[alias] = len(groups)
groups.append([])
groups[index_by_alias[alias]].append(task)
return groups
def _run_task_group(self, tasks, account_by_alias, counters, total):
for task in tasks:
if self.should_cancel():
break
outcome = self._apply_one_task(task, account_by_alias)
self._record_outcome(counters, total, outcome)
def _preview_task(self, task, account_by_alias):
account = account_by_alias.get(str(task.alias).strip())
if account is None:
reason = "别名未匹配账号"
self._log_run_event(
f"dry-run：任务 {task.id} 商品 {task.item_id} 将略过：{reason}",
task=task,
level="warning",
)
return "skipped"
action_parts = []
if getattr(task, "new_title", None):
action_parts.append("标题")
if getattr(task, "new_cover_path", None):
action_parts.append("封面")
action_text = "+".join(action_parts) or "无变更"
self._log_run_event(
"dry-run：任务 {task_id} 商品 {item_id} 账号 {alias} 将更新 {action}".format(
task_id=task.id,
item_id=task.item_id,
alias=account.alias,
action=action_text,
),
task=task,
)
return "applied"
def _apply_one_task(self, task, account_by_alias):
account = account_by_alias.get(str(task.alias).strip())
if account is None:
reason = "别名未匹配账号"
db.mark_skipped(task.id, reason, path=self.db_path)
self.row_updated.emit(task.id, {"status": "skipped", "last_error": reason})
self._log_run_event(
f"任务 {task.id} 商品 {task.item_id} 已略过：{reason}",
task=task,
level="warning",
)
return "skipped"
try:
self._log_run_event(
f"任务 {task.id} 商品 {task.item_id} 开始更新，账号 {account.alias}",
task=task,
)
db.mark_running(task.id, "apply", path=self.db_path)
self.row_updated.emit(task.id, {"status": "running", "last_error": None})
result = editor.apply_task(
account,
task,
close_success_tab=self.close_success_tab,
)
committed = bool(result.get("committed")) and not result.get("error")
error = result.get("error")
if committed:
db.set_applied(task.id, True, path=self.db_path)
self.row_updated.emit(
task.id,
{
"stage": "applied",
"status": "success",
"committed": 1,
"last_error": None,
},
)
self._log_run_event(
f"任务 {task.id} 商品 {task.item_id} 更新成功",
task=task,
)
return "applied"
error = error or "更新未提交"
db.set_applied(task.id, False, error, path=self.db_path)
self.failed.emit(task.id, str(error))
self.row_updated.emit(
task.id,
{"status": "failed", "last_error": str(error), "committed": 0},
)
self._log_run_event(
f"任务 {task.id} 商品 {task.item_id} 更新失败：{error}",
task=task,
level="error",
)
return "failed"
except Exception as exc:
error = str(exc) or exc.__class__.__name__
db.set_applied(task.id, False, error, path=self.db_path)
self.failed.emit(task.id, error)
self.row_updated.emit(
task.id,
{"status": "failed", "last_error": error, "committed": 0},
)
self._log_run_event(
f"任务 {task.id} 商品 {task.item_id} 更新异常：{error}",
task=task,
level="error",
)
return "failed"
def _record_outcome(self, counters, total, outcome):
with self._progress_lock:
counters["done"] += 1
if outcome == "applied":
counters["applied"] += 1
elif outcome == "skipped":
counters["skipped"] += 1
else:
counters["failed"] += 1
self._emit_progress(
counters["done"],
total,
counters["applied"],
counters["skipped"],
counters["failed"],
)
def _account_payload(self, account, reason=None):
payload = {
"account_name": account.account_name,
@@ -2206,6 +2441,7 @@ if QT_IMPORT_ERROR is None:
"applied": applied,
"skipped": skipped,
"failed": failed,
"dry_run": self.dry_run,
}
)
@@ -2230,6 +2466,77 @@ if QT_IMPORT_ERROR is None:
batch_ids.append(batch_id)
return batch_ids
def _summary(self, ok, total, counters, batch_ids, blocked=False, extra=None):
summary = {
"ok": ok,
"total": total,
"done": counters["done"],
"applied": counters["applied"],
"skipped": counters["skipped"],
"failed": counters["failed"],
"batch_ids": batch_ids,
"dry_run": self.dry_run,
"parallel_accounts": self.parallel_accounts,
"run_id": self._run_id,
}
if blocked:
summary["blocked"] = True
if extra:
summary.update(extra)
return summary
def _create_run_log(self, eligible, batch_ids):
try:
return db.create_run_log(
"apply",
dry_run=self.dry_run,
total=len(eligible),
options={
"batch_ids": batch_ids,
"close_success_tab": self.close_success_tab,
"dry_run": self.dry_run,
"parallel_accounts": self.parallel_accounts,
"max_parallel_accounts": self.max_parallel_accounts,
},
path=self.db_path,
)
except Exception:
return None
def _finish_run_log(self, status, summary):
if self._run_id is None:
return
try:
db.finish_run_log(
self._run_id,
status=status,
done=summary.get("done", 0),
success_count=summary.get("applied", 0),
skipped_count=summary.get("skipped", 0),
failed_count=summary.get("failed", 0),
summary_json=summary,
path=self.db_path,
)
except Exception:
return
def _log_run_event(self, message, task=None, level="info"):
self.log.emit(str(message))
if self._run_id is None:
return
try:
db.add_run_log_event(
self._run_id,
message,
task_id=getattr(task, "id", None),
alias=getattr(task, "alias", None),
item_id=getattr(task, "item_id", None),
level=level,
path=self.db_path,
)
except Exception:
return
class CollectWorker(BaseWorker):
"""Collect old title and cover for imported tasks."""
@@ -2641,6 +2948,13 @@ if QT_IMPORT_ERROR is None:
self.max_items_per_run_spin.setRange(1, 9999)
self.close_success_tab_checkbox = QCheckBox("成功后关闭本次新开编辑页")
self.close_success_tab_checkbox.setObjectName("closeSuccessTabCheckbox")
self.dry_run_checkbox = QCheckBox("dry-run 只预览不提交")
self.dry_run_checkbox.setObjectName("dryRunCheckbox")
self.parallel_accounts_checkbox = QCheckBox("多账号并行更新")
self.parallel_accounts_checkbox.setObjectName("parallelAccountsCheckbox")
self.max_parallel_accounts_spin = QSpinBox()
self.max_parallel_accounts_spin.setObjectName("maxParallelAccountsSpin")
self.max_parallel_accounts_spin.setRange(1, 16)
form = QFormLayout()
form.addRow("", self.enabled_checkbox)
@@ -2687,6 +3001,9 @@ if QT_IMPORT_ERROR is None:
update_form.addRow("", self.allow_cover_update_checkbox)
update_form.addRow("单次最大更新条数", self.max_items_per_run_spin)
update_form.addRow("", self.close_success_tab_checkbox)
update_form.addRow("", self.dry_run_checkbox)
update_form.addRow("", self.parallel_accounts_checkbox)
update_form.addRow("最大并行账号数", self.max_parallel_accounts_spin)
right_panel = QWidget()
right_layout = QVBoxLayout(right_panel)
@@ -2926,6 +3243,9 @@ if QT_IMPORT_ERROR is None:
"allow_cover_update": self.allow_cover_update_checkbox.isChecked(),
"max_items_per_run": self.max_items_per_run_spin.value(),
"close_success_tab": self.close_success_tab_checkbox.isChecked(),
"dry_run": self.dry_run_checkbox.isChecked(),
"parallel_accounts": self.parallel_accounts_checkbox.isChecked(),
"max_parallel_accounts": self.max_parallel_accounts_spin.value(),
},
}
)
@@ -2991,6 +3311,13 @@ if QT_IMPORT_ERROR is None:
self.close_success_tab_checkbox.setChecked(
bool(update_cfg.get("close_success_tab", False))
)
self.dry_run_checkbox.setChecked(bool(update_cfg.get("dry_run", False)))
self.parallel_accounts_checkbox.setChecked(
bool(update_cfg.get("parallel_accounts", False))
)
self.max_parallel_accounts_spin.setValue(
max(1, int(update_cfg.get("max_parallel_accounts", 2) or 2))
)
self._update_response_timeout_label()
def _shopee_update_config(self):