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

fix(product-suite): finalize and cancel generation
Tests / Python 3.11 / Windows (push) Has been cancelled
Details

Browse Source
This commit is contained in:
chengma
2026-07-16 17:12:41 +08:00
parent 2675f85598
commit b68e22c094
11 changed files with 1094 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/gui/tabs/product_suite.py
+393 -27
View File
@@ -5,9 +5,20 @@ from __future__ import annotations
import os
import re
import time
import uuid
from dataclasses import dataclass, field
from PySide6.QtCore import QByteArray, QBuffer, QIODevice, QRect, QSize, Qt, QTimer, Signal
from PySide6.QtCore import (
QByteArray,
QBuffer,
QIODevice,
QRect,
QSize,
Qt,
QTimer,
Signal,
Slot,
)
from PySide6.QtGui import QColor, QIcon, QImage, QImageReader, QKeySequence, QPainter, QPixmap
from PySide6.QtWidgets import (
QApplication,
@@ -688,6 +699,9 @@ class SuiteTaskState:
show_history: bool = False
worker: object = None
thread: object = None
generation_run_token: str = ""
generation_stop_requested: bool = False
generation_terminal_streak: int = 0
pull_worker: object = None
pull_thread: object = None
import_worker: object = None
@@ -739,6 +753,7 @@ class ProductSuiteTab(QWidget):
self._next_serial = 1
self._displayed_state = None
self._prompt_save_timers = {}
self._generation_run_states = {}
self._original_list_context = None
self._loading = False
self._result_refresh_pending = False
@@ -759,6 +774,10 @@ class ProductSuiteTab(QWidget):
self.elapsed_timer.setInterval(1000)
self.elapsed_timer.timeout.connect(self._refresh_elapsed)
self.elapsed_timer.start()
self.generation_watchdog_timer = QTimer(self)
self.generation_watchdog_timer.setInterval(750)
self.generation_watchdog_timer.timeout.connect(self._check_generation_watchdogs)
self.generation_watchdog_timer.start()
if self._prompt_template_init_error:
self._status(self._prompt_template_init_error, "danger")
@@ -1324,6 +1343,7 @@ class ProductSuiteTab(QWidget):
destructive=True,
):
return
state.generation_stop_requested = True
state.worker.cancel()
if state.ai_worker is not None:
state.ai_worker.cancel()
@@ -1347,6 +1367,8 @@ class ProductSuiteTab(QWidget):
except Exception as exc:
self._status("清理空临时草稿失败：%s" % _user_error(exc), "danger")
self._release_prompt_save_timer(state)
if state.generation_run_token:
self._generation_run_states.pop(state.generation_run_token, None)
self._retired_states.append(state)
self._states.pop(state.key, None)
self.task_tabs.removeTab(index)
@@ -1884,6 +1906,10 @@ class ProductSuiteTab(QWidget):
token = id(thread)
_PRODUCT_SUITE_THREAD_REFS[token] = (thread, worker)
thread.finished.connect(lambda token=token: _PRODUCT_SUITE_THREAD_REFS.pop(token, None))
run_token = str(getattr(worker, "run_token", "") or "")
if run_token:
thread.setProperty("productSuiteRunToken", run_token)
thread.finished.connect(self._on_generation_thread_finished_signal)
thread.start()
return thread
@@ -2491,14 +2517,22 @@ class ProductSuiteTab(QWidget):
if state is None:
return
if state.generation_running():
if state.generation_stop_requested:
self._status("正在停止当前套图任务", "warning")
return
if self._confirm(
"停止生成套图",
"确认取消当前任务吗？已提交任务会在安全边界停止。",
destructive=True,
):
state.generation_stop_requested = True
state.worker.cancel()
self.generate_button.setText("正在停止...")
self.generate_button.setEnabled(False)
self._log_generation_lifecycle(
state,
state.generation_run_token,
"stop_requested",
)
self._apply_running_state(state)
self._status("已请求停止当前套图任务", "warning")
return
self.start_generation(state)
@@ -2547,26 +2581,38 @@ class ProductSuiteTab(QWidget):
):
return False
self._persist_state(state)
run_token = uuid.uuid4().hex
worker = ProductSuiteGenerateWorker(
state.project_id,
specs,
run_token=run_token,
aspect_ratio=state.settings["ratio"],
db_path=self.db_path,
config=self.config,
cmhub_config_path=self.cmhub_config_path,
)
state.worker = worker
state.generation_run_token = run_token
state.generation_stop_requested = False
state.generation_terminal_streak = 0
state.done = 0
state.failed = 0
state.total = len(specs)
state.started_at = time.monotonic()
state.current_job_ids = []
state.show_history = False
worker.progress.connect(lambda payload, state=state: self._on_generation_progress(state, payload))
worker.finished.connect(lambda result, state=state: self._on_generation_finished(state, result))
worker.cancelled.connect(lambda result, state=state: self._on_generation_finished(state, result))
worker.failed.connect(lambda row, error, state=state: self._on_generation_failed(state, error))
self._generation_run_states[run_token] = state.key
worker.progress.connect(self._on_generation_progress_signal)
worker.finished.connect(self._on_generation_finished_signal)
worker.cancelled.connect(self._on_generation_finished_signal)
worker.failed.connect(self._on_generation_failed_signal)
state.thread = self._start_thread(worker, "商品套图生成")
self._log_generation_lifecycle(
state,
run_token,
"started",
{"total": len(specs), "job_ids": 0},
)
if state is self._displayed_state:
self._loading = True
try:
@@ -2578,52 +2624,367 @@ class ProductSuiteTab(QWidget):
self._status("商品套图生成已开始，共%d张；可切换到其他任务" % len(specs), "info")
return True
def _generation_signal_token(self, payload=None):
token = str((payload or {}).get("run_token") or "")
sender = self.sender()
return token or str(getattr(sender, "run_token", "") or "")
def _generation_state(self, run_token):
token = str(run_token or "")
state = self._states.get(self._generation_run_states.get(token))
if state is None or state.generation_run_token != token:
return None
return state
@Slot(dict)
def _on_generation_progress_signal(self, payload):
token = self._generation_signal_token(payload)
state = self._generation_state(token)
if state is None:
return
self._on_generation_progress(state, payload)
def _on_generation_progress(self, state, payload):
state.total = int(payload.get("total", state.total) or state.total)
if "total" in payload:
state.total = max(0, int(payload.get("total") or 0))
state.done = int(payload.get("done", state.done) or 0)
state.failed = int(payload.get("failed", state.failed) or 0)
job_ids = payload.get("job_ids")
if job_ids:
if job_ids is not None:
state.current_job_ids = [int(job_id) for job_id in job_ids]
state.generation_terminal_streak = 0
if state is self._displayed_state:
self._refresh_results(state)
self._refresh_elapsed()
def _on_generation_failed(self, state, error):
@Slot(int, str)
def _on_generation_failed_signal(self, row, error):
token = self._generation_signal_token()
state = self._generation_state(token)
if state is None:
return
self._log_generation_lifecycle(
state,
token,
"worker_failed",
{"has_error": True},
level="ERROR",
)
self._status("商品套图生成失败：%s" % _user_error(error), "danger")
def _on_generation_finished(self, state, result):
@Slot(dict)
def _on_generation_finished_signal(self, result):
token = self._generation_signal_token(result)
state = self._generation_state(token)
if state is None:
return
self._log_generation_lifecycle(
state,
token,
"worker_finished",
{
"ok": result.get("ok", True),
"success": result.get("success", 0),
"failed": result.get("failed", 0),
"cancelled": result.get(
"cancelled_count",
result.get("cancelled", 0),
),
},
)
if result.get("ok") is False:
self._reconcile_generation_jobs(state, "生成线程异常结束")
if (
state.generation_stop_requested
and not self._generation_job_ids(state)
and result.get("cancelled") is True
):
result = dict(result)
result["total"] = state.total
result["cancelled_count"] = state.total
self._finalize_generation(state, token, result, source="worker")
@Slot()
def _on_generation_thread_finished_signal(self):
sender = self.sender()
token = str(
sender.property("productSuiteRunToken")
if sender is not None
else ""
)
self._handle_generation_thread_finished(token)
def _handle_generation_thread_finished(self, run_token):
state = self._generation_state(run_token)
if state is None:
return
snapshot = self._generation_job_snapshot(state)
had_active = bool(snapshot["active"])
self._log_generation_lifecycle(
state,
run_token,
"thread_finished_fallback",
snapshot,
level="WARNING" if snapshot["active"] else "INFO",
)
if had_active:
self._reconcile_generation_jobs(state, "生成线程已结束")
snapshot = self._generation_job_snapshot(state)
result = self._generation_result_from_snapshot(snapshot)
if not snapshot["job_ids"] and state.generation_stop_requested:
result.update(
{
"total": state.total,
"cancelled_count": state.total,
}
)
elif had_active:
result.update(
{
"ok": False,
"error": "生成线程已结束，部分任务可稍后继续查询",
}
)
self._finalize_generation(
state,
run_token,
result,
source="thread_finished",
)
def _check_generation_watchdogs(self):
for state in list(self._states.values()):
token = state.generation_run_token
if not token or not state.generation_running():
continue
snapshot = self._generation_job_snapshot(state)
if snapshot["all_terminal"]:
state.generation_terminal_streak += 1
else:
state.generation_terminal_streak = 0
if state.generation_terminal_streak < 2:
continue
self._log_generation_lifecycle(
state,
token,
"terminal_watchdog_finalize",
snapshot,
level="WARNING",
)
self._finalize_generation(
state,
token,
self._generation_result_from_snapshot(snapshot),
source="terminal_watchdog",
)
def _generation_job_ids(self, state):
job_ids = list(state.current_job_ids)
if not job_ids and state.worker is not None:
worker_job_ids = getattr(state.worker, "job_ids", [])
if not isinstance(worker_job_ids, (list, tuple, set)):
worker_job_ids = []
job_ids = [
int(job_id)
for job_id in list(worker_job_ids or [])
]
if job_ids:
state.current_job_ids = job_ids
return job_ids
def _generation_job_snapshot(self, state):
job_ids = self._generation_job_ids(state)
counts = {
"success": 0,
"failed": 0,
"cancelled": 0,
"active": 0,
"job_ids": len(job_ids),
"all_terminal": False,
}
if not job_ids:
counts["active"] = max(0, int(state.total or 0))
return counts
for job_id in job_ids:
try:
job = image_studio.get_job(job_id, path=self.db_path)
except Exception:
job = None
status = str(getattr(job, "status", "") or "")
if status == "succeeded":
counts["success"] += 1
elif status in {"failed", "expired"}:
counts["failed"] += 1
elif status == "cancelled":
counts["cancelled"] += 1
else:
counts["active"] += 1
counts["all_terminal"] = (
len(job_ids) == int(state.total or 0)
and counts["active"] == 0
)
return counts
def _generation_result_from_snapshot(self, snapshot):
return {
"total": int(snapshot.get("job_ids", 0) or 0),
"success": int(snapshot.get("success", 0) or 0),
"failed": int(snapshot.get("failed", 0) or 0),
"cancelled_count": int(snapshot.get("cancelled", 0) or 0),
}
def _reconcile_generation_jobs(self, state, reason):
for job_id in self._generation_job_ids(state):
try:
job = image_studio.get_job(job_id, path=self.db_path)
if job is None or job.status in {
"succeeded",
"failed",
"expired",
"cancelled",
}:
continue
recovery = (
image_studio.JOB_RECOVERY_RESUME
if job.task_id
else image_studio.JOB_RECOVERY_REGENERATE
)
image_studio.update_job_status(
job.id,
"cancelled",
error="%s，任务可稍后继续处理" % reason,
recovery_action=recovery,
path=self.db_path,
)
except Exception as exc:
self._status(
"商品套图任务状态收尾失败：%s" % _user_error(exc),
"danger",
)
def _finalize_generation(self, state, run_token, result, *, source):
if self._generation_state(run_token) is not state:
return False
snapshot = self._generation_job_snapshot(state)
if snapshot["job_ids"] and (
snapshot["all_terminal"]
or source in {"thread_finished", "terminal_watchdog"}
):
result = dict(result or {})
result.update(self._generation_result_from_snapshot(snapshot))
result = dict(result or {})
stop_requested = state.generation_stop_requested
success = int(result.get("success", 0) or 0)
failed = int(result.get("failed", 0) or 0)
cancelled = int(
result.get("cancelled_count", result.get("cancelled", 0)) or 0
)
total = int(result.get("total", state.total) or state.total)
active = max(0, total - success - failed - cancelled)
elapsed = (
int(max(0, time.monotonic() - state.started_at))
if state.started_at
else 0
)
self._generation_run_states.pop(run_token, None)
state.generation_run_token = ""
state.generation_stop_requested = False
state.generation_terminal_streak = 0
state.worker = None
state.thread = None
state.done = int(result.get("success", 0) or 0) + int(result.get("failed", 0) or 0) + int(
result.get("cancelled", 0) or 0
)
state.failed = int(result.get("failed", state.failed) or 0)
state.total = int(result.get("total", state.total) or state.total)
elapsed = int(max(0, time.monotonic() - state.started_at)) if state.started_at else 0
state.done = success + failed + cancelled
state.failed = failed
state.total = total
state.started_at = None
if state is self._displayed_state:
self._apply_running_state(state)
self._refresh_results(state)
self._refresh_elapsed()
self._log_generation_lifecycle(
state,
run_token,
"finalized",
{
"source": source,
"total": total,
"success": success,
"failed": failed,
"cancelled": cancelled,
"active": active,
"elapsed_seconds": elapsed,
},
level="WARNING" if active or result.get("ok") is False else "INFO",
)
if state.key not in self._states:
return
if result.get("ok") is False:
return True
if active or result.get("ok") is False:
message = str(
result.get("error")
or "生成线程已结束，部分任务可稍后继续查询"
)
if state is self._displayed_state:
self._message("商品套图生成失败", _user_error(result.get("error")))
self._message("商品套图生成未完整结束", _user_error(message))
else:
self._status("套图任务%d生成失败" % state.serial, "danger")
return
success = int(result.get("success", 0) or 0)
cancelled = int(result.get("cancelled", 0) or 0)
self._status(
"套图任务%d生成未完整结束" % state.serial,
"danger",
)
return True
if stop_requested or cancelled:
if state is self._displayed_state:
self._message(
"商品套图生成已停止",
"本轮共%d张：成功%d张，失败%d张，停止%d张；"
"已提交任务可稍后继续查询；总用时%d秒。"
% (total, success, failed, cancelled, elapsed),
icon=QMessageBox.Information,
)
self._status(
"商品套图生成已停止：成功%d张，失败%d张，停止%d张"
% (success, failed, cancelled),
"warning",
)
return True
if state is self._displayed_state:
self._message(
"商品套图生成完成",
"本轮共%d张：成功%d张，失败%d张，停止%d张；总用时%d秒。"
% (state.total, success, state.failed, cancelled, elapsed),
% (total, success, failed, cancelled, elapsed),
icon=QMessageBox.Information,
)
self._status("商品套图生成完成：成功%d张，失败%d张" % (success, state.failed), "success")
self._status(
"商品套图生成完成：成功%d张，失败%d张" % (success, failed),
"success",
)
return True
def _log_generation_lifecycle(
self,
state,
run_token,
event,
payload=None,
*,
level="INFO",
):
data = {
"run_token": str(run_token or "")[:8],
"project_id": getattr(state, "project_id", None),
"event": str(event or ""),
}
data.update(dict(payload or {}))
try:
diagnostics.write_diagnostic_log(
"商品套图生成生命周期",
level=level,
step="product_suite_generation",
task_id=getattr(state, "project_id", None),
item_id=getattr(state, "item_id", None),
payload=data,
log_dir=appconfig.diagnostic_log_dir(self.config),
)
except Exception:
pass
def _apply_running_state(self, state):
generation_running = state.generation_running()
@@ -2648,7 +3009,11 @@ class ProductSuiteTab(QWidget):
row.set_controls_enabled(not generation_running)
self.generate_button.setEnabled(True)
if generation_running:
self.generate_button.setText("停止生成")
self.generate_button.setText(
"正在停止..."
if state.generation_stop_requested
else "停止生成"
)
self.generate_button.setStyleSheet(
"QPushButton { background: #cf222e; color: white; border-color: #a40e26; font-weight: 600; }"
"QPushButton:hover { background: #a40e26; }"
@@ -2909,4 +3274,5 @@ class ProductSuiteTab(QWidget):
worker.cancel()
for state in list(self._states.values()) + list(self._retired_states):
self._release_prompt_save_timer(state)
self._generation_run_states.clear()
super().closeEvent(event)