256 lines
9.7 KiB
Python
256 lines
9.7 KiB
Python
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()
|