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): path = self.make_workbook( [ sample_row("S-1", "P-1", "黑色", "M码"), sample_row("S-1", "P-1", "白色", "L码"), sample_row("S-2", "P-2", "小码", "红色"), sample_row("S-2", "P-2", "大码", "蓝色"), ] ) plan = catalog.prepare_batches(path, 200, "fill_missing", True) payload = json.loads(plan.batches[0].body) skus = payload["shopee_skus"] self.assertEqual((plan.formalized_groups, plan.formalized_skus), (2, 4)) self.assertEqual((plan.pending_groups, plan.pending_skus), (0, 0)) self.assertEqual(payload["shopee_products"], []) self.assertEqual(payload["pdd_products"], []) self.assertEqual(payload["associations"], []) self.assertEqual((skus[0]["color"], skus[0]["size"]), ("黑色", "M码")) self.assertEqual((skus[2]["color"], skus[2]["size"]), ("红色", "小码")) self.assertTrue(all(sku["parse_ok"] for sku in skus)) def test_正式化模式支持确定的单维规格(self): path = self.make_workbook( [ sample_row("S-1", "P-1", "均码【40-60公斤】", ""), sample_row("S-2", "P-2", "黑色", ""), ] ) plan = catalog.prepare_batches(path, 200, "fill_missing", True) payload = json.loads(plan.batches[0].body) by_goods = {sku["goods_id"]: sku for sku in payload["shopee_skus"]} self.assertEqual(by_goods["S-1"]["size"], "均码【40-60公斤】") self.assertEqual(by_goods["S-1"]["color"], "") self.assertEqual(by_goods["S-2"]["color"], "黑色") self.assertEqual(by_goods["S-2"]["size"], "") def test_正式化模式不提交歧义商品(self): path = self.make_workbook( [ sample_row("S-1", "P-1", "款式甲", "套餐甲"), sample_row("S-1", "P-1", "款式乙", "套餐乙"), sample_row("S-2", "P-2", "黑色", "M码"), ] ) plan = catalog.prepare_batches(path, 200, "fill_missing", True) payload = json.loads(plan.batches[0].body) self.assertEqual((plan.groups, plan.skus), (2, 3)) self.assertEqual((plan.formalized_groups, plan.formalized_skus), (1, 1)) self.assertEqual((plan.pending_groups, plan.pending_skus), (1, 2)) self.assertEqual( {sku["goods_id"] for sku in payload["shopee_skus"]}, {"S-2"} ) def test_同一商品列数不一致时整体保持待补(self): path = self.make_workbook( [ sample_row("S-1", "P-1", "黑色", "M码"), sample_row("S-1", "P-1", "白色", ""), ] ) plan = catalog.prepare_batches(path, 200, "fill_missing", True) self.assertEqual((plan.formalized_groups, plan.pending_groups), (0, 1)) self.assertEqual(plan.batches, ()) self.assertIn( ("歧义：同一商品规格列数不一致", 1), plan.decision_counts ) def test_另一列也出现尺码证据时不猜测(self): path = self.make_workbook( [ sample_row("S-1", "P-1", "小码", "黑色"), sample_row("S-1", "P-1", "大码", "白色均码"), ] ) plan = catalog.prepare_batches(path, 200, "fill_missing", True) self.assertEqual((plan.formalized_groups, plan.pending_groups), (0, 1)) self.assertEqual(plan.batches, ()) self.assertIn(("歧义：无法唯一确定尺码列", 1), plan.decision_counts) def test_正式化模式固定fill_missing且使用独立批次版本(self): path = self.make_workbook([sample_row("S-1", "P-1", "黑色", "M码")]) raw = catalog.prepare_batches(path, 200, "fill_missing") formal = catalog.prepare_batches(path, 200, "fill_missing", True) self.assertTrue(raw.batches[0].batch_id.startswith("xlsx-v1-")) self.assertTrue(formal.batches[0].batch_id.startswith("xlsx-spec-v1-")) self.assertNotEqual(raw.batches[0].batch_id, formal.batches[0].batch_id) with self.assertRaisesRegex(catalog.CatalogImportError, "fill_missing"): catalog.prepare_batches(path, 200, "overwrite_same_source", True) 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()