From 10238d3da3a7f3cbe739f2d60574a40f424a2271 Mon Sep 17 00:00:00 2001 From: ila Date: Thu, 18 Jun 2026 17:35:00 +0800 Subject: [PATCH] feat: add outfit excel service --- src/services/excel_service.py | 139 ++++++++++++++++++++++++++++++++++ tasks.md | 2 +- tests/test_excel_service.py | 113 +++++++++++++++++++++++++++ 3 files changed, 253 insertions(+), 1 deletion(-) create mode 100644 src/services/excel_service.py create mode 100644 tests/test_excel_service.py diff --git a/src/services/excel_service.py b/src/services/excel_service.py new file mode 100644 index 0000000..bd4d0b5 --- /dev/null +++ b/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() diff --git a/tasks.md b/tasks.md index 3a12dce..c0dd735 100644 --- a/tasks.md +++ b/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 diff --git a/tests/test_excel_service.py b/tests/test_excel_service.py new file mode 100644 index 0000000..0c54c87 --- /dev/null +++ b/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()