feat: 完成Excel导入入库
实现 app/excel.py 多文件导入,解析账号名、别名、商品id,记录 source_file_abs/source_sheet/source_row/row_key。 导入时支持缺必需列整文件拒绝、脏行逐行跳过统计,并写入 batches/tasks;补充 match_summary 与回写占位。 新增 tests/test_excel.py 覆盖入库、缺列拒绝、多文件容错和别名匹配;同步任务看板、API 合约、当前状态和 progress。
This commit is contained in:
@@ -0,0 +1,167 @@
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from _helpers import TempDirMixin
|
||||
|
||||
try:
|
||||
from openpyxl import 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 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),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user