cmshoppe/tests/test_excel.py

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