cmautobuy/tools/test_import_catalog_xlsx.py

import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from openpyxl import Workbook


MODULE_PATH = Path(__file__).with_name("import_catalog_xlsx.py")
SPEC = importlib.util.spec_from_file_location("import_catalog_xlsx", MODULE_PATH)
assert SPEC and SPEC.loader
catalog = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = catalog
SPEC.loader.exec_module(catalog)


HEADERS = [
    "商品ID",
    "商品标题",
    "主货号",
    "货源URL",
    "货源平台",
    "货源ID",
    "货源店铺名",
    "价格",
    "币种",
    "库存",
    "销量",
    "浏览量",
    "收藏量",
    "评论数",
    "商品状态",
    "店铺显示名",
    "平台店铺ID",
    "创建时间",
    "更新时间",
    "缩略图URL",
    "变种属性值一",
    "变种属性值二",
]


def sample_row(shopee_id, pdd_id, first, second):
    return [
        shopee_id,
        f"商品 {shopee_id}",
        "MAIN",
        f"https://mobile.yangkeduo.com/goods.html?goods_id={pdd_id}&tracking=removed",
        "拼多多",
        pdd_id,
        "PDD 店铺",
        "100",
        "TWD",
        10,
        "",
        "",
        "",
        "",
        "NORMAL",
        "蝦皮店铺",
        "SHOP",
        "2026-08-10 08:00:00",
        "2026-08-11 09:30:00",
        "https://img.example.com/a.jpg",
        first,
        second,
    ]


class CatalogXLSXTest(unittest.TestCase):
    def make_workbook(self, rows):
        temp = tempfile.TemporaryDirectory()
        path = Path(temp.name) / "sample.xlsx"
        workbook = Workbook()
        sheet = workbook.active
        sheet.title = "在线商品"
        sheet.append(HEADERS)
        for row in rows:
            sheet.append(row)
        workbook.save(path)
        workbook.close()
        self.addCleanup(temp.cleanup)
        return path

    def test_聚合规格且不猜颜色尺码并按PDD去重(self):
        path = self.make_workbook(
            [
                sample_row("S-1", "P-1", "黑色", "M"),
                sample_row("S-1", "P-1", "白色", "L"),
                sample_row("S-2", "P-1", "均码", "红色"),
            ]
        )
        plan = catalog.prepare_batches(path, 200, "fill_missing")
        self.assertEqual((plan.groups, plan.skus, plan.unique_pdd_products), (2, 3, 1))
        self.assertEqual(len(plan.batches), 1)
        payload = json.loads(plan.batches[0].body)
        self.assertEqual(len(payload["pdd_products"]), 1)
        self.assertEqual(payload["pdd_products"][0]["url"], "https://mobile.yangkeduo.com/goods.html?goods_id=P-1")
        self.assertEqual(payload["shopee_skus"][2]["spec_raw"], "均码,红色")
        self.assertEqual(payload["shopee_skus"][2]["color"], "")
        self.assertEqual(payload["shopee_skus"][2]["size"], "")
        self.assertFalse(payload["shopee_skus"][2]["parse_ok"])

    def test_相同文件重复规划生成完全相同请求(self):
        path = self.make_workbook([sample_row("S-1", "P-1", "黑色", "M")])
        first = catalog.prepare_batches(path, 200, "fill_missing")
        second = catalog.prepare_batches(path, 200, "fill_missing")
        self.assertEqual(first.batches[0].batch_id, second.batches[0].batch_id)
        self.assertEqual(first.batches[0].body, second.batches[0].body)
        changed_policy = catalog.prepare_batches(path, 200, "insert_only")
        changed_size = catalog.prepare_batches(path, 1, "fill_missing")
        self.assertNotEqual(first.batches[0].batch_id, changed_policy.batches[0].batch_id)
        self.assertNotEqual(first.batches[0].batch_id, changed_size.batches[0].batch_id)

    def test_商品不跨批且批次满足限制(self):
        rows = []
        for product in range(3):
            rows.extend(
                sample_row(f"S-{product}", f"P-{product}", f"颜色-{sku}", "M")
                for sku in range(3)
            )
        path = self.make_workbook(rows)
        plan = catalog.prepare_batches(path, 2, "fill_missing")
        self.assertEqual([batch.shopee_products for batch in plan.batches], [2, 1])
        self.assertEqual([batch.shopee_skus for batch in plan.batches], [6, 3])
        for batch in plan.batches:
            self.assertLessEqual(batch.shopee_products + batch.pdd_products, 500)
            self.assertLessEqual(batch.shopee_skus, 5000)
            self.assertLessEqual(len(batch.body), 5 << 20)

    def test_重复规格明确停止(self):
        row = sample_row("S-1", "P-1", "黑色", "M")
        path = self.make_workbook([row, row])
        with self.assertRaisesRegex(catalog.CatalogImportError, "规格重复"):
            catalog.prepare_batches(path, 200, "fill_missing")

    def test_临时接口错误重试且不改变请求(self):
        batch = catalog.PreparedBatch("B-1", b"{}", 1, 1, 1, 1)
        success = {"batch_id": "B-1", "status": "succeeded"}
        with patch.object(
            catalog,
            "post_batch",
            side_effect=[
                catalog.CatalogHTTPError(503, "TEMPORARY", "稍后重试", True),
                success,
            ],
        ) as post:
            got = catalog.submit_with_retry(
                "https://example.com/api", "secret", batch, 10, 1, sleep=lambda _: None
            )
        self.assertEqual(got, success)
        self.assertEqual(post.call_count, 2)
        self.assertEqual(post.call_args_list[0].args[2], post.call_args_list[1].args[2])


if __name__ == "__main__":
    unittest.main()