chengma
3bc1776d07
新增excel.write_back_results，按原Excel文件、工作表和行号回写新标题、新封面图片路径与更新状态。 ③更新完成后自动用WriteBackWorker(mode=results)回写结果，并弹窗汇总成功、失败、略过数量；文件被占用时提示关闭Excel后手动重试。 补充Excel结果回写、GUI自动回写/汇总/锁文件提示和模块契约测试，同步任务看板、API、流程和当前状态文档。
342 lines
14 KiB
Python
342 lines
14 KiB
Python
import os
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from _helpers import TempDirMixin
|
|
|
|
try:
|
|
from openpyxl import Workbook, load_workbook
|
|
except ModuleNotFoundError:
|
|
raise unittest.SkipTest("openpyxl 未安装")
|
|
|
|
from app import db, excel
|
|
|
|
|
|
class ExcelImportTests(TempDirMixin, unittest.TestCase):
|
|
def save_workbook(self, path, sheets):
|
|
workbook = Workbook()
|
|
default = workbook.active
|
|
workbook.remove(default)
|
|
for title, rows in sheets:
|
|
sheet = workbook.create_sheet(title=title)
|
|
for row in rows:
|
|
sheet.append(row)
|
|
workbook.save(path)
|
|
workbook.close()
|
|
|
|
def row_values_by_header(self, path, row_number=2):
|
|
workbook = load_workbook(path, data_only=True)
|
|
try:
|
|
sheet = workbook.active
|
|
headers = {
|
|
sheet.cell(row=1, column=column).value: column
|
|
for column in range(1, sheet.max_column + 1)
|
|
}
|
|
return {
|
|
header: sheet.cell(row=row_number, column=column).value
|
|
for header, column in headers.items()
|
|
}
|
|
finally:
|
|
workbook.close()
|
|
|
|
def headers(self, path):
|
|
workbook = load_workbook(path, data_only=True)
|
|
try:
|
|
sheet = workbook.active
|
|
return [
|
|
sheet.cell(row=1, column=column).value
|
|
for column in range(1, sheet.max_column + 1)
|
|
]
|
|
finally:
|
|
workbook.close()
|
|
|
|
def test_import_tasks_parses_rows_and_writes_batch_tasks(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", "52999999", ""],
|
|
["缺别名", "", "123456", ""],
|
|
["坏商品", "alias-c", "abc123", ""],
|
|
[None, None, None, None],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
|
|
result = excel.import_tasks([excel_path], path=db_path, note="导入测试")
|
|
|
|
self.assertIsNotNone(result["batch_id"])
|
|
self.assertEqual(2, len(result["rows"]))
|
|
self.assertEqual(1, result["stats"]["files"])
|
|
self.assertEqual(4, result["stats"]["total"])
|
|
self.assertEqual(2, result["stats"]["valid"])
|
|
self.assertEqual(2, result["stats"]["invalid"])
|
|
self.assertEqual(2, result["stats"]["inserted"])
|
|
self.assertEqual([], result["stats"]["file_errors"])
|
|
self.assertEqual(2, len(result["stats"]["row_errors"]))
|
|
self.assertEqual("商品", result["rows"][0]["source_sheet"])
|
|
self.assertEqual(2, result["rows"][0]["source_row"])
|
|
self.assertEqual("51100639510", result["rows"][0]["item_id"])
|
|
self.assertTrue(result["rows"][0]["row_key"].startswith(result["batch_id"] + ":"))
|
|
|
|
batch = db.get_batch(result["batch_id"], path=db_path)
|
|
tasks = db.list_tasks(batch_id=result["batch_id"], path=db_path)
|
|
self.assertEqual("导入测试", batch.note)
|
|
self.assertEqual(2, len(tasks))
|
|
self.assertEqual("alias-a", tasks[0].alias)
|
|
self.assertEqual(os.path.abspath(excel_path), tasks[0].source_file_abs)
|
|
self.assertEqual(2, tasks[0].source_row)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_missing_required_column_rejects_whole_file(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, "missing_alias.xlsx")
|
|
self.save_workbook(
|
|
excel_path,
|
|
[
|
|
(
|
|
"商品",
|
|
[
|
|
["账号名", "商品id"],
|
|
["主店", "51100639510"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
|
|
result = excel.import_tasks([excel_path], path=db_path)
|
|
|
|
self.assertIsNone(result["batch_id"])
|
|
self.assertEqual([], result["rows"])
|
|
self.assertEqual(1, len(result["stats"]["file_errors"]))
|
|
self.assertEqual(["别名"], result["stats"]["file_errors"][0]["missing_columns"])
|
|
self.assertFalse(os.path.exists(db_path))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_import_tasks_continues_when_another_file_has_column_error(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
good_path = os.path.join(temp_dir, "good.xlsx")
|
|
bad_path = os.path.join(temp_dir, "missing_item.xlsx")
|
|
self.save_workbook(
|
|
good_path,
|
|
[
|
|
(
|
|
"商品",
|
|
[
|
|
["账号名", "别名", "商品id"],
|
|
["主店", "alias-a", "51100639510"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
self.save_workbook(
|
|
bad_path,
|
|
[
|
|
(
|
|
"商品",
|
|
[
|
|
["账号名", "别名"],
|
|
["主店", "alias-a"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
|
|
result = excel.import_tasks([good_path, bad_path], path=db_path)
|
|
|
|
self.assertIsNotNone(result["batch_id"])
|
|
self.assertEqual(2, result["stats"]["files"])
|
|
self.assertEqual(1, result["stats"]["valid"])
|
|
self.assertEqual(1, result["stats"]["inserted"])
|
|
self.assertEqual(1, len(result["stats"]["file_errors"]))
|
|
self.assertEqual(["商品id"], result["stats"]["file_errors"][0]["missing_columns"])
|
|
self.assertEqual(os.path.abspath(good_path), result["rows"][0]["source_file_abs"])
|
|
self.assertEqual(1, len(db.list_tasks(batch_id=result["batch_id"], path=db_path)))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_match_summary_counts_known_and_unknown_aliases(self):
|
|
rows = [
|
|
{"alias": "a"},
|
|
{"alias": "b"},
|
|
{"alias": "a"},
|
|
{"alias": "missing"},
|
|
]
|
|
accounts = [
|
|
{"alias": "a"},
|
|
{"alias": "b"},
|
|
]
|
|
|
|
self.assertEqual(
|
|
{
|
|
"matched": 3,
|
|
"unmatched": 1,
|
|
"by_account": {"a": 2, "b": 1},
|
|
"unmatched_aliases": ["missing"],
|
|
},
|
|
excel.match_summary(rows, accounts),
|
|
)
|
|
|
|
def test_write_back_writes_old_fields_to_original_excel(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"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
result = excel.import_tasks([excel_path], path=db_path)
|
|
task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0]
|
|
db.set_collected(
|
|
task.id,
|
|
"原始商品标题",
|
|
r"D:\images\51100639510_old.jpg",
|
|
path=db_path,
|
|
)
|
|
|
|
summary = excel.write_back(result["batch_id"], path=db_path)
|
|
|
|
self.assertEqual(True, summary["ok"])
|
|
self.assertEqual(1, summary["files"])
|
|
self.assertEqual(1, summary["rows"])
|
|
self.assertEqual([os.path.abspath(excel_path)], summary["written_files"])
|
|
values = self.row_values_by_header(excel_path)
|
|
self.assertEqual("原始商品标题", values["旧标题"])
|
|
self.assertEqual(r"D:\images\51100639510_old.jpg", values["旧封面图片路径"])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_write_back_locked_file_raises_clear_error(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"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
result = excel.import_tasks([excel_path], path=db_path)
|
|
task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0]
|
|
db.set_collected(task.id, "原始商品标题", "old.jpg", path=db_path)
|
|
|
|
with mock.patch("app.excel.load_workbook", side_effect=PermissionError("locked")):
|
|
with self.assertRaisesRegex(excel.ExcelError, "Excel 文件被占用"):
|
|
excel.write_back(result["batch_id"], path=db_path)
|
|
|
|
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")
|
|
excel_path = os.path.join(temp_dir, "input.xlsx")
|
|
out_dir = os.path.join(temp_dir, "out")
|
|
self.save_workbook(
|
|
excel_path,
|
|
[
|
|
(
|
|
"待处理任务",
|
|
[
|
|
["账号名", "别名", "商品id"],
|
|
["主店", "alias-a", "51100639510"],
|
|
],
|
|
)
|
|
],
|
|
)
|
|
result = excel.import_tasks([excel_path], path=db_path)
|
|
task = db.list_tasks(batch_id=result["batch_id"], path=db_path)[0]
|
|
db.set_collected(task.id, "原始商品标题", "old.jpg", path=db_path)
|
|
|
|
summary = excel.export_copy(result["batch_id"], out_dir, path=db_path)
|
|
|
|
self.assertEqual(1, summary["files"])
|
|
self.assertEqual(1, summary["rows"])
|
|
copy_path = summary["written_files"][0]
|
|
self.assertTrue(os.path.exists(copy_path))
|
|
self.assertIn("input_cmshopee回写", os.path.basename(copy_path))
|
|
self.assertNotIn("旧标题", self.headers(excel_path))
|
|
values = self.row_values_by_header(copy_path)
|
|
self.assertEqual("原始商品标题", values["旧标题"])
|
|
self.assertEqual("old.jpg", values["旧封面图片路径"])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|