feat: 完成T-403结果回写与汇总

新增excel.write_back_results,按原Excel文件、工作表和行号回写新标题、新封面图片路径与更新状态。

③更新完成后自动用WriteBackWorker(mode=results)回写结果,并弹窗汇总成功、失败、略过数量;文件被占用时提示关闭Excel后手动重试。

补充Excel结果回写、GUI自动回写/汇总/锁文件提示和模块契约测试,同步任务看板、API、流程和当前状态文档。
This commit is contained in:
chengma
2026-06-27 17:28:33 +08:00
parent 04ff06a2e3
commit 3bc1776d07
10 changed files with 465 additions and 30 deletions
+49
View File
@@ -252,6 +252,55 @@ class ExcelImportTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_write_back_results_writes_new_fields_and_update_status(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
excel_path = os.path.join(temp_dir, "input.xlsx")
self.save_workbook(
excel_path,
[
(
"待处理任务",
[
["账号名", "别名", "商品id"],
["主店", "alias-a", "51100639510"],
["副店", "alias-b", "51100639511"],
["未知", "missing", "51100639512"],
],
)
],
)
result = excel.import_tasks([excel_path], path=db_path)
tasks = db.list_tasks(batch_id=result["batch_id"], path=db_path)
for task in tasks:
db.set_collected(task.id, "旧标题", "old.jpg", path=db_path)
db.set_generated(
task.id,
"新标题" + task.item_id[-2:],
rf"D:\images\{task.item_id}_new.jpg",
path=db_path,
)
refreshed = db.list_tasks(batch_id=result["batch_id"], path=db_path)
db.set_applied(refreshed[0].id, True, path=db_path)
db.set_applied(refreshed[1].id, False, "UPDATE_DISABLED", path=db_path)
db.mark_skipped(refreshed[2].id, "别名未匹配账号", path=db_path)
summary = excel.write_back_results(result["batch_id"], path=db_path)
self.assertEqual(True, summary["ok"])
self.assertEqual(1, summary["files"])
self.assertEqual(3, summary["rows"])
first = self.row_values_by_header(excel_path, row_number=2)
second = self.row_values_by_header(excel_path, row_number=3)
third = self.row_values_by_header(excel_path, row_number=4)
self.assertEqual("新标题10", first["新标题"])
self.assertEqual(r"D:\images\51100639510_new.jpg", first["新封面图片路径"])
self.assertEqual("成功", first["更新状态"])
self.assertEqual("失败:UPDATE_DISABLED", second["更新状态"])
self.assertEqual("略过:别名未匹配账号", third["更新状态"])
self.assert_removed(temp_dir)
def test_export_copy_writes_copy_without_touching_original(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")