feat: add outfit excel service

src/services/excel_service.py

@@ -0,0 +1,139 @@
+import logging
+from pathlib import Path
+from typing import List
+
+from openpyxl import load_workbook
+
+from core.models import OutfitResult, OutfitTask
+
+logger = logging.getLogger(__name__)
+
+
+STATUS_PENDING = "待处理"
+STATUS_RUNNING = "生成中"
+STATUS_DONE = "完成"
+STATUS_FAILED = "失败"
+STATUS_SKIPPED = "跳过"
+
+COL_TITLE = 1
+COL_PRODUCT_ID = 2
+COL_GARMENT_PATH = 3
+COL_OUTPUT_PATH = 4
+COL_STATUS = 5
+COL_ERROR = 6
+
+
+class ExcelInUseError(OSError):
+    """Raised when an Excel file cannot be opened for writing."""
+
+
+def _cell_text(value):
+    """Convert an Excel cell value to text without stripping user content."""
+    if value is None:
+        return ""
+    return str(value)
+
+
+def _is_empty(value):
+    """Return True only for genuinely empty cells.
+
+    Whitespace is preserved as user data, matching docs/11's "do not clean
+    fields" rule.
+    """
+    return value is None or value == ""
+
+
+def check_excel_writable(excel_path):
+    """Return True if *excel_path* can be opened for writing.
+
+    This detects the common Windows case where Office keeps the workbook locked.
+    The actual save still happens through openpyxl and may raise OSError for
+    other filesystem errors.
+    """
+    path = Path(excel_path)
+    try:
+        with open(str(path), "a+b"):
+            pass
+        return True
+    except OSError:
+        return False
+
+
+def ensure_excel_writable(excel_path):
+    """Raise ExcelInUseError if *excel_path* is not writable."""
+    if not check_excel_writable(excel_path):
+        raise ExcelInUseError("Excel 文件无法写入，请关闭后重试: {}".format(excel_path))
+
+
+def load_outfit_tasks(excel_path, retry_failed=False):
+    """Load processable outfit tasks from the first worksheet.
+
+    Column mapping follows docs/11-ai-outfit.md:
+    A title, B product id, C garment path, D output path, E status, F error.
+
+    Rows start at 2. Completed rows are always skipped. Failed rows are included
+    only when *retry_failed* is True. Rows missing title/product/path are skipped
+    silently and are not written back.
+    """
+    path = Path(excel_path)
+    workbook = load_workbook(str(path))
+    try:
+        sheet = workbook.worksheets[0]
+        tasks = []
+
+        for row_index in range(2, sheet.max_row + 1):
+            title = sheet.cell(row_index, COL_TITLE).value
+            product_id = sheet.cell(row_index, COL_PRODUCT_ID).value
+            garment_path = sheet.cell(row_index, COL_GARMENT_PATH).value
+
+            if _is_empty(title) or _is_empty(product_id) or _is_empty(garment_path):
+                logger.debug("Skipped incomplete outfit row %s in %s", row_index, path)
+                continue
+
+            raw_status = sheet.cell(row_index, COL_STATUS).value
+            status = _cell_text(raw_status) or STATUS_PENDING
+            if status == STATUS_DONE:
+                continue
+            if status == STATUS_FAILED and not retry_failed:
+                continue
+
+            tasks.append(
+                OutfitTask(
+                    row_index=row_index,
+                    title=_cell_text(title),
+                    product_id=_cell_text(product_id),
+                    garment_path=_cell_text(garment_path),
+                    status=status,
+                )
+            )
+
+        logger.info("Loaded %d outfit task(s) from %s", len(tasks), path)
+        return tasks
+    finally:
+        workbook.close()
+
+
+def write_outfit_result(excel_path, result):
+    """Write one outfit result to columns D/E/F and save immediately."""
+    if not isinstance(result, OutfitResult):
+        raise TypeError("result must be OutfitResult")
+
+    path = Path(excel_path)
+    workbook = load_workbook(str(path))
+    try:
+        sheet = workbook.worksheets[0]
+        row_index = result.task.row_index
+
+        if result.success:
+            sheet.cell(row_index, COL_OUTPUT_PATH).value = result.output_path
+            sheet.cell(row_index, COL_STATUS).value = STATUS_DONE
+            sheet.cell(row_index, COL_ERROR).value = ""
+            logger.info("Outfit row %s completed: %s", row_index, result.output_path)
+        else:
+            sheet.cell(row_index, COL_STATUS).value = STATUS_FAILED
+            sheet.cell(row_index, COL_ERROR).value = result.error
+            logger.warning("Outfit row %s failed: %s", row_index, result.error)
+
+        workbook.save(str(path))
+    finally:
+        workbook.close()

tasks.md

@@ -1047,7 +1047,7 @@
 
 - [x] 前置依赖：`requirements.txt` 增加 Python 3.7 兼容的 `requests` / `urllib3` / `openpyxl` 锁定版本，并同步 `docs/03-technical-stack.md`
 - [x] `core/models.py` 新增 `OutfitTask` / `OutfitResult`（纯 dataclass，Python 3.7 兼容，不依赖 PySide6）
-- [ ] `services/excel_service.py`：读行 → `List[OutfitTask]`、写回 D/E/F、占用检测、跳过「完成」/按设置重试「失败」/空字段安全跳过、每行即存
+- [x] `services/excel_service.py`：读行 → `List[OutfitTask]`、写回 D/E/F、占用检测、跳过「完成」/按设置重试「失败」/空字段安全跳过、每行即存
 - [ ] `services/ai_image_service.py`：移植旧项目 `ImageApiClient`（多模型、多请求格式 `chat/gemini/images/images_edits`、传图 data-url、递归取图、URL 归一化、字段校验）；注意 PEP585 类型注解改 Python 3.7 写法
 - [ ] `core/ai_outfit.py`：单行生成纯逻辑编排（提示词渲染 + 调用 + 保存 JPG + 产出 `OutfitResult`）
 - [ ] 单测：Excel 读写、提示词渲染、取图、命名去重（API 用 mock）；可在 Python 3.7 运行、不依赖 GUI

tests/test_excel_service.py

@@ -0,0 +1,113 @@
+"""Tests for AI outfit Excel service.
+
+Pure service tests: no GUI dependency.
+"""
+import shutil
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from openpyxl import Workbook, load_workbook
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+
+class TestOutfitExcelService(unittest.TestCase):
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp())
+        self.excel_path = self.tmp / "outfit.xlsx"
+        self._write_workbook()
+
+    def tearDown(self):
+        shutil.rmtree(str(self.tmp), ignore_errors=True)
+
+    def _write_workbook(self):
+        wb = Workbook()
+        ws = wb.active
+        ws.append(["标题", "货号", "衣服图", "结果图", "状态", "原因"])
+        ws.append(["T恤", "TY001", r"D:\img\ty001.png", "", "", ""])
+        ws.append(["已完成", "TY002", r"D:\img\ty002.png", r"D:\out\ty002.jpg", "完成", ""])
+        ws.append(["失败行", "TY003", r"D:\img\ty003.png", "", "失败", "旧错误"])
+        ws.append(["缺路径", "TY004", "", "", "", ""])
+        ws.append(["  保留空格  ", 1005, r"D:\img\ty005.png", "", "", ""])
+        wb.save(str(self.excel_path))
+        wb.close()
+
+    def _read_row(self, row):
+        wb = load_workbook(str(self.excel_path))
+        try:
+            ws = wb.active
+            return (
+                ws.cell(row, 4).value,
+                ws.cell(row, 5).value,
+                ws.cell(row, 6).value,
+            )
+        finally:
+            wb.close()
+
+    def test_load_tasks_skips_done_failed_and_incomplete_by_default(self):
+        from services.excel_service import load_outfit_tasks
+
+        tasks = load_outfit_tasks(self.excel_path)
+
+        self.assertEqual([task.row_index for task in tasks], [2, 6])
+        self.assertEqual(tasks[0].title, "T恤")
+        self.assertEqual(tasks[0].product_id, "TY001")
+        self.assertEqual(tasks[0].garment_path, r"D:\img\ty001.png")
+
+    def test_load_tasks_includes_failed_when_retry_enabled(self):
+        from services.excel_service import load_outfit_tasks
+
+        tasks = load_outfit_tasks(self.excel_path, retry_failed=True)
+
+        self.assertEqual([task.row_index for task in tasks], [2, 4, 6])
+        self.assertEqual(tasks[1].status, "失败")
+
+    def test_load_tasks_preserves_user_cell_text(self):
+        from services.excel_service import load_outfit_tasks
+
+        tasks = load_outfit_tasks(self.excel_path)
+        task = tasks[-1]
+
+        self.assertEqual(task.title, "  保留空格  ")
+        self.assertEqual(task.product_id, "1005")
+
+    def test_write_success_result_updates_d_e_f_and_saves(self):
+        from core.models import OutfitResult, OutfitTask
+        from services.excel_service import write_outfit_result
+
+        task = OutfitTask(row_index=2, title="T恤", product_id="TY001", garment_path="g.png")
+        result = OutfitResult(task=task, success=True, output_path=r"D:\out\TY001.jpg")
+
+        write_outfit_result(self.excel_path, result)
+
+        self.assertEqual(self._read_row(2), (r"D:\out\TY001.jpg", "完成", None))
+
+    def test_write_failure_result_updates_e_f_and_saves(self):
+        from core.models import OutfitResult, OutfitTask
+        from services.excel_service import write_outfit_result
+
+        task = OutfitTask(row_index=2, title="T恤", product_id="TY001", garment_path="g.png")
+        result = OutfitResult(task=task, success=False, error="衣服图打不开")
+
+        write_outfit_result(self.excel_path, result)
+
+        self.assertEqual(self._read_row(2), (None, "失败", "衣服图打不开"))
+
+    def test_check_excel_writable_true_for_existing_workbook(self):
+        from services.excel_service import check_excel_writable
+
+        self.assertTrue(check_excel_writable(self.excel_path))
+
+    def test_ensure_excel_writable_raises_when_probe_fails(self):
+        from services.excel_service import ExcelInUseError, ensure_excel_writable
+
+        with patch("builtins.open", side_effect=OSError("locked")):
+            with self.assertRaises(ExcelInUseError):
+                ensure_excel_writable(self.excel_path)
+
+
+if __name__ == "__main__":
+    unittest.main()