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

feat: add generate progress and readable logs

Browse Source
This commit is contained in:
chengma
2026-07-02 14:02:53 +08:00
parent 4c7d4fc624
commit f9c6dd3331
7 changed files with 337 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/ai.py
+97 -7
View File
@@ -62,6 +62,14 @@ def gen_title(
cfg,
attempts,
request_kind="json",
on_retry=lambda attempt, total_attempts, exc: _notify_retry(
on_step,
"title_request",
attempt,
total_attempts,
exc,
model,
),
)
_notify_step(on_step, "title_parse_response")
text = _extract_text(data).strip()
@@ -110,11 +118,33 @@ def gen_cover(
attempts,
request_kind="multipart",
content_type=content_type,
on_retry=lambda attempt, total_attempts, exc: _notify_retry(
on_step,
"cover_request",
attempt,
total_attempts,
exc,
model,
),
)
else:
payload = _image_chat_payload(model, cover_prompt, old_cover_path, resolution)
_notify_step(on_step, "cover_request")
data = _call_with_retry(model, payload, cfg, attempts, request_kind="json")
data = _call_with_retry(
model,
payload,
cfg,
attempts,
request_kind="json",
on_retry=lambda attempt, total_attempts, exc: _notify_retry(
on_step,
"cover_request",
attempt,
total_attempts,
exc,
model,
),
)
_notify_step(on_step, "cover_parse_response")
image_bytes = _extract_image_bytes(data, model, cfg)
@@ -131,6 +161,23 @@ def _notify_step(callback, step):
pass
def _notify_retry(callback, step, attempt, attempts, exc, model):
if callback is None:
return
try:
callback(
{
"step": step,
"result": "retry",
"attempt": attempt,
"attempts": attempts,
"detail": _redact(str(exc), model),
}
)
except Exception:
pass
def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=None):
"""Generate titles first, then covers, and persist each successful task."""
@@ -186,9 +233,25 @@ def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=No
return step_by_task.get(getattr(task, "id", None), fallback)
def step_callback(task, phase):
def callback(step):
set_step(task, step)
_emit_generation_event(on_event, task, phase, step, "start")
def callback(event):
if isinstance(event, dict):
step = event.get("step") or "unknown"
result = event.get("result") or "start"
set_step(task, step)
_emit_generation_event(
on_event,
task,
phase,
step,
result,
detail=event.get("detail"),
level=event.get("level") or ("warning" if result == "retry" else "info"),
attempt=event.get("attempt"),
attempts=event.get("attempts"),
)
return
set_step(task, event)
_emit_generation_event(on_event, task, phase, event, "start")
return callback
with ThreadPoolExecutor(
@@ -300,7 +363,7 @@ def generate_batch(tasks, prompts, ai_cfg=None, on_progress=None, should_stop=No
"new_cover_path": new_cover_path,
},
)
_emit_generation_event(on_event, task, "cover", "db_write", "success")
_emit_generation_event(on_event, task, "cover", "db_write", "success", detail=new_cover_path)
except CancelledError:
summary["cancelled"] = True
_emit_generation_event(on_event, task, "cover", get_step(task, "cover_request"), "cancelled", level="warning")
@@ -401,7 +464,17 @@ def _mark_generate_failed(task, exc, db_path, on_task_update):
return error
def _emit_generation_event(callback, task, phase, step, result, detail=None, level="info"):
def _emit_generation_event(
callback,
task,
phase,
step,
result,
detail=None,
level="info",
attempt=None,
attempts=None,
):
if callback is None:
return
payload = {
@@ -413,6 +486,10 @@ def _emit_generation_event(callback, task, phase, step, result, detail=None, lev
}
if detail is not None:
payload["detail"] = diagnostics.redact_log_text(detail)
if attempt is not None:
payload["attempt"] = attempt
if attempts is not None:
payload["attempts"] = attempts
try:
callback(payload)
except Exception:
@@ -475,7 +552,15 @@ def _headers(model, content_type):
}
def _call_with_retry(model, body, config, attempts, request_kind, content_type=None):
def _call_with_retry(
model,
body,
config,
attempts,
request_kind,
content_type=None,
on_retry=None,
):
last_exc = None
for index in range(attempts):
try:
@@ -484,6 +569,11 @@ def _call_with_retry(model, body, config, attempts, request_kind, content_type=N
last_exc = exc
if index + 1 >= attempts:
break
if on_retry is not None:
try:
on_retry(index + 1, attempts, exc)
except Exception:
pass
time.sleep(min(2.0, 0.4 * (index + 1)))
raise AIError(
"AI 调用失败（已尝试 %s 次）: %s"
app/gui.py
+167 -15
View File
@@ -30,6 +30,7 @@ try:
QMenu,
QMessageBox,
QPlainTextEdit,
QProgressBar,
QPushButton,
QScrollArea,
QSplitter,
@@ -829,6 +830,22 @@ if QT_IMPORT_ERROR is None:
self.reset_generate_button.setObjectName("resetGenerateButton")
self.stop_generate_button.setEnabled(False)
self.progress_label = QLabel("进度：标题0/0 · 封面0/0 · 失败0")
self.title_progress_label = QLabel("标题 0/0")
self.title_progress_label.setObjectName("generateTitleProgressLabel")
self.title_progress_bar = QProgressBar()
self.title_progress_bar.setObjectName("generateTitleProgressBar")
self.title_progress_bar.setTextVisible(False)
self.title_progress_bar.setRange(0, 1)
self.title_progress_bar.setValue(0)
self.cover_progress_label = QLabel("图片 0/0")
self.cover_progress_label.setObjectName("generateCoverProgressLabel")
self.cover_progress_bar = QProgressBar()
self.cover_progress_bar.setObjectName("generateCoverProgressBar")
self.cover_progress_bar.setTextVisible(False)
self.cover_progress_bar.setRange(0, 1)
self.cover_progress_bar.setValue(0)
self.failed_progress_label = QLabel("失败 0")
self.failed_progress_label.setObjectName("generateFailedProgressLabel")
left_panel = QWidget()
left_layout = QVBoxLayout(left_panel)
@@ -916,12 +933,26 @@ if QT_IMPORT_ERROR is None:
self.splitter.setStretchFactor(1, 3)
self.splitter.setSizes([280, 860])
title_progress_layout = QHBoxLayout()
title_progress_layout.addWidget(self.title_progress_label)
title_progress_layout.addWidget(self.title_progress_bar, 1)
cover_progress_layout = QHBoxLayout()
cover_progress_layout.addWidget(self.cover_progress_label)
cover_progress_layout.addWidget(self.cover_progress_bar, 1)
cover_progress_layout.addWidget(self.failed_progress_label)
progress_layout = QVBoxLayout()
progress_layout.addLayout(title_progress_layout)
progress_layout.addLayout(cover_progress_layout)
button_layout = QHBoxLayout()
button_layout.addStretch(1)
button_layout.addWidget(self.generate_button)
button_layout.addWidget(self.stop_generate_button)
button_layout.addWidget(self.reset_generate_button)
bottom_layout = QHBoxLayout()
bottom_layout.addWidget(self.progress_label)
bottom_layout.addStretch(1)
bottom_layout.addWidget(self.generate_button)
bottom_layout.addWidget(self.stop_generate_button)
bottom_layout.addWidget(self.reset_generate_button)
bottom_layout.addLayout(progress_layout, 1)
bottom_layout.addLayout(button_layout)
layout = QVBoxLayout(self)
layout.setContentsMargins(18, 18, 18, 18)
@@ -961,6 +992,8 @@ if QT_IMPORT_ERROR is None:
def _append_generate_log(self, message):
self.run_log_view.appendPlainText(str(message))
scroll_bar = self.run_log_view.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
def _load_latest_generate_run_log(self):
try:
@@ -972,9 +1005,11 @@ if QT_IMPORT_ERROR is None:
return
lines = [
f"{event.created_at} [{event.level}] {event.message}"
for event in events
for event in reversed(events)
]
self.run_log_view.setPlainText("\n".join(lines))
scroll_bar = self.run_log_view.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
def save_title_prompt(self, checked=False):
try:
@@ -1159,6 +1194,7 @@ if QT_IMPORT_ERROR is None:
def stop_generate(self, checked=False):
if self.generate_worker is not None:
self.generate_worker.cancel()
self._append_generate_log("[停止] 已收到停止请求，当前正在运行的任务结束后停止")
self._set_status("正在停止 AI 生成...")
def reset_generated_result(self, checked=False):
@@ -1312,7 +1348,22 @@ if QT_IMPORT_ERROR is None:
self._set_status("AI 生成已停止：" + self._generate_progress_text(payload))
def _update_generate_progress(self, payload):
total = max(0, int(payload.get("total", 0) or 0))
title_done = max(0, int(payload.get("title_done", 0) or 0))
cover_done = max(0, int(payload.get("cover_done", 0) or 0))
failed = max(0, int(payload.get("failed", 0) or 0))
self.progress_label.setText("进度：" + self._generate_progress_text(payload))
self.title_progress_label.setText(f"标题 {title_done}/{total}")
self.cover_progress_label.setText(f"图片 {cover_done}/{total}")
self.failed_progress_label.setText(f"失败 {failed}")
self._set_progress_bar(self.title_progress_bar, title_done, total)
self._set_progress_bar(self.cover_progress_bar, cover_done, total)
def _set_progress_bar(self, bar, done, total):
maximum = max(1, int(total or 0))
value = min(max(0, int(done or 0)), maximum)
bar.setRange(0, maximum)
bar.setValue(value)
def _generate_progress_text(self, payload):
return "标题{title}/{total} · 封面{cover}/{total} · 失败{failed}".format(
@@ -2108,9 +2159,11 @@ if QT_IMPORT_ERROR is None:
return
lines = [
f"{event.created_at} [{event.level}] {event.message}"
for event in events
for event in reversed(events)
]
self.run_log_view.setPlainText("\n".join(lines))
scroll_bar = self.run_log_view.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
def _show_apply_blocked(self, payload):
lines = ["更新前检查未通过。"]
@@ -2453,9 +2506,11 @@ if QT_IMPORT_ERROR is None:
return
lines = [
f"{event.created_at} [{event.level}] {event.message}"
for event in events
for event in reversed(events)
]
self.run_log_view.setPlainText("\n".join(lines))
scroll_bar = self.run_log_view.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
def _log_collect_run_event(self, run_id, message, level="info"):
safe_message = diagnostics.redact_log_text(message)
@@ -3215,6 +3270,9 @@ if QT_IMPORT_ERROR is None:
self.config = config
self.diagnostic_log_dir = diagnostic_log_dir
self._run_id = None
self._account_by_alias = {}
self._task_positions = {}
self._eligible_total = 0
def execute(self):
account_rows = accounts.list_accounts(path=self.db_path, config=self.config)
@@ -3223,14 +3281,25 @@ if QT_IMPORT_ERROR is None:
for account in account_rows
if str(account.alias).strip()
}
self._account_by_alias = account_by_alias
eligible = [
task for task in self.tasks
if getattr(task, "stage", None) == "collected"
]
self._eligible_total = len(eligible)
self._task_positions = {
getattr(task, "id", None): index
for index, task in enumerate(eligible, start=1)
}
batch_ids = self._batch_ids(eligible)
self._run_id = self._create_run_log(eligible, batch_ids)
ai_cfg = appconfig.ai_config(self.config)
self._log_run_event(
f"phase=preflight step=start result=start detail=AI生成开始 total={len(eligible)}"
"[开始] 本轮生成 {total} 条：标题{total}，图片{total}；标题并发{title_concurrency}，图片并发{image_concurrency}".format(
total=len(eligible),
title_concurrency=ai_cfg.get("title_concurrency", 1),
image_concurrency=ai_cfg.get("image_concurrency", 1),
)
)
try:
summary = ai.generate_batch(
@@ -3260,7 +3329,7 @@ if QT_IMPORT_ERROR is None:
"cancelled": self.should_cancel(),
}
self._log_run_event(
f"phase=worker step=execute result=failed detail={error}",
f"[失败] AI 生成运行失败：{error}",
level="error",
)
self._write_diagnostic_log(
@@ -3273,6 +3342,8 @@ if QT_IMPORT_ERROR is None:
summary["run_id"] = self._run_id
summary["batch_ids"] = batch_ids
status = "cancelled" if summary.get("cancelled") else "done"
level = "warning" if summary.get("cancelled") or summary.get("error") else "info"
self._log_run_event(self._format_generate_completion(summary), level=level)
self._finish_run_log(status, summary)
return summary
@@ -3280,15 +3351,96 @@ if QT_IMPORT_ERROR is None:
self.row_updated.emit(int(task_id), dict(fields or {}))
def _on_generation_event(self, payload):
task = payload.get("task")
message = self._format_generation_event(payload)
if not message:
return
self._log_run_event(message, task=task, level=payload.get("level") or "info")
def _format_generation_event(self, payload):
task = payload.get("task")
phase = payload.get("phase") or "generate"
step = payload.get("step") or "unknown"
result = payload.get("result") or "start"
detail = payload.get("detail")
message = f"phase={phase} step={step} result={result}"
if detail:
message += f" detail={detail}"
self._log_run_event(message, task=task, level=payload.get("level") or "info")
detail = self._short_detail(payload.get("detail"))
if phase == "title":
if result == "start" and step == "title_submit":
return f"[标题] {self._task_progress_label(task)} 开始生成"
if result == "success" and step == "title_done":
return f"[标题] {self._task_progress_label(task)} 成功"
if result == "retry":
return self._retry_message("标题", task, payload, detail)
if result == "failed":
return f"[失败] {self._task_plain_label(task)} 标题生成失败：{detail or '未知错误'}"
if result == "cancelled":
return f"[停止] {self._task_plain_label(task)} 标题生成已取消"
return None
if phase == "cover":
if result == "start" and step == "cover_submit":
return f"[图片] {self._task_progress_label(task)} 开始生成"
if result == "success" and step == "db_write":
suffix = f"，已保存 {detail}" if detail else ""
return f"[图片] {self._task_progress_label(task)} 成功{suffix}"
if result == "retry":
return self._retry_message("图片", task, payload, detail)
if result == "failed":
return f"[失败] {self._task_plain_label(task)} 图片生成失败：{detail or '未知错误'}"
if result == "cancelled":
return f"[停止] {self._task_plain_label(task)} 图片生成已取消"
return None
return None
def _retry_message(self, label, task, payload, detail):
attempt = int(payload.get("attempt", 0) or 0)
attempts = int(payload.get("attempts", 0) or 0)
max_retries = max(0, attempts - 1)
retry_text = f"准备重试 {attempt}/{max_retries}" if max_retries else "准备重试"
reason = f"：{detail}" if detail else ""
return f"[{label}] {self._task_progress_label(task)} 调用失败，{retry_text}{reason}"
def _task_progress_label(self, task):
index = self._task_positions.get(getattr(task, "id", None), 0)
total = self._eligible_total or 0
item_id = getattr(task, "item_id", "") or "未知商品"
shop = self._task_shop_label(task)
shop_text = f"（{shop}）" if shop else ""
return f"{index}/{total} 商品 {item_id}{shop_text}"
def _task_plain_label(self, task):
item_id = getattr(task, "item_id", "") or "未知商品"
shop = self._task_shop_label(task)
return f"商品 {item_id}（{shop}）" if shop else f"商品 {item_id}"
def _task_shop_label(self, task):
alias = str(getattr(task, "alias", "") or "").strip()
account = self._account_by_alias.get(alias)
if account is not None:
return getattr(account, "account_name", None) or getattr(account, "alias", None) or alias
return getattr(task, "account_name", None) or alias
def _short_detail(self, detail):
if detail is None:
return ""
text = diagnostics.redact_log_text(str(detail)).replace("\r", " ").replace("\n", " ").strip()
if len(text) > 180:
return text[:177] + "..."
return text
def _format_generate_completion(self, summary):
progress = self._summary_text(summary)
if summary.get("cancelled"):
return f"[停止] AI 生成已停止：{progress}"
if summary.get("error"):
return f"[失败] AI 生成失败：{summary.get('error')}，{progress}"
return f"[完成] AI 生成完成：{progress}"
def _summary_text(self, summary):
return "标题{title}/{total}，图片{cover}/{total}，失败{failed}".format(
title=summary.get("title_done", 0),
cover=summary.get("cover_done", 0),
total=summary.get("total", 0),
failed=summary.get("failed", 0),
)
def _on_generation_error(self, payload):
task = payload.get("task")