chengma/cmautobuy
2
0
Fork 0
Code Issues 89 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 确定性转换蝦皮待补规格 (#152)

Browse Source
This commit is contained in:
chengma
2026-08-11 14:31:13 +08:00
parent c68efe1a55
commit 30bdab3e2e
3 changed files with 312 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/import_catalog_xlsx.py
+177 -21
View File
@@ -11,9 +11,10 @@ import argparse
import hashlib
import json
import os
import re
import sys
import time
from collections import OrderedDict
from collections import Counter, OrderedDict
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
@@ -28,7 +29,8 @@ from openpyxl import load_workbook
CATALOG_PATH = "/api/v1/integrations/catalog/batches"
CATALOG_TOKEN_ENV = "CMAUTOBUY_CATALOG_TOKEN"
CATALOG_BASE_URL_ENV = "CMAUTOBUY_CATALOG_BASE_URL"
CONVERTER_VERSION = "xlsx-v1"
RAW_CONVERTER_VERSION = "xlsx-v1"
FORMALIZE_CONVERTER_VERSION = "xlsx-spec-v1"
SHEET_NAME = "在线商品"
MAX_CATALOG_PRODUCTS = 500
MAX_CATALOG_SKUS = 5000
@@ -76,6 +78,7 @@ class ProductGroup:
association: dict[str, str]
updated_at: datetime
shopee_skus: list[dict[str, Any]] = field(default_factory=list)
variant_values: list[tuple[str, str]] = field(default_factory=list)
seen_specs: set[str] = field(default_factory=set)
@@ -96,9 +99,81 @@ class FilePlan:
groups: int
unique_pdd_products: int
skus: int
formalized_groups: int
formalized_skus: int
pending_groups: int
pending_skus: int
decision_counts: tuple[tuple[str, int], ...]
batches: tuple[PreparedBatch, ...]
SIZE_MARKER_PATTERN = re.compile(
r"(?:尺[碼码]|[均小中大][碼码]|(?:^|[^A-Za-z])(?:[2-9]?XL|X{0,4}[SML])(?:$|[^A-Za-z])|"
r"公斤|(?:^|[^公])斤|體重|体重|建議體重|建议体重|SIZE)",
re.IGNORECASE,
)
COLOR_MARKER_PATTERN = re.compile(
r"(?:黑色|白色|紅色|红色|粉色|藍色|蓝色|綠色|绿色|黃色|黄色|紫色|"
r"灰色|棕色|咖啡色|杏色|米色|卡其色|酒紅|酒红|膚色|肤色|裸色|"
r"金色|銀色|银色|橙色|橘色|香檳色|香槟色|透明色)",
re.IGNORECASE,
)
def dimension_has_only_marker(values: Iterable[str], pattern: re.Pattern[str]) -> bool:
"""所有非空取值都有同类明确标记时，才认为该维度可以确定。"""
unique_values = {value.strip() for value in values if value.strip()}
return bool(unique_values) and all(pattern.search(value) for value in unique_values)
def dimension_has_any_marker(values: Iterable[str], pattern: re.Pattern[str]) -> bool:
"""判断一列中是否出现过某类明确标记，用于发现维度证据冲突。"""
return any(pattern.search(value.strip()) for value in values if value.strip())
def formalize_product_specs(group: ProductGroup) -> str:
"""确定一个商品两列规格的含义；不确定时不修改任何 SKU。"""
first_values = [first for first, _ in group.variant_values]
second_values = [second for _, second in group.variant_values]
has_second = [bool(value.strip()) for value in second_values]
if any(has_second) and not all(has_second):
return "歧义：同一商品规格列数不一致"
if not any(has_second):
first_is_size = dimension_has_only_marker(first_values, SIZE_MARKER_PATTERN)
first_is_color = dimension_has_only_marker(first_values, COLOR_MARKER_PATTERN)
if first_is_size == first_is_color:
return "歧义：单维规格含义不确定"
target = "size" if first_is_size else "color"
for sku, first in zip(group.shopee_skus, first_values):
sku[target] = first
sku["parse_ok"] = True
return "确定：单维尺码" if first_is_size else "确定：单维颜色"
first_is_size = dimension_has_only_marker(first_values, SIZE_MARKER_PATTERN)
second_is_size = dimension_has_only_marker(second_values, SIZE_MARKER_PATTERN)
first_has_size_evidence = dimension_has_any_marker(
first_values, SIZE_MARKER_PATTERN
)
second_has_size_evidence = dimension_has_any_marker(
second_values, SIZE_MARKER_PATTERN
)
first_is_unique_size = first_is_size and not second_has_size_evidence
second_is_unique_size = second_is_size and not first_has_size_evidence
if first_is_unique_size == second_is_unique_size:
return "歧义：无法唯一确定尺码列"
size_index = 0 if first_is_unique_size else 1
for sku, values in zip(group.shopee_skus, group.variant_values):
sku["size"] = values[size_index]
sku["color"] = values[1 - size_index]
sku["parse_ok"] = True
return "确定：第一列为尺码" if first_is_size else "确定：第二列为尺码"
def cell_text(value: Any) -> str:
"""把 Excel 单元格变成稳定文本，ID 不出现无意义的 .0。"""
@@ -238,17 +313,18 @@ def read_product_groups(path: Path) -> list[ProductGroup]:
)
group.updated_at = max(group.updated_at, observed_at)
first_variant = cell_text(row[indexes["变种属性值一"]])
second_variant = cell_text(row[indexes["变种属性值二"]])
spec_raw = combine_spec_raw(
row[indexes["变种属性值一"]],
row[indexes["变种属性值二"]],
path,
row_number,
first_variant, second_variant, path, row_number
)
if spec_raw in group.seen_specs:
raise CatalogImportError(
f"{path.name} 第 {row_number} 行商品 {goods_id} 规格重复: {spec_raw!r}"
)
group.seen_specs.add(spec_raw)
# 单独保存两列原值，不能以后再按逗号拆 spec_raw：规格值本身可能含逗号。
group.variant_values.append((first_variant, second_variant))
group.shopee_skus.append(
{
"sku_id": None,
@@ -279,21 +355,27 @@ def make_payload(
batch_number: int,
groups: Sequence[ProductGroup],
update_policy: str,
converter_version: str,
sku_only: bool,
) -> tuple[dict[str, Any], bytes]:
pdd_products: OrderedDict[str, dict[str, Any]] = OrderedDict()
for group in groups:
pdd_products.setdefault(group.pdd_product["goods_id"], group.pdd_product)
batch_id = f"{CONVERTER_VERSION}-{batch_namespace}-{batch_number:04d}"
batch_id = f"{converter_version}-{batch_namespace}-{batch_number:04d}"
observed_at = max(group.updated_at for group in groups).isoformat(timespec="seconds")
payload = {
"schema_version": 1,
"batch_id": batch_id,
"observed_at": observed_at,
"update_policy": update_policy,
"shopee_products": [group.shopee_product for group in groups],
"shopee_products": (
[] if sku_only else [group.shopee_product for group in groups]
),
"shopee_skus": [sku for group in groups for sku in group.shopee_skus],
"pdd_products": list(pdd_products.values()),
"associations": [group.association for group in groups],
"pdd_products": [] if sku_only else list(pdd_products.values()),
"associations": (
[] if sku_only else [group.association for group in groups]
),
}
return payload, encode_payload(payload)
@@ -316,21 +398,46 @@ def validate_batch(payload: dict[str, Any], body: bytes) -> None:
def prepare_batches(
path: Path, max_shopee_products: int, update_policy: str
path: Path,
max_shopee_products: int,
update_policy: str,
formalize_specs: bool = False,
) -> FilePlan:
if not 1 <= max_shopee_products <= MAX_CATALOG_PRODUCTS:
raise CatalogImportError("每批蝦皮商品数必须在 1 到 500 之间")
if formalize_specs and update_policy != "fill_missing":
raise CatalogImportError("正式化规格只能使用 fill_missing 更新策略")
file_hash = stable_file_sha256(path)
converter_version = (
FORMALIZE_CONVERTER_VERSION if formalize_specs else RAW_CONVERTER_VERSION
)
batch_namespace = hashlib.sha256(
f"{file_hash}\0{max_shopee_products}\0{update_policy}".encode("utf-8")
f"{converter_version}\0{file_hash}\0{max_shopee_products}\0{update_policy}".encode(
"utf-8"
)
).hexdigest()[:16]
groups = read_product_groups(path)
source_groups = read_product_groups(path)
decision_counts: Counter[str] = Counter()
if formalize_specs:
groups = []
for group in source_groups:
decision = formalize_product_specs(group)
decision_counts[decision] += 1
if decision.startswith("确定："):
groups.append(group)
else:
groups = source_groups
batches: list[PreparedBatch] = []
current: list[ProductGroup] = []
def finish_batch(selected: Sequence[ProductGroup]) -> PreparedBatch:
payload, body = make_payload(
batch_namespace, len(batches) + 1, selected, update_policy
batch_namespace,
len(batches) + 1,
selected,
update_policy,
converter_version,
formalize_specs,
)
validate_batch(payload, body)
return PreparedBatch(
@@ -345,7 +452,12 @@ def prepare_batches(
for group in groups:
candidate = [*current, group]
payload, body = make_payload(
batch_namespace, len(batches) + 1, candidate, update_policy
batch_namespace,
len(batches) + 1,
candidate,
update_policy,
converter_version,
formalize_specs,
)
exceeds = (
len(candidate) > max_shopee_products
@@ -363,7 +475,12 @@ def prepare_batches(
batches.append(finish_batch(current))
current = [group]
single_payload, single_body = make_payload(
batch_namespace, len(batches) + 1, current, update_policy
batch_namespace,
len(batches) + 1,
current,
update_policy,
converter_version,
formalize_specs,
)
validate_batch(single_payload, single_body)
else:
@@ -374,11 +491,23 @@ def prepare_batches(
return FilePlan(
path=path,
sha256=file_hash,
groups=len(groups),
groups=len(source_groups),
unique_pdd_products=len(
{group.pdd_product["goods_id"] for group in groups}
{group.pdd_product["goods_id"] for group in source_groups}
),
skus=sum(len(group.shopee_skus) for group in groups),
skus=sum(len(group.shopee_skus) for group in source_groups),
formalized_groups=len(groups) if formalize_specs else 0,
formalized_skus=(
sum(len(group.shopee_skus) for group in groups) if formalize_specs else 0
),
pending_groups=(len(source_groups) - len(groups)) if formalize_specs else 0,
pending_skus=(
sum(len(group.shopee_skus) for group in source_groups)
- sum(len(group.shopee_skus) for group in groups)
if formalize_specs
else 0
),
decision_counts=tuple(sorted(decision_counts.items())),
batches=tuple(batches),
)
@@ -473,22 +602,44 @@ def submit_with_retry(
def print_plan(plan: FilePlan) -> None:
max_body = max(len(batch.body) for batch in plan.batches)
max_body = max((len(batch.body) for batch in plan.batches), default=0)
print(
f"{plan.path.name}: 蝦皮商品 {plan.groups}，SKU {plan.skus}，"
f"PDD 商品 {plan.unique_pdd_products}，批次 {len(plan.batches)}，"
f"最大请求 {max_body} 字节"
)
if plan.decision_counts:
print(
f" 可正式化商品 {plan.formalized_groups}、SKU {plan.formalized_skus}；"
f"保持待补商品 {plan.pending_groups}、SKU {plan.pending_skus}"
)
for decision, count in plan.decision_counts:
print(f" {decision}：{count} 个商品")
def run(args: argparse.Namespace) -> int:
files = [Path(value) for value in args.files] if args.files else list(DEFAULT_FILES)
plans = [
prepare_batches(path, args.batch_products, args.update_policy)
prepare_batches(
path,
args.batch_products,
args.update_policy,
formalize_specs=args.formalize_specs,
)
for path in files
]
for plan in plans:
print_plan(plan)
if args.formalize_specs:
print(
"正式化汇总："
f"源商品 {sum(plan.groups for plan in plans)}、"
f"源 SKU {sum(plan.skus for plan in plans)}；"
f"可正式化商品 {sum(plan.formalized_groups for plan in plans)}、"
f"SKU {sum(plan.formalized_skus for plan in plans)}；"
f"保持待补商品 {sum(plan.pending_groups for plan in plans)}、"
f"SKU {sum(plan.pending_skus for plan in plans)}"
)
total_batches = sum(len(plan.batches) for plan in plans)
if args.dry_run:
print(f"预检完成：共 {total_batches} 个批次，未发送任何请求。")
@@ -557,6 +708,11 @@ def build_parser() -> argparse.ArgumentParser:
parser.add_argument(
"--dry-run", action="store_true", help="只解析和检查批次，不发送请求"
)
parser.add_argument(
"--formalize-specs",
action="store_true",
help="只提交能够确定颜色/尺码维度的规格，歧义商品保持待补",
)
return parser
tools/test_import_catalog_xlsx.py
+95
View File
@@ -115,6 +115,101 @@ class CatalogXLSXTest(unittest.TestCase):
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):