From 30bdab3e2e9680b11510dae5b15c943769c438c9 Mon Sep 17 00:00:00 2001 From: chengma Date: Tue, 11 Aug 2026 14:31:13 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=A1=AE=E5=AE=9A=E6=80=A7=E8=BD=AC?= =?UTF-8?q?=E6=8D=A2=E8=9D=A6=E7=9A=AE=E5=BE=85=E8=A1=A5=E8=A7=84=E6=A0=BC?= =?UTF-8?q?=20(#152)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/admin/10-商品目录接入接口.md | 40 ++++++ tools/import_catalog_xlsx.py | 198 ++++++++++++++++++++++++++---- tools/test_import_catalog_xlsx.py | 95 ++++++++++++++ 3 files changed, 312 insertions(+), 21 deletions(-) diff --git a/docs/admin/10-商品目录接入接口.md b/docs/admin/10-商品目录接入接口.md index 9bd1a7e..c307051 100644 --- a/docs/admin/10-商品目录接入接口.md +++ b/docs/admin/10-商品目录接入接口.md @@ -146,3 +146,43 @@ Remove-Item Env:CMAUTOBUY_CATALOG_TOKEN python tools/import_catalog_xlsx.py raw_data/shopee_chanpin_3_已处理.xlsx python tools/import_catalog_xlsx.py --update-policy overwrite_same_source --dry-run ``` + +### 6.1 将确定的待补规格转换为正式规格 + +三个工作簿没有颜色、尺码的维度名称，不能固定认为第一列是颜色、第二列是尺码。 +需要补齐已经导入的待补规格时，使用独立的确定性转换模式： + +```powershell +python tools/import_catalog_xlsx.py --formalize-specs --dry-run +``` + +该模式按商品整体判断两列，不逐行改变列含义。只有整列取值都包含明确尺码证据时， +才把它作为尺码列；另一列作为颜色/款式。单维商品只有在全部取值都能明确识别为 +尺码或颜色时才转换。证据冲突、列数不一致或含义不明的商品不会进入写请求，继续 +显示“待补”。 + +当前三个样本的预检基准如下： + +| 结果 | 商品数 | SKU 数 | +|---|---:|---:| +| 可以确定并正式化 | 1920 | 31178 | +| 含义不确定，继续待补 | 155 | 1871 | +| 源文件合计 | 2075 | 33049 | + +正式化模式强制使用 `fill_missing`：只补数据库中的空颜色/尺码，不覆盖人工记录或已有 +正式字段；`spec_raw`、`spec_key` 和真实 SKU ID 均不改变。它使用独立的 +`xlsx-spec-v1` 批次版本，相同文件和参数可以幂等重跑。 + +生产执行前必须重新备份数据库，并先在 MySQL 8.4、库名以 `_test` 结尾的测试库或 +生产备份副本演练。确认预检计数和前后不变量后，才在当前 PowerShell 进程设置 Token： + +```powershell +$env:CMAUTOBUY_CATALOG_BASE_URL = "https://buy.833729.com" +$env:CMAUTOBUY_CATALOG_TOKEN = Read-Host "请输入商品目录专用 Token" +python tools/import_catalog_xlsx.py --formalize-specs +Remove-Item Env:CMAUTOBUY_CATALOG_TOKEN +``` + +执行后核对商品总数仍为 2075、SKU 总数仍为 33049，`spec_raw/spec_key` 不变，正式 +SKU 增量与预检一致，歧义 SKU 仍为待补，且没有失败批次。Token 和数据库凭据不得 +写入命令文件、日志、工单或任务归档。 diff --git a/tools/import_catalog_xlsx.py b/tools/import_catalog_xlsx.py index f7bd308..ffdd40f 100644 --- a/tools/import_catalog_xlsx.py +++ b/tools/import_catalog_xlsx.py @@ -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 diff --git a/tools/test_import_catalog_xlsx.py b/tools/test_import_catalog_xlsx.py index 8600881..ca15397 100644 --- a/tools/test_import_catalog_xlsx.py +++ b/tools/test_import_catalog_xlsx.py @@ -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):