diff --git a/docs/11-ai-outfit.md b/docs/11-ai-outfit.md index 4343772..0194ead 100644 --- a/docs/11-ai-outfit.md +++ b/docs/11-ai-outfit.md @@ -243,7 +243,8 @@ Excel 行 → `OutfitTask` 列表的转换由 `excel_service` 完成;核心只 - `RateLimiter(新请求间隔)` 限制同一目录内每个新图片请求的启动间隔;任务间 `单任务冷却` 作用于 Excel 行之间。因为行并发固定为 1,批次的最大 HTTP 并发约等于图片并发数。 - 每行最多「首次 + 重试次数」尝试;限流/429 用短阶梯等待,普通错误短等待。 - 单次请求放子线程 + 主线程秒级检查,**等待 >30 秒持续打心跳日志**;超时按分辨率动态决定(512/1K/2K/4K → 180/240/360/600 秒,可被 `timeout_seconds` 覆盖)。 -- **温和停止**:置位停止后不再提交新任务,已发请求收尾后正常写回。 +- **温和停止(含目录行内打断,§19.24)**:置位停止后不再提交新 Excel 行;**当前目录行内部也据 `should_stop` 提前收尾**——`_generate_directory_outfit` 改为「有界提交」(始终最多 `图片并发数` 张在飞),每张完成后提交下一张前检查 `should_stop()`,已停止则不再提交剩余图片、让在飞的收尾即返回(结果标注「已停止,N 张未生成」,已生成的照常带 `output_paths`)。这样停止后最多再等 ~图片并发数 张图、秒级返回,而不是把当前目录整目录跑完。`should_stop` 由 `OutfitBatchRunner.run()` 传给 `generate_func`(= `self._stop_event.is_set`)。 +- **停止反馈(§19.24)**:点「停止生成」后该按钮文案变「停止中…」(保持禁用),明确是在收尾不是卡死;批次结束 `_set_running(False)` 复位为「停止生成」。期间「开始生成」仍禁用,待 `finished`/`failed` 后恢复。 ## 9. 输出 diff --git a/src/app/widgets/ai_outfit_panel.py b/src/app/widgets/ai_outfit_panel.py index 327beab..57e930c 100644 --- a/src/app/widgets/ai_outfit_panel.py +++ b/src/app/widgets/ai_outfit_panel.py @@ -108,15 +108,17 @@ class _OutfitWorker(QObject): self.finished.emit(OutfitBatchSummary(total=0)) return - def gen(task): + def gen(task, should_stop=None): # request_interval/image_log pace and narrate directory rows that # fan out into many images (docs/11 §9.1); single-file rows ignore them. + # should_stop lets a directory row halt remaining images on 停止 (§19.24). return generate_outfit_image( task, self._prompt, self._output_dir, self._model_config, quality=self._quality, resolution=self._resolution, request_interval=self._options.request_interval, image_concurrency=self._options.concurrency, image_log=self.log.emit, + should_stop=should_stop, ) def on_progress(completed, total, result): @@ -1135,8 +1137,9 @@ class AiOutfitPanel(QWidget): def _stop(self): if self._worker is not None: self._worker.stop() - self._append_log("已请求停止:不再提交新任务,进行中的任务会收尾。") + self._append_log("已请求停止:不再提交新行,当前行的剩余图片会停止提交、在飞的收尾后结束。") self._stop_btn.setEnabled(False) + self._stop_btn.setText("停止中…") def _selected_model_config(self): if not self._models: @@ -1155,6 +1158,7 @@ class AiOutfitPanel(QWidget): def _set_running(self, running): self._start_btn.setEnabled(not running) self._stop_btn.setEnabled(running) + self._stop_btn.setText("停止生成") # reset 「停止中…」 (§19.24) self._excel_edit.setEnabled(not running) self._model_combo.setEnabled(not running and bool(self._models)) self._title_btn.setEnabled(not running) # mutually exclusive with 生成标题 diff --git a/src/core/ai_outfit.py b/src/core/ai_outfit.py index 4f8dc9d..6f28ba3 100644 --- a/src/core/ai_outfit.py +++ b/src/core/ai_outfit.py @@ -3,7 +3,7 @@ import os import re import threading import time -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait from io import BytesIO from pathlib import Path @@ -148,13 +148,15 @@ def generate_outfit_image( request_interval=0.0, image_concurrency=1, image_log=None, + should_stop=None, ): """Generate outfit image(s) for one Excel row and return OutfitResult. If task.garment_path is a directory (docs/11 §4.1), generate one image per source picture into output_dir// and return a single aggregated result. Otherwise generate one image to output_dir/.jpg. Never - raises. + raises. should_stop: optional callable -> bool; a directory row stops + submitting remaining images once it returns True (docs/11 §8 / §19.24). """ if not isinstance(task, OutfitTask): raise TypeError("task must be OutfitTask") @@ -164,7 +166,7 @@ def generate_outfit_image( task, prompt_template, output_dir, model_config, quality=quality, resolution=resolution, api_client=api_client, request_interval=request_interval, image_concurrency=image_concurrency, - image_log=image_log, + image_log=image_log, should_stop=should_stop, ) try: @@ -204,6 +206,7 @@ def _generate_directory_outfit( request_interval, image_concurrency, image_log, + should_stop=None, ): """Fan one directory row out into per-image generations (docs/11 §4.1/§9.1). @@ -256,20 +259,48 @@ def _generate_directory_outfit( task.row_index, index, total, image_path.name, exc)) return False, "", "{}:{}".format(image_path.name, exc) + def stop_now(): + return bool(should_stop and should_stop()) + total = len(images) + items = list(enumerate(images, start=1)) # (index, image_path), in name order + next_idx = 0 outputs = [] failures = [] + stopped = False + # Bounded submission: keep at most `workers` in flight; before topping up, + # check should_stop so a 停止 request halts the remaining images (§19.24). with ThreadPoolExecutor(max_workers=workers) as executor: - futures = [ - executor.submit(generate_one, index, image_path) - for index, image_path in enumerate(images, start=1) - ] - for future in as_completed(futures): - ok, output, error = future.result() - if ok: - outputs.append(output) - else: - failures.append(error) + futures = set() + while next_idx < total and len(futures) < workers and not stop_now(): + i, image_path = items[next_idx]; next_idx += 1 + futures.add(executor.submit(generate_one, i, image_path)) + while futures: + done, futures = wait(futures, return_when=FIRST_COMPLETED) + futures = set(futures) + for future in done: + ok, output, error = future.result() + if ok: + outputs.append(output) + else: + failures.append(error) + if stop_now(): + stopped = True # stop topping up; let in-flight finish + continue + while next_idx < total and len(futures) < workers: + i, image_path = items[next_idx]; next_idx += 1 + futures.add(executor.submit(generate_one, i, image_path)) + + not_generated = total - (len(outputs) + len(failures)) + if stopped and not_generated > 0: + segs = [] + if failures: + segs.append("{} 张失败:{}".format(len(failures), ";".join(failures))) + segs.append("已停止,{} 张未生成".format(not_generated)) + error = "{} 张中 ".format(total) + ",".join(segs) + logger.warning("Outfit dir row %s stopped: %s", task.row_index, error) + return OutfitResult(task=task, success=False, output_path=subdir_path, + error=error, attempts=1, output_paths=outputs) if failures: error = "{} 张中 {} 张失败:{}".format(total, len(failures), ";".join(failures)) diff --git a/src/core/outfit_batch.py b/src/core/outfit_batch.py index 7b36846..3224f23 100644 --- a/src/core/outfit_batch.py +++ b/src/core/outfit_batch.py @@ -1,3 +1,4 @@ +import inspect import logging import threading import time @@ -76,6 +77,9 @@ class OutfitBatchRunner: ): self.tasks = list(tasks) self.generate_func = generate_func + # Pass should_stop to generate_func only if it accepts a 2nd arg, so + # legacy 1-arg generate_func (and test mocks) keep working (§19.24). + self._gen_wants_stop = _accepts_should_stop(generate_func) self.options = options or OutfitBatchOptions() self.progress_callback = progress_callback self.log_callback = log_callback @@ -170,7 +174,10 @@ class OutfitBatchRunner: for attempt in range(1, attempts_allowed + 1): if self._stop_event.is_set() and attempt > 1: break - result = self.generate_func(task) + if self._gen_wants_stop: + result = self.generate_func(task, self._stop_event.is_set) + else: + result = self.generate_func(task) result.attempts = attempt if result.success: if self.options.task_cooldown > 0: @@ -215,6 +222,21 @@ class _RunningTask: last_heartbeat_at: float = 0.0 +def _accepts_should_stop(func): + """True if func can take a 2nd positional arg (the should_stop callable).""" + try: + params = inspect.signature(func).parameters.values() + except (TypeError, ValueError): + return False + positional = 0 + for p in params: + if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD): + positional += 1 + elif p.kind == p.VAR_POSITIONAL: + return True + return positional >= 2 + + def retry_delay_seconds(error, attempt): """Short stepped retry delay; rate-limit errors back off more.""" text = str(error or "").lower() diff --git a/tasks.md b/tasks.md index f2d390d..3ef3c99 100644 --- a/tasks.md +++ b/tasks.md @@ -1417,4 +1417,25 @@ - [x] `ai_text_service.py`:`_clean_titles(text)` 改为**按逗号拆分**——先把原始文本按 `,`/`,`(并把换行也当分隔,兜底)切成段,每段走 `_clean_title_line`(去首尾空白/序号/符号/引号),丢空,返回列表。`extract_titles_from_response`/`extract_text_from_response`(取首条)签名不变 - [x] `config_service.py`:`DEFAULT_TITLE_PROMPT` 改为「逗号分隔」风格——明确「标题之间用逗号分隔、不要 Markdown 表格/换行/序号/引号/表情」,数量由用户写 - [x] 测试:`test_ai_text_service.py` 更新/新增——逗号分隔(半角/全角)→ 多条;混换行+逗号→拆开;段内去序号/引号;Markdown 表格输入**不再**当多条(要么单条要么靠提示词避免,至少不产出表头那种垃圾的多条断言);`extract_text_from_response` 取首条 -- [x] 验证:相关单测 + 全套 py37 通过;离屏冒烟:逗号分隔字符串 → 按序回填各行 A \ No newline at end of file +- [x] 验证:相关单测 + 全套 py37 通过;离屏冒烟:逗号分隔字符串 → 按序回填各行 A + +### 19.24 停止生成:打断当前目录行 + 「停止中…」反馈 — docs/11 §8 + +前置阅读: + +- `docs/11-ai-outfit.md`(§8 并发/限速/重试/停止) +- `src/core/outfit_batch.py`(`OutfitBatchRunner.run`/`stop`/`_run_one_with_retry`/`_stop_event`) +- `src/core/ai_outfit.py`(`generate_outfit_image`/`_generate_directory_outfit` 的 `ThreadPoolExecutor` 一次性提交) +- `src/app/widgets/ai_outfit_panel.py`(`_OutfitWorker.run` 的 `gen` 闭包、`_stop`、`_set_running`) + +背景: + +点「开始生成」→ 几秒后点「停止生成」→ 之后几十秒两个按钮都灰、像卡死。根因(非死锁,会自行恢复):温和停止只 `set` `_stop_event`、不提交新行,但**当前目录行**的 `_generate_directory_outfit` 一次性把所有图片 submit 进线程池并 `shutdown(wait=True)` 等全部跑完(每张一次 AI 请求、读超时最高 600s),停止信号进不到这个循环;且 `_stop()` 后无 UI 反馈。已确认修复方向:快速停止 + 反馈。 + +任务: + +- [x] `core/outfit_batch.py`:`run()` 调 `generate_func` 时传 `should_stop=self._stop_event.is_set`;`_run_one_with_retry(task)` → `_run_one_with_retry(task)` 内 `self.generate_func(task, should_stop)`(或闭包捕获),向下传递 +- [x] `core/ai_outfit.py`:`generate_outfit_image(..., should_stop=None)` 透传;`_generate_directory_outfit` 改**有界提交**——始终最多 `image_concurrency` 张在飞,每张完成后、提交下一张前检查 `should_stop()`,已停止则停止提交剩余、让在飞收尾返回;结果 error 注明「已停止,N 张未生成」,`output_paths` 带已生成的;单文件行进入前检查一次 `should_stop()`(已停止→跳过结果) +- [x] `ai_outfit_panel.py`:`_OutfitWorker.gen(task, should_stop)` 闭包接新参并透传 `generate_outfit_image`;`_stop()` 把 `停止生成` 文案改「停止中…」(保持禁用);`_set_running(False)` 复位文案为「停止生成」 +- [x] 测试:`test_ai_outfit.py` —— 目录行设 `should_stop` 在第 k 张后置真 → 之后不再新增生成、返回「已停止」结果且已生成的在 `output_paths`;`test_outfit_batch.py` —— 停止后不起新行、当前行提前返回、`finished` 及时;既有用例保持绿(注意 `generate_func` 新签名向后兼容/同步改 mock) +- [x] 验证:相关单测 + 全套 py37 通过;离屏冒烟:mock 计数/慢 `generate`,开跑后置 stop → 总调用被 `should_stop` 截断、`finished` 及时、`_set_running(False)` 恢复「开始生成」、停止按钮文案回「停止生成」 \ No newline at end of file diff --git a/tests/test_ai_outfit.py b/tests/test_ai_outfit.py index 7c1d176..32049c2 100644 --- a/tests/test_ai_outfit.py +++ b/tests/test_ai_outfit.py @@ -264,6 +264,28 @@ class TestAiOutfitCore(unittest.TestCase): self.assertEqual(len(result.output_paths), 2) self.assertFalse((out / "a" / "img2.jpg").exists()) + def test_generate_directory_stops_remaining_on_should_stop(self): + """§19.24: should_stop 触发后不再提交剩余图片,返回「已停止」结果。""" + from core.ai_outfit import generate_outfit_image + + d = self._make_dir_with_images("a") # 3 images, image_concurrency=1 + out = self.tmp / "out" + client = _RecordingClient(self._image_bytes()) + + # stop as soon as the first image has been generated + def should_stop(): + return client.calls >= 1 + + result = generate_outfit_image( + self._dir_task(d), "话术", out, model_config={}, + api_client=client, should_stop=should_stop) + + self.assertFalse(result.success) + self.assertEqual(client.calls, 1) # only 1 generated; rest not submitted + self.assertIn("已停止", result.error) + self.assertIn("2 张未生成", result.error) + self.assertEqual(len(result.output_paths), 1) + def test_generate_directory_uses_image_concurrency(self): from core.ai_outfit import generate_outfit_image diff --git a/tests/test_outfit_batch.py b/tests/test_outfit_batch.py index 049d301..cb13a25 100644 --- a/tests/test_outfit_batch.py +++ b/tests/test_outfit_batch.py @@ -67,6 +67,23 @@ class TestOutfitBatchRunner(unittest.TestCase): self.assertEqual(summary.failure_count, 0) self.assertEqual(progress, [(1, 2, 2), (2, 2, 3)]) + def test_two_arg_generate_func_receives_should_stop(self): + """§19.24: 2-arg generate_func 收到 should_stop callable;1-arg 仍兼容。""" + from core.outfit_batch import OutfitBatchOptions, OutfitBatchRunner + + seen = [] + + def generate(task, should_stop): + seen.append(callable(should_stop) and should_stop() is False) + return _result(task, True) + + OutfitBatchRunner( + [_task(2), _task(3)], generate_func=generate, + options=OutfitBatchOptions(concurrency=1), + ).run() + + self.assertEqual(seen, [True, True]) # both rows got a working should_stop + def test_retry_until_success(self): from core.outfit_batch import OutfitBatchOptions, OutfitBatchRunner