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_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()