From 71e51d06087c20effe79f33510f60eeb73cffa42 Mon Sep 17 00:00:00 2001 From: chengma Date: Thu, 9 Jul 2026 11:23:18 +0800 Subject: [PATCH] =?UTF-8?q?T-566=20=E5=B0=81=E9=9D=A2=E5=8E=86=E5=8F=B2?= =?UTF-8?q?=E5=BD=92=E6=A1=A3=E6=95=B0=E6=8D=AE=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/db.py | 70 +++++++++++++-- app/image_paths.py | 40 ++++++++- docs/tasks/T-566.md | 19 +++- tests/test_db.py | 185 +++++++++++++++++++++++++++++++++++++- tests/test_gui.py | 9 +- tests/test_image_paths.py | 43 ++++++++- 6 files changed, 354 insertions(+), 12 deletions(-) diff --git a/app/db.py b/app/db.py index 3eddc34..cc1996b 100644 --- a/app/db.py +++ b/app/db.py @@ -292,6 +292,34 @@ def _now() -> str: return datetime.now().isoformat(timespec="seconds") +def _cover_archive_timestamp() -> str: + return datetime.now().strftime("%Y%m%d%H%M%S") + + +def _archive_generated_cover_file(file_path) -> Optional[str]: + path_text = str(file_path or "").strip() + if not path_text or not os.path.isfile(path_text): + return None + directory = os.path.dirname(path_text) + filename = os.path.basename(path_text) + stem, ext = os.path.splitext(filename) + timestamp = _cover_archive_timestamp() + suffix = 1 + while True: + archive_name = f"{stem}_{timestamp}{ext}" if suffix == 1 else f"{stem}_{timestamp}_{suffix}{ext}" + archive_path = os.path.join(directory, archive_name) + if not os.path.exists(archive_path): + break + suffix += 1 + try: + os.rename(path_text, archive_path) + except PermissionError as exc: + raise DbError("请先关闭正在查看的封面图片再重置") from exc + except OSError as exc: + raise DbError(f"归档当前封面图片失败: {exc}") from exc + return archive_path + + def _db_path(path=None) -> str: return path or appconfig.db_path() @@ -846,6 +874,38 @@ def update_generated_title(task_id, new_title, path=None, conn=None) -> None: (title, now, int(task_id)), ) + +def update_generated_cover(task_id, new_cover_path, path=None, conn=None) -> None: + """Point a task at an existing local generated cover without copying the file.""" + + cover_path = str(new_cover_path or "").strip() + if not cover_path: + raise DbError("新封面路径不能为空") + abs_cover_path = os.path.abspath(cover_path) + if not os.path.isfile(abs_cover_path): + raise DbError("新封面图片不存在") + with _connection(conn, path) as database: + task = get_task(task_id, conn=database) + if task is None: + raise DbError(f"任务不存在: {task_id}") + if task.status == "running": + raise DbError("任务正在运行，不能修改新封面") + now = _now() + with database: + database.execute( + """ + UPDATE tasks + SET new_cover_path = ?, + stage = 'generated', + status = 'pending', + last_error = NULL, + updated_at = ? + WHERE id = ? + """, + (abs_cover_path, now, int(task_id)), + ) + + def set_applied(task_id, committed, error=None, path=None, conn=None, step=None) -> None: now = _now() success = bool(committed) and error is None @@ -901,10 +961,9 @@ def reset_generated( if before is None: raise DbError(f"任务不存在: {task_id}") new_cover_path = before.new_cover_path - deleted_file = None - if delete_file and reset_cover and new_cover_path and os.path.isfile(str(new_cover_path)): - os.remove(str(new_cover_path)) - deleted_file = new_cover_path + archived_file = None + if reset_cover: + archived_file = _archive_generated_cover_file(new_cover_path) now = _now() new_title = None if reset_title else before.new_title new_cover = None if reset_cover else before.new_cover_path @@ -932,7 +991,8 @@ def reset_generated( "before": before, "after": after, "new_cover_path": new_cover_path, - "deleted_file": deleted_file, + "archived_file": archived_file, + "deleted_file": None, "reset_title": bool(reset_title), "reset_cover": bool(reset_cover), } diff --git a/app/image_paths.py b/app/image_paths.py index 70986e0..ad0376d 100644 --- a/app/image_paths.py +++ b/app/image_paths.py @@ -1,4 +1,5 @@ import os +import re from .config import make_slug @@ -22,6 +23,27 @@ def task_image_path(image_root, task, account=None, suffix="old", ext=".jpg"): ) +def list_task_cover_candidates(image_root, task, account=None): + """Return generated cover candidates for one task from the image directory.""" + + canonical_path = task_image_path(image_root, task, account=account, suffix="new", ext=".jpg") + directory = os.path.dirname(canonical_path) + prefix = os.path.splitext(os.path.basename(canonical_path))[0] + if not os.path.isdir(directory): + return [] + candidates = [] + for filename in os.listdir(directory): + full_path = os.path.join(directory, filename) + if not os.path.isfile(full_path): + continue + stem, ext = os.path.splitext(filename) + if ext.lower() != ".jpg": + continue + if stem == prefix or stem.startswith(prefix + "_"): + candidates.append(os.path.abspath(full_path)) + return sorted(candidates, key=lambda path: _cover_candidate_sort_key(path, prefix)) + + def _account_slug(account, task): slug = _get(account, "slug") if slug: @@ -32,6 +54,22 @@ def _account_slug(account, task): return "unknown_account" +def _cover_candidate_sort_key(path, prefix): + filename = os.path.basename(path) + stem, _ext = os.path.splitext(filename) + if stem == prefix: + return (0, 0, 0, filename) + marker = prefix + "_" + if stem.startswith(marker): + suffix = stem[len(marker):] + match = re.fullmatch(r"(\d{14})(?:_(\d+))?", suffix) + if match: + timestamp = int(match.group(1)) + collision = int(match.group(2) or 1) + return (1, -timestamp, -collision, filename) + return (2, 0, 0, filename) + + def _get(obj, name, default=None): if obj is None: return default @@ -45,4 +83,4 @@ def _safe_component(value, default): if not text: text = str(default) safe = "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in text).strip("_") - return safe or str(default) \ No newline at end of file + return safe or str(default) diff --git a/docs/tasks/T-566.md b/docs/tasks/T-566.md index b8cb62e..9306f63 100644 --- a/docs/tasks/T-566.md +++ b/docs/tasks/T-566.md @@ -3,7 +3,7 @@ id: T-566 title: 封面历史归档数据层：重置改名保留旧图 + update_generated_cover + 目录枚举 phase: 7 deps: [T-564] -status: TODO +status: DONE created: 2026-07-09 --- @@ -84,4 +84,19 @@ created: 2026-07-09 ## 执行记录 -（做完在这里写：改了什么文件、跑了什么验证命令及结果、遇到的阻塞、关键决策。） +- 2026-07-09 完成 T-566。 +- 修改 `app/db.py`： + - `reset_generated(..., reset_cover=True)` 在清空 DB 指针前先把当前 `new_cover_path` 指向的本地文件改名归档为 `{stem}_YYYYMMDDHHMMSS{ext}`；同秒撞名追加 `_2/_3`；旧 `delete_file=True` 不再物理删除，归档取代删除，返回结果新增 `archived_file`，保留 `deleted_file=None` 兼容旧调用。 + - 文件不存在时幂等跳过归档；`PermissionError` 抛中文 `DbError` 并保持 DB 指针不变。 + - 新增 `update_generated_cover(task_id, new_cover_path)`，只切换当前生效封面指针，写入后退回 `generated/pending`，不增加 attempts，不清 `committed/applied_at/apply_attempts/image_task_id/image_task_key`。 +- 修改 `app/image_paths.py`：新增 `list_task_cover_candidates(image_root, task, account=None)`，只读扫描任务目录下 `_new*.jpg` 候选，返回绝对路径，排序为规范名优先、归档时间倒序、同秒后缀数字倒序、异常文件名最后。 +- 修改测试： + - `tests/test_db.py` 覆盖归档撞名、文件锁失败 DB 不变、缺失文件幂等、`update_generated_cover` 状态语义、运行中拒绝切换、批次软删除不删图片目录。 + - `tests/test_image_paths.py` 覆盖候选图过滤和稳定排序。 + - `tests/test_gui.py` 同步重置后本地文件“归档保留而非原名保留”的新语义。 +- 验证通过： + - `py -3.10 -m unittest tests.test_db tests.test_image_paths` + - `python -m ruff check app tests main.py` + - `py -3.10 -m compileall app main.py` + - `py -3.10 -m unittest discover -s tests`（285 tests；PySide6 字体目录警告不影响结果） + - `git diff --check` diff --git a/tests/test_db.py b/tests/test_db.py index c3e37d5..8638214 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1,6 +1,7 @@ import os import sys import unittest +from unittest import mock sys.path.insert(0, os.path.dirname(__file__)) @@ -302,8 +303,10 @@ class DbTests(TempDirMixin, unittest.TestCase): reset_generated = db.reset_generated(task.id, path=db_path) after_generated = reset_generated["after"] self.assertEqual(new_cover, reset_generated["new_cover_path"]) + self.assertIsNotNone(reset_generated["archived_file"]) self.assertIsNone(reset_generated["deleted_file"]) - self.assertTrue(os.path.exists(new_cover)) + self.assertFalse(os.path.exists(new_cover)) + self.assertTrue(os.path.exists(reset_generated["archived_file"])) self.assertEqual("generated", after_generated.stage) self.assertEqual("success", after_generated.status) self.assertIsNone(after_generated.new_title) @@ -378,8 +381,10 @@ class DbTests(TempDirMixin, unittest.TestCase): self.assertEqual("success", after_cover.status) self.assertEqual("手动标题B", after_cover.new_title) self.assertIsNone(after_cover.new_cover_path) - self.assertEqual(second_cover, cover_reset["deleted_file"]) + self.assertIsNone(cover_reset["deleted_file"]) + self.assertIsNotNone(cover_reset["archived_file"]) self.assertFalse(os.path.exists(second_cover)) + self.assertTrue(os.path.exists(cover_reset["archived_file"])) with self.assertRaises(db.DbError): db.reset_generated(first.id, reset_title=False, reset_cover=False, path=db_path) @@ -389,6 +394,175 @@ class DbTests(TempDirMixin, unittest.TestCase): self.assert_removed(temp_dir) + def test_reset_generated_archives_cover_with_collision_name(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + db.init_db(db_path) + batch_id = db.create_batch(["input.xlsx"], path=db_path) + db.insert_tasks( + batch_id, + [ + { + "source_file_abs": os.path.join(temp_dir, "input.xlsx"), + "source_sheet": "Sheet1", + "source_row": 2, + "account_name": "shop", + "alias": "alias", + "item_id": "51100639510", + } + ], + path=db_path, + ) + task = db.list_tasks(batch_id=batch_id, path=db_path)[0] + cover = os.path.join(temp_dir, "42_51100639510_new.jpg") + collision = os.path.join(temp_dir, "42_51100639510_new_20260709094800.jpg") + expected_archive = os.path.join(temp_dir, "42_51100639510_new_20260709094800_2.jpg") + with open(cover, "wb") as fh: + fh.write(b"current") + with open(collision, "wb") as fh: + fh.write(b"existing") + db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) + db.set_generated(task.id, "新标题", cover, path=db_path) + + with mock.patch("app.db._cover_archive_timestamp", return_value="20260709094800"): + result = db.reset_generated( + task.id, + reset_title=False, + reset_cover=True, + path=db_path, + ) + + self.assertEqual(expected_archive, result["archived_file"]) + self.assertFalse(os.path.exists(cover)) + self.assertTrue(os.path.exists(collision)) + self.assertTrue(os.path.exists(expected_archive)) + with open(collision, "rb") as fh: + self.assertEqual(b"existing", fh.read()) + with open(expected_archive, "rb") as fh: + self.assertEqual(b"current", fh.read()) + after = db.get_task(task.id, path=db_path) + self.assertEqual("新标题", after.new_title) + self.assertIsNone(after.new_cover_path) + + self.assert_removed(temp_dir) + + def test_reset_generated_permission_error_keeps_db_pointer(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + db.init_db(db_path) + batch_id = db.create_batch(["input.xlsx"], path=db_path) + db.insert_tasks( + batch_id, + [ + { + "source_file_abs": os.path.join(temp_dir, "input.xlsx"), + "source_sheet": "Sheet1", + "source_row": 2, + "account_name": "shop", + "alias": "alias", + "item_id": "51100639510", + } + ], + path=db_path, + ) + task = db.list_tasks(batch_id=batch_id, path=db_path)[0] + cover = os.path.join(temp_dir, "new.jpg") + with open(cover, "wb") as fh: + fh.write(b"jpeg") + db.set_generated(task.id, "新标题", cover, path=db_path) + + with mock.patch("app.db.os.rename", side_effect=PermissionError): + with self.assertRaisesRegex(db.DbError, "请先关闭正在查看的封面图片再重置"): + db.reset_generated(task.id, reset_title=False, reset_cover=True, path=db_path) + + after = db.get_task(task.id, path=db_path) + self.assertEqual(cover, after.new_cover_path) + self.assertEqual("新标题", after.new_title) + self.assertTrue(os.path.exists(cover)) + + self.assert_removed(temp_dir) + + def test_update_generated_cover_switches_pointer_and_keeps_apply_history(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + db.init_db(db_path) + batch_id = db.create_batch(["input.xlsx"], path=db_path) + db.insert_tasks( + batch_id, + [ + { + "source_file_abs": os.path.join(temp_dir, "input.xlsx"), + "source_sheet": "Sheet1", + "source_row": 2, + "account_name": "shop", + "alias": "alias", + "item_id": "51100639510", + } + ], + path=db_path, + ) + task = db.list_tasks(batch_id=batch_id, path=db_path)[0] + current_cover = os.path.join(temp_dir, "current.jpg") + selected_cover = os.path.join(temp_dir, "selected.jpg") + for cover in (current_cover, selected_cover): + with open(cover, "wb") as fh: + fh.write(b"jpeg") + db.set_collected(task.id, "旧标题", "old.jpg", path=db_path) + db.set_generated(task.id, "新标题", current_cover, path=db_path) + key = db.ensure_image_task_key(task.id, path=db_path) + db.set_image_task_submitted(task.id, "cmhub-task-1", key, path=db_path) + db.set_applied(task.id, True, path=db_path) + before = db.get_task(task.id, path=db_path) + + db.update_generated_cover(task.id, selected_cover, path=db_path) + + after = db.get_task(task.id, path=db_path) + self.assertEqual(os.path.abspath(selected_cover), after.new_cover_path) + self.assertEqual("generated", after.stage) + self.assertEqual("pending", after.status) + self.assertIsNone(after.last_error) + self.assertEqual(before.generate_attempts, after.generate_attempts) + self.assertEqual(before.apply_attempts, after.apply_attempts) + self.assertEqual(before.committed, after.committed) + self.assertEqual(before.applied_at, after.applied_at) + self.assertEqual(before.image_task_id, after.image_task_id) + self.assertEqual(before.image_task_key, after.image_task_key) + + with self.assertRaisesRegex(db.DbError, "新封面图片不存在"): + db.update_generated_cover(task.id, os.path.join(temp_dir, "missing.jpg"), path=db_path) + + self.assert_removed(temp_dir) + + def test_update_generated_cover_rejects_running_task(self): + with self.make_temp_dir() as temp_dir: + db_path = os.path.join(temp_dir, "cmshopee.db") + db.init_db(db_path) + batch_id = db.create_batch(["input.xlsx"], path=db_path) + db.insert_tasks( + batch_id, + [ + { + "source_file_abs": os.path.join(temp_dir, "input.xlsx"), + "source_sheet": "Sheet1", + "source_row": 2, + "account_name": "shop", + "alias": "alias", + "item_id": "51100639510", + } + ], + path=db_path, + ) + task = db.list_tasks(batch_id=batch_id, path=db_path)[0] + cover = os.path.join(temp_dir, "selected.jpg") + with open(cover, "wb") as fh: + fh.write(b"jpeg") + db.mark_running(task.id, "generate", path=db_path) + + with self.assertRaisesRegex(db.DbError, "任务正在运行"): + db.update_generated_cover(task.id, cover, path=db_path) + + self.assert_removed(temp_dir) + def test_delete_batch_soft_hides_batch_and_tasks(self): with self.make_temp_dir() as temp_dir: db_path = os.path.join(temp_dir, "cmshopee.db") @@ -419,6 +593,11 @@ class DbTests(TempDirMixin, unittest.TestCase): tasks = db.list_tasks(batch_id=batch_id, path=db_path) old_cover = os.path.join(temp_dir, "old.jpg") new_cover = os.path.join(temp_dir, "new.jpg") + batch_image_dir = os.path.join(temp_dir, "images", batch_id, "alias") + os.makedirs(batch_image_dir, exist_ok=True) + generated_cover = os.path.join(batch_image_dir, "1_51100639510_new.jpg") + with open(generated_cover, "wb") as fh: + fh.write(b"jpeg") db.set_collected(tasks[0].id, "旧标题", old_cover, path=db_path) db.set_generated(tasks[0].id, "新标题", new_cover, path=db_path) db.set_applied(tasks[0].id, True, path=db_path) @@ -440,6 +619,8 @@ class DbTests(TempDirMixin, unittest.TestCase): self.assertEqual("导错文件", deleted_batch.deleted_reason) deleted_tasks = db.list_tasks(batch_id=batch_id, path=db_path, include_deleted=True) self.assertEqual(2, len(deleted_tasks)) + self.assertTrue(os.path.exists(batch_image_dir)) + self.assertTrue(os.path.exists(generated_cover)) self.assert_removed(temp_dir) diff --git a/tests/test_gui.py b/tests/test_gui.py index 8c1b1ec..7985e2e 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -2151,7 +2151,14 @@ class GuiTests(TempDirMixin, unittest.TestCase): self.assertEqual("success", updated.status) self.assertIsNone(updated.new_title) self.assertIsNone(updated.new_cover_path) - self.assertTrue(os.path.exists(new_cover)) + self.assertFalse(os.path.exists(new_cover)) + archived_covers = [ + filename + for filename in os.listdir(temp_dir) + if filename.startswith("new_") and filename.endswith(".jpg") + ] + self.assertEqual(1, len(archived_covers)) + self.assertTrue(os.path.exists(os.path.join(temp_dir, archived_covers[0]))) run_log = db.list_run_logs(limit=1, run_type="reset", path=cfg["db_path"])[0] self.assertEqual("done", run_log.status) self.assertEqual("reset_generated", run_log.options["action"]) diff --git a/tests/test_image_paths.py b/tests/test_image_paths.py index a5df7c3..9212c68 100644 --- a/tests/test_image_paths.py +++ b/tests/test_image_paths.py @@ -1,5 +1,6 @@ import os import sys +import tempfile import unittest from types import SimpleNamespace @@ -33,6 +34,46 @@ class ImagePathTests(unittest.TestCase): path, ) + def test_list_task_cover_candidates_filters_and_sorts_generated_covers(self): + with tempfile.TemporaryDirectory() as temp_dir: + task = SimpleNamespace(id=42, batch_id="batch-a", item_id="51100639510", alias="alias-a") + canonical = image_paths.task_image_path(temp_dir, task, suffix="new") + directory = os.path.dirname(canonical) + prefix = os.path.splitext(os.path.basename(canonical))[0] + os.makedirs(directory, exist_ok=True) + filenames = [ + f"{prefix}_20260709094800.jpg", + f"{prefix}_20260709094800_2.jpg", + f"{prefix}_20260709094900.jpg", + f"{prefix}.jpg", + f"{prefix}_bad.jpg", + f"{prefix.replace('_new', '_old')}.jpg", + f"{prefix}.png", + "99_51100639510_new.jpg", + ] + for filename in filenames: + with open(os.path.join(directory, filename), "wb") as fh: + fh.write(b"jpeg") + + paths = image_paths.list_task_cover_candidates(temp_dir, task) + + self.assertEqual( + [ + canonical, + os.path.join(directory, f"{prefix}_20260709094900.jpg"), + os.path.join(directory, f"{prefix}_20260709094800_2.jpg"), + os.path.join(directory, f"{prefix}_20260709094800.jpg"), + os.path.join(directory, f"{prefix}_bad.jpg"), + ], + paths, + ) + + def test_list_task_cover_candidates_returns_empty_for_missing_directory(self): + with tempfile.TemporaryDirectory() as temp_dir: + task = SimpleNamespace(id=42, batch_id="batch-a", item_id="51100639510", alias="alias-a") + + self.assertEqual([], image_paths.list_task_cover_candidates(temp_dir, task)) + if __name__ == "__main__": - unittest.main() \ No newline at end of file + unittest.main()