From abe46c4aebb8eb17210430663cc7b5d079fcb1f8 Mon Sep 17 00:00:00 2001 From: ila Date: Mon, 22 Jun 2026 17:03:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(ai-outfit):=20C=20=E5=88=97=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E5=9B=BE=E7=89=87=E7=9B=AE=E5=BD=95=20=E2=86=92=20?= =?UTF-8?q?=E5=A4=9A=E5=9B=BE=E6=89=87=E5=87=BA=E5=88=B0=E5=90=8C=E5=90=8D?= =?UTF-8?q?=E5=AD=90=E7=9B=AE=E5=BD=95=20(=C2=A719.12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Excel C 列除单张图片外也可填一个目录（d:/images/a/）：对目录内每张图各 生成一张穿搭图（N→N），全部存到 输出目录/<目录叶子名>/，文件名沿用源图名； Excel 仍整行一个状态：D=子目录、E=全成功才「完成」、F=失败张数/原因。 - core/ai_outfit.py: list_directory_images（顶层、扩展名过滤、排序、忽略子目录）、 make_outfit_subdir_path（不加 _n 后缀）、generate_outfit_image 目录分支 （新参 request_interval/image_log；逐张跳过已存在→幂等重试、本地节流、发日志、 聚合成单个 OutfitResult） - core/models.py: OutfitResult 加 output_paths（目录行各 jpg，供缩略图） - ai_outfit_panel.py: gen 闭包传 request_interval/image_log；缩略图逐张； 明细「结果」列显示「子目录（N 张）」；_basename 处理目录末尾分隔符 - tests/test_ai_outfit.py: 扇出/幂等跳过/空目录/缺目录/部分失败/命名过滤等用例 - docs/11 §4.1+§9.1、tasks.md §19.12 离屏冒烟：3 图目录 + mock API → output// 3 jpg、D=子目录、E=完成、缩略图 3 张； 全套 12 测试文件在 Python 3.7 全绿。 Co-Authored-By: Claude Opus 4.8 --- docs/11-ai-outfit.md | 26 +++++ src/app/widgets/ai_outfit_panel.py | 30 ++++-- src/core/ai_outfit.py | 134 ++++++++++++++++++++++++- src/core/models.py | 7 +- tasks.md | 15 +++ tests/test_ai_outfit.py | 152 +++++++++++++++++++++++++++++ 6 files changed, 354 insertions(+), 10 deletions(-) diff --git a/docs/11-ai-outfit.md b/docs/11-ai-outfit.md index 3c79aa4..9da62d9 100644 --- a/docs/11-ai-outfit.md +++ b/docs/11-ai-outfit.md @@ -71,6 +71,18 @@ services/excel_service：把结果写回 Excel（D 新图路径 / E 状态 / F - **每处理完一行即保存 Excel**（降低崩溃丢结果风险）。 - 成功 → D=新图绝对路径、E=`完成`；失败 → E=`失败`、F=原因（同步写日志）。 +### 4.1 C 列为「图片目录」（多图扇出，§9.1） + +C 列除了单张图片文件，**也可以是一个目录**（如 `d:/images/a/`）。约定： + +- 判定：`os.path.isdir(C)` 为真，或路径以分隔符结尾。仅取该目录**顶层**的图片 + （扩展名 `.png/.jpg/.jpeg/.webp/.gif`），**不递归**子目录，按文件名排序。 +- 该行仍是**一个任务、一次回写**；对目录内**每一张**图各生成一张穿搭图（N→N）， + 共用本行的标题/货号与话术。 +- 输出落到 `输出目录/<目录叶子名>/`（见 §9.1）；Excel 回写仍是**整行一个状态**： + D=子目录绝对路径、E=全部成功才 `完成` 否则 `失败`、F=失败张数/原因。 +- 目录不存在或目录内没有图片 → 该行 `失败` 并记录原因，不中断其它行。 + ## 5. 内部数据模型 `core/models.py` 新增（纯 dataclass，Python 3.7 兼容，不依赖 PySide6）： @@ -165,6 +177,20 @@ Excel 行 → `OutfitTask` 列表的转换由 `excel_service` 完成；核心只 - 命名：`货号.jpg`，重名自动 `_1`/`_2`，非法字符替换为 `_`（不改 Excel 原始货号）。 - 成功后把**实际新图绝对路径**写回 Excel D 列。 +### 9.1 目录行的输出（多图 → 同名子目录） + +当 C 列是目录（§4.1）时： + +- 每张源图各生成一张穿搭图，存到 `输出目录/<目录叶子名>/<源图名>.jpg` + （子目录名 = 目录叶子名，文件名沿用源图名；二者均按 §9 规则替换非法字符）。 + 例：`d:/images/a/img1.png` → `输出目录/a/img1.jpg`。 +- **不加 `_1`/`_2` 去重后缀**：目标文件已存在视为「已生成」并**跳过**，使整行可 + 幂等重试——重试只补做缺失/失败的那几张，已成功的不重复调 API。 +- 整行回写 Excel 时，D 列写**子目录**绝对路径（而非单个文件）。 +- 目录内逐张**顺序**生成（同一 worker 内）；为避免压垮中转 API，逐张之间按 + 「新请求间隔」本地 sleep 节流。并发=1（默认）时即等于全局节流；并发>1 时为近似。 + 进度条按 Excel 行前进（一个目录=1 格），逐张进度通过实时日志反馈。 + ## 10. 界面（「2 AI 穿搭」页签） 界面效果图见 **`docs/ui-ai-outfit.html`**（浏览器打开）/ **`docs/ui-ai-outfit.png`**，完全沿用 cmbot 现有视觉、与「1 添加印花」严格统一（同 `docs/ui-v1`：浅灰底、`#0067c0` 蓝单一主色、12px 雅黑、3px 圆角、pill 状态徽章；不引入第二识别色）。「生成中」状态用蓝，与印花页「导出中」同色。把主窗口当前禁用的「2 AI 穿搭」页签启用，做成独立工作页（与「1 添加印花」并列、互不干扰）。 diff --git a/src/app/widgets/ai_outfit_panel.py b/src/app/widgets/ai_outfit_panel.py index 7c89dfd..234eb88 100644 --- a/src/app/widgets/ai_outfit_panel.py +++ b/src/app/widgets/ai_outfit_panel.py @@ -106,9 +106,13 @@ class _OutfitWorker(QObject): return def gen(task): + # request_interval/image_log pace and narrate directory rows that + # fan out into many images (docs/11 §9.1); single-file rows ignore them. 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_log=self.log.emit, ) def on_progress(completed, total, result): @@ -870,7 +874,11 @@ class AiOutfitPanel(QWidget): if row is not None: if result.success: self._set_cell(row, 4, "完成") - self._set_cell(row, 5, result.output_path) + if result.output_paths: # 目录行：子目录 + 张数 + self._set_cell(row, 5, "{}（{} 张）".format( + result.output_path, len(result.output_paths))) + else: + self._set_cell(row, 5, result.output_path) else: self._set_cell(row, 4, "失败") self._set_cell(row, 5, result.error) @@ -936,12 +944,17 @@ class AiOutfitPanel(QWidget): # -- helpers -------------------------------------------------------- def _add_result_thumb(self, result): - pix = QPixmap(result.output_path) - item = QListWidgetItem(result.task.product_id) - if not pix.isNull(): - item.setIcon(QIcon(pix)) - item.setData(Qt.UserRole, result.output_path) - self._results.insertItem(0, item) + # Directory rows produce several images (output_paths); single-file rows + # one (output_path). Add a thumbnail for each (docs/11 §9.1). + for path in (result.output_paths or [result.output_path]): + if not path: + continue + pix = QPixmap(path) + item = QListWidgetItem(result.task.product_id) + if not pix.isNull(): + item.setIcon(QIcon(pix)) + item.setData(Qt.UserRole, path) + self._results.insertItem(0, item) def _open_result(self, item): path = item.data(Qt.UserRole) @@ -991,7 +1004,8 @@ class AiOutfitPanel(QWidget): def _basename(path): import os - return os.path.basename(str(path)) + # normpath so a directory path "d:/images/a/" shows its leaf "a" (docs/11 §4.1). + return os.path.basename(os.path.normpath(str(path))) def _csv(value): diff --git a/src/core/ai_outfit.py b/src/core/ai_outfit.py index d330432..a6873eb 100644 --- a/src/core/ai_outfit.py +++ b/src/core/ai_outfit.py @@ -1,5 +1,7 @@ import logging +import os import re +import time from io import BytesIO from pathlib import Path @@ -24,6 +26,7 @@ QUALITY_PRESETS = { MAX_JPG_BYTES = 2 * 1024 * 1024 _INVALID_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') +_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".gif"} # Auto-appended to every prompt (ported from 标题生成产品图工具; docs/11 §7.1). # Only 参考解析度 is dynamic (the chosen resolution); the rest is fixed. @@ -74,6 +77,40 @@ def make_outfit_output_path(output_dir, product_id): return candidate +def looks_like_directory(path): + """Return True if *path* should be treated as an image directory (docs/11 §4.1).""" + text = str(path) + if os.path.isdir(text): + return True + return text.endswith("/") or text.endswith("\\") + + +def list_directory_images(dir_path): + """Return top-level image files in *dir_path*, sorted by filename. + + Non-recursive; only known image extensions; subdirectories are ignored + (docs/11 §4.1). + """ + directory = Path(dir_path) + images = [] + for entry in sorted(directory.iterdir(), key=lambda p: p.name.lower()): + if entry.is_file() and entry.suffix.lower() in _IMAGE_EXTS: + images.append(entry) + return images + + +def make_outfit_subdir_path(output_dir, subdir_name, source_stem): + """Return output_dir//.jpg, no de-dup suffix. + + Used with skip-existing so a directory row can be retried idempotently + (docs/11 §9.1). Both parts are run through safe_product_filename so illegal + characters cannot escape the output tree. + """ + safe_sub = safe_product_filename(subdir_name) + safe_stem = safe_product_filename(source_stem) + return Path(output_dir) / safe_sub / (safe_stem + ".jpg") + + def save_jpg_under_limit(image_bytes, output_path, quality=QUALITY_BALANCED, max_bytes=MAX_JPG_BYTES): """Save image bytes as 1:1 JPG, reducing quality/size until under limit.""" output_path = Path(output_path) @@ -106,11 +143,26 @@ def generate_outfit_image( quality=QUALITY_BALANCED, resolution="1K", api_client=None, + request_interval=0.0, + image_log=None, ): - """Generate one outfit image and return OutfitResult. Never raises.""" + """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. + """ if not isinstance(task, OutfitTask): raise TypeError("task must be OutfitTask") + if looks_like_directory(task.garment_path): + return _generate_directory_outfit( + task, prompt_template, output_dir, model_config, + quality=quality, resolution=resolution, api_client=api_client, + request_interval=request_interval, image_log=image_log, + ) + try: prompt = render_prompt(prompt_template, task, resolution=resolution) client = api_client or ImageApiClient(model_config) @@ -134,6 +186,86 @@ def generate_outfit_image( ) +def _generate_directory_outfit( + task, + prompt_template, + output_dir, + model_config, + quality, + resolution, + api_client, + request_interval, + image_log, +): + """Fan one directory row out into per-image generations (docs/11 §4.1/§9.1). + + Each source image -> output_dir//.jpg. Existing outputs + are skipped so a failed row can be retried idempotently. Returns one + aggregated OutfitResult: output_path = the subdirectory (Excel D), output_paths + = each produced jpg, success only when every image succeeded. + """ + def emit(message): + logger.info(message) + if image_log is not None: + try: + image_log(message) + except Exception: # noqa: BLE001 - logging must not break the run + pass + + directory = task.garment_path + if not os.path.isdir(directory): + return OutfitResult(task=task, success=False, + error="目录不存在：{}".format(directory), attempts=1) + + images = list_directory_images(directory) + if not images: + return OutfitResult(task=task, success=False, + error="目录内没有图片：{}".format(directory), attempts=1) + + subdir_name = os.path.basename(os.path.normpath(directory)) + subdir_path = str(Path(output_dir) / safe_product_filename(subdir_name)) + prompt = render_prompt(prompt_template, task, resolution=resolution) + client = api_client or ImageApiClient(model_config) + interval = float(request_interval or 0.0) + + total = len(images) + outputs = [] + failures = [] + called = False + for index, image_path in enumerate(images, start=1): + output_path = make_outfit_subdir_path(output_dir, subdir_name, image_path.stem) + if output_path.exists(): + outputs.append(str(output_path)) + emit("第 {} 行 第 {}/{} 张已存在，跳过：{}".format( + task.row_index, index, total, image_path.name)) + continue + try: + if called and interval > 0: + time.sleep(interval) + image_bytes = client.generate(prompt, str(image_path), resolution=resolution) + called = True + save_jpg_under_limit(image_bytes, output_path, quality=quality) + outputs.append(str(output_path)) + emit("第 {} 行 第 {}/{} 张完成：{}".format( + task.row_index, index, total, image_path.name)) + except Exception as exc: # noqa: BLE001 - record and continue + called = True + failures.append("{}：{}".format(image_path.name, exc)) + emit("第 {} 行 第 {}/{} 张失败：{}（{}）".format( + task.row_index, index, total, image_path.name, exc)) + + if failures: + error = "{} 张中 {} 张失败：{}".format(total, len(failures), "；".join(failures)) + logger.warning("Outfit dir row %s partial failure: %s", task.row_index, error) + return OutfitResult(task=task, success=False, output_path=subdir_path, + error=error, attempts=1, output_paths=outputs) + + logger.info("Generated outfit dir row %s -> %s (%d images)", + task.row_index, subdir_path, len(outputs)) + return OutfitResult(task=task, success=True, output_path=subdir_path, + attempts=1, output_paths=outputs) + + def _coerce_quality(quality): if isinstance(quality, str): return QUALITY_PRESETS.get(quality, QUALITY_BALANCED) diff --git a/src/core/models.py b/src/core/models.py index 2757616..24edb26 100644 --- a/src/core/models.py +++ b/src/core/models.py @@ -169,12 +169,17 @@ class OutfitTask: @dataclass class OutfitResult: - """AI 穿搭单行生成结果。""" + """AI 穿搭单行生成结果。 + + output_path: 单文件行 = 结果图路径；目录行 = 输出子目录路径（写回 Excel D 列）。 + output_paths: 目录行下每张结果图的路径（供缩略图逐张展示）；单文件行留空。 + """ task: OutfitTask success: bool output_path: str = "" error: str = "" attempts: int = 0 + output_paths: List[str] = field(default_factory=list) # --------------------------------------------------------------------------- diff --git a/tasks.md b/tasks.md index 8bdd6d4..ed26a1f 100644 --- a/tasks.md +++ b/tasks.md @@ -1145,3 +1145,18 @@ - [x] 主操作蓝（`#openFolderBtn`/`#queueBatchBtn` 同款）= 开始生成（`aiStartBtn`）；危险安静红（`#templateDeleteBtn` 同款）= 停止生成（`aiStopBtn`）+ 话术删除（`aiPromptDeleteBtn`）；其余走面板内 `QPushButton` 默认次级灰（`#queueActionBtn`/`#templateBtn` 同款） - [x] 给 开始/停止/话术删除 三个按钮设 objectName；纯样式、不动行为 - [x] 离屏 Qt grab 截图核对：开始=蓝、停止=红(禁用→灰)、话术删除=红、其余=灰描边；面板 stylesheet 非空；全套 12 文件绿 + +### 19.12 C 列支持「图片目录」→ 多图扇出到同名子目录 — docs/11 §4.1 / §9.1 + +前置阅读：`docs/11-ai-outfit.md`（§4.1、§9.1）、`src/core/ai_outfit.py`（`generate_outfit_image`/`make_outfit_output_path`/`safe_product_filename`）、`src/core/outfit_batch.py`、`src/app/widgets/ai_outfit_panel.py`（`_OutfitWorker`/`_add_result_thumb`/`_on_progress`/`_basename`） + +背景：C 列除单张图片文件外，也可填一个目录（`d:/images/a/`）。对目录内每张图各生成一张穿搭图（N→N），全部存到 `输出目录/a/`，文件名沿用源图名；Excel 仍整行一个状态：D=子目录、E=全成功才「完成」、F=失败张数/原因。已确认：① N→N；② D 写子目录路径、整行一状态；③ 沿用源图名。 + +设计取舍：保持「一行=一个 OutfitTask=一个 worker=一次回写」，多图扇出放进 `generate_outfit_image` 内部 → 批处理器/进度/明细表/回写几乎不动。目录内顺序生成、逐张按「新请求间隔」本地节流（并发=1 即全局节流）；失败重试跳过已存在输出，幂等。 + +- [x] `core/ai_outfit.py`：`list_directory_images(dir)`（顶层、扩展名过滤、排序、忽略子目录）；`make_outfit_subdir_path(output_dir, subdir, stem)`（`输出目录/<安全子目录>/<安全源图名>.jpg`，不加 `_n` 后缀） +- [x] `core/ai_outfit.py`：`generate_outfit_image` 加目录分支（新参 `request_interval`/`image_log`）——列图片、空目录→失败、复用一个 `ImageApiClient`、逐张跳过已存在/节流/存子目录/发日志、聚合成单个 `OutfitResult`（`output_path`=子目录、`output_paths`=各 jpg、`success`=全成功且≥1、`error`=失败张数） +- [x] `core/models.py`：`OutfitResult` 加 `output_paths: List[str]`（单文件模式留空） +- [x] `ai_outfit_panel.py`：`gen` 闭包传 `request_interval`/`image_log=self.log.emit`；`_add_result_thumb` 逐张加缩略图；`_on_progress` 明细「结果」列显示「子目录 (N 张)」；`_basename` 处理目录末尾分隔符 +- [x] `tests/test_ai_outfit.py`：扇出/幂等跳过/空目录/部分失败/命名过滤等用例；既有单文件用例保持绿；全套 12 文件 py37 全绿 +- [x] 离屏冒烟：临时 Excel 的 C 指向含 3 张图的目录 + mock API，断言 `输出/<目录名>/` 下 3 个 jpg、D=子目录、E=完成、缩略图 3 张 diff --git a/tests/test_ai_outfit.py b/tests/test_ai_outfit.py index 93b824c..1957fee 100644 --- a/tests/test_ai_outfit.py +++ b/tests/test_ai_outfit.py @@ -1,4 +1,5 @@ """Tests for single-row AI outfit generation core.""" +import os import shutil import sys import tempfile @@ -123,6 +124,128 @@ class TestAiOutfitCore(unittest.TestCase): self.assertFalse(result.success) self.assertIn("boom", result.error) + # -- directory rows (docs/11 §4.1 / §9.1) --------------------------- + + def _make_image_file(self, path, color=(10, 20, 30)): + Image.new("RGB", (64, 64), color).save(str(path), format="PNG") + + def _dir_task(self, garment_path, product_id="DIRA"): + from core.models import OutfitTask + + return OutfitTask(row_index=3, title="目录款", product_id=product_id, + garment_path=str(garment_path)) + + def _make_dir_with_images(self, name="a", files=("img1.png", "img2.png", "img3.png")): + d = self.tmp / name + d.mkdir() + for fname in files: + self._make_image_file(d / fname) + return d + + def test_list_directory_images_filters_sorts_ignores_subdirs(self): + from core.ai_outfit import list_directory_images + + d = self.tmp / "imgs" + d.mkdir() + self._make_image_file(d / "b.png") + self._make_image_file(d / "a.jpg") + (d / "note.txt").write_text("x", encoding="utf-8") + (d / "sub").mkdir() + self._make_image_file(d / "sub" / "c.png") + + images = list_directory_images(d) + + self.assertEqual([p.name for p in images], ["a.jpg", "b.png"]) + + def test_make_outfit_subdir_path_sanitizes_without_suffix(self): + from core.ai_outfit import make_outfit_subdir_path + + p = make_outfit_subdir_path(self.tmp, "a:b", "img/1") + + self.assertEqual(p.parent.name, "a_b") + self.assertEqual(p.name, "img_1.jpg") + + def test_generate_directory_fans_out_to_named_subdir(self): + from core.ai_outfit import generate_outfit_image + + d = self._make_dir_with_images("a") + out = self.tmp / "out" + client = _RecordingClient(self._image_bytes()) + + result = generate_outfit_image( + self._dir_task(d), "话术 {title}", out, + model_config={}, api_client=client, + ) + + self.assertTrue(result.success, result.error) + self.assertEqual(client.calls, 3) + self.assertEqual(Path(result.output_path), out / "a") + self.assertEqual(len(result.output_paths), 3) + names = sorted(p.name for p in (out / "a").iterdir()) + self.assertEqual(names, ["img1.jpg", "img2.jpg", "img3.jpg"]) + + def test_generate_directory_skips_existing_outputs_on_retry(self): + from core.ai_outfit import generate_outfit_image + + d = self._make_dir_with_images("a") + out = self.tmp / "out" + first = generate_outfit_image( + self._dir_task(d), "x {title}", out, + model_config={}, api_client=_RecordingClient(self._image_bytes())) + self.assertTrue(first.success) + + # Re-run: every output already exists -> no API calls, still success. + again_client = _RecordingClient(self._image_bytes()) + again = generate_outfit_image( + self._dir_task(d), "x {title}", out, + model_config={}, api_client=again_client) + + self.assertTrue(again.success) + self.assertEqual(again_client.calls, 0) + self.assertEqual(len(again.output_paths), 3) + + def test_generate_directory_empty_fails(self): + from core.ai_outfit import generate_outfit_image + + d = self.tmp / "empty" + d.mkdir() + + result = generate_outfit_image( + self._dir_task(d), "x", self.tmp / "out", + model_config={}, api_client=_RecordingClient(self._image_bytes())) + + self.assertFalse(result.success) + self.assertIn("没有图片", result.error) + + def test_generate_directory_missing_fails(self): + from core.ai_outfit import generate_outfit_image + + # Trailing separator marks it as a directory even though it doesn't exist. + missing = str(self.tmp / "nope") + os.sep + result = generate_outfit_image( + self._dir_task(missing), "x", self.tmp / "out", + model_config={}, api_client=_RecordingClient(self._image_bytes())) + + self.assertFalse(result.success) + self.assertIn("目录不存在", result.error) + + def test_generate_directory_partial_failure_aggregates(self): + from core.ai_outfit import generate_outfit_image + + d = self._make_dir_with_images("a") + out = self.tmp / "out" + client = _FailOnClient(self._image_bytes(), fail_name="img2.png") + + result = generate_outfit_image( + self._dir_task(d), "x", out, model_config={}, api_client=client) + + self.assertFalse(result.success) + self.assertIn("3 张中 1 张失败", result.error) + self.assertIn("img2.png", result.error) + # The two that succeeded are still written (and listed for thumbnails). + self.assertEqual(len(result.output_paths), 2) + self.assertFalse((out / "a" / "img2.jpg").exists()) + class _FakeClient: def __init__(self, image_bytes): @@ -141,5 +264,34 @@ class _FailingClient: raise RuntimeError("boom") +class _RecordingClient: + """Records every generate() call (count + image paths) for directory tests.""" + + def __init__(self, image_bytes): + self._image_bytes = image_bytes + self.calls = 0 + self.image_paths = [] + + def generate(self, prompt, image_path, resolution="1K"): + self.calls += 1 + self.image_paths.append(str(image_path)) + return self._image_bytes + + +class _FailOnClient: + """Fails only for the source image whose filename ends with *fail_name*.""" + + def __init__(self, image_bytes, fail_name): + self._image_bytes = image_bytes + self._fail_name = fail_name + self.calls = 0 + + def generate(self, prompt, image_path, resolution="1K"): + self.calls += 1 + if str(image_path).endswith(self._fail_name): + raise RuntimeError("bad image") + return self._image_bytes + + if __name__ == "__main__": unittest.main()