diff --git a/docs/admin/10-商品目录接入接口.md b/docs/admin/10-商品目录接入接口.md index dc7a5c2..9bd1a7e 100644 --- a/docs/admin/10-商品目录接入接口.md +++ b/docs/admin/10-商品目录接入接口.md @@ -100,3 +100,49 @@ HTTP/HTTPS URL，最长 2048 字节；店铺名最长 500 个字符。Admin 只 旧版 Admin 不理解无真实 SKU ID 的新记录。应用回退前应暂停目录接口并确认没有此类 记录；不要删除 v8 列或把内部主键复制成伪造的外部 SKU ID。 + +## 6. 三个已处理工作簿的分批导入脚本 + +仓库根目录的 `tools/import_catalog_xlsx.py` 用于导入以下三个本地商业样本，文件本身 +不提交 Git： + +- `raw_data/shopee_chanpin_1_已处理.xlsx` +- `raw_data/shopee_chanpin_2_已处理.xlsx` +- `raw_data/shopee_chanpin_3_已处理.xlsx` + +脚本需要 Python 3.10+ 和 `openpyxl`。从仓库根目录安装固定依赖并先预检： + +```powershell +python -m pip install -r tools/requirements-catalog-import.txt +python tools/import_catalog_xlsx.py --dry-run +``` + +预检会读取全部行、校验固定字段和关联、生成批次并检查 500 商品、5000 SKU、5 MB +三个接口限制，但不会读取 Token，也不会发送请求。当前三个样本的基准结果是 2075 个 +蝦皮商品、33049 条 SKU、11 个批次；样本变化后以脚本实际输出为准。 + +真实提交前先备份生产数据库，然后在当前 PowerShell 窗口设置环境变量并运行： + +```powershell +$env:CMAUTOBUY_CATALOG_BASE_URL = "https://buy.833729.com" +$env:CMAUTOBUY_CATALOG_TOKEN = Read-Host "请输入商品目录专用 Token" +python tools/import_catalog_xlsx.py +Remove-Item Env:CMAUTOBUY_CATALOG_TOKEN +``` + +脚本默认使用 `fill_missing`，不会用空字段清除现有数据，也不会覆盖人工字段。文件没有 +真实蝦皮 SKU ID和可靠的颜色/尺码维度名称，因此脚本只按两个变种列的原顺序生成 +`spec_raw`，不猜测 `sku_id`、颜色或尺码；PDD 只提交确定存在的商品 ID、规范化 URL、 +店铺名及关联，不伪造 PDD 标题、价格或规格。 + +每个批次 ID 由转换器版本、文件内容哈希、批量大小、更新策略和批次序号确定， +`observed_at` 取批内 Excel 最大更新时间。因此同一文件和参数重新运行会提交完全相同的请求：已成功批次返回 +`replayed=true`，可用于断点续传。临时网络错误、429 和可重试 5xx 会指数退避；校验、 +鉴权或业务冲突会立即停止，不会跳过失败批次继续写后续数据。 + +需要只导入指定文件时，把路径作为位置参数传入；要改变更新策略必须显式指定： + +```powershell +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 +``` diff --git a/tools/import_catalog_xlsx.py b/tools/import_catalog_xlsx.py new file mode 100644 index 0000000..f7bd308 --- /dev/null +++ b/tools/import_catalog_xlsx.py @@ -0,0 +1,576 @@ +#!/usr/bin/env python3 +"""把已处理的蝦皮商品工作簿分批提交到 Admin 商品目录接口。 + +Token 只从 CMAUTOBUY_CATALOG_TOKEN 环境变量读取。相同文件和参数会生成完全 +相同的请求，可直接重新运行：服务端会幂等重放已经成功的批次。 +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +import time +from collections import OrderedDict +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable, Iterable, Sequence +from urllib.error import HTTPError, URLError +from urllib.parse import parse_qs, urlencode, urlparse +from urllib.request import Request, urlopen + +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" +SHEET_NAME = "在线商品" +MAX_CATALOG_PRODUCTS = 500 +MAX_CATALOG_SKUS = 5000 +MAX_BODY_BYTES = 5 << 20 +CHINA_TIMEZONE = timezone(timedelta(hours=8)) +DEFAULT_FILES = ( + Path("raw_data/shopee_chanpin_1_已处理.xlsx"), + Path("raw_data/shopee_chanpin_2_已处理.xlsx"), + Path("raw_data/shopee_chanpin_3_已处理.xlsx"), +) +REQUIRED_HEADERS = ( + "商品ID", + "商品标题", + "主货号", + "货源URL", + "货源ID", + "货源店铺名", + "商品状态", + "店铺显示名", + "更新时间", + "缩略图URL", + "变种属性值一", + "变种属性值二", +) + + +class CatalogImportError(RuntimeError): + """表示源文件、批次或接口响应不符合导入要求。""" + + +class CatalogHTTPError(CatalogImportError): + """只保留安全的接口错误字段，不保存请求头或 Token。""" + + def __init__(self, status: int, code: str, message: str, retryable: bool): + super().__init__(f"HTTP {status} {code}: {message}") + self.status = status + self.code = code + self.retryable = retryable + + +@dataclass +class ProductGroup: + shopee_product: dict[str, Any] + pdd_product: dict[str, Any] + association: dict[str, str] + updated_at: datetime + shopee_skus: list[dict[str, Any]] = field(default_factory=list) + seen_specs: set[str] = field(default_factory=set) + + +@dataclass(frozen=True) +class PreparedBatch: + batch_id: str + body: bytes + shopee_products: int + shopee_skus: int + pdd_products: int + associations: int + + +@dataclass(frozen=True) +class FilePlan: + path: Path + sha256: str + groups: int + unique_pdd_products: int + skus: int + batches: tuple[PreparedBatch, ...] + + +def cell_text(value: Any) -> str: + """把 Excel 单元格变成稳定文本，ID 不出现无意义的 .0。""" + + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, int): + return str(value) + if isinstance(value, float) and value.is_integer(): + return str(int(value)) + return str(value).strip() + + +def parse_observed_at(value: Any, path: Path, row_number: int) -> datetime: + if isinstance(value, datetime): + parsed = value + else: + text = cell_text(value) + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise CatalogImportError( + f"{path.name} 第 {row_number} 行更新时间不是有效日期: {text!r}" + ) from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=CHINA_TIMEZONE) + return parsed + + +def stable_file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def canonical_pdd_url(goods_id: str, raw_url: str, path: Path, row_number: int) -> str: + parsed = urlparse(raw_url) + query_id = parse_qs(parsed.query).get("goods_id", [""])[0].strip() + if parsed.scheme not in {"http", "https"} or query_id != goods_id: + raise CatalogImportError( + f"{path.name} 第 {row_number} 行货源URL与货源ID不一致" + ) + return "https://mobile.yangkeduo.com/goods.html?" + urlencode( + {"goods_id": goods_id} + ) + + +def combine_spec_raw(first: Any, second: Any, path: Path, row_number: int) -> str: + values = [text for text in (cell_text(first), cell_text(second)) if text] + if not values: + raise CatalogImportError(f"{path.name} 第 {row_number} 行两个变种属性都为空") + # 文件没有维度名称，严格保留两个变种列的先后顺序，不猜颜色和尺码。 + return ",".join(values) + + +def read_product_groups(path: Path) -> list[ProductGroup]: + """读取一个工作簿，按蝦皮商品聚合并严格检查重复行的一致性。""" + + if not path.is_file(): + raise CatalogImportError(f"找不到文件: {path}") + workbook = load_workbook(path, read_only=True, data_only=True) + try: + if SHEET_NAME not in workbook.sheetnames: + raise CatalogImportError(f"{path.name} 缺少工作表 {SHEET_NAME!r}") + worksheet = workbook[SHEET_NAME] + rows = worksheet.iter_rows(values_only=True) + try: + headers = [cell_text(value) for value in next(rows)] + except StopIteration as exc: + raise CatalogImportError(f"{path.name} 是空工作簿") from exc + missing = [name for name in REQUIRED_HEADERS if name not in headers] + if missing: + raise CatalogImportError(f"{path.name} 缺少列: {', '.join(missing)}") + indexes = {name: headers.index(name) for name in REQUIRED_HEADERS} + groups: OrderedDict[str, ProductGroup] = OrderedDict() + pdd_definitions: dict[str, dict[str, Any]] = {} + + for row_number, row in enumerate(rows, start=2): + goods_id = cell_text(row[indexes["商品ID"]]) + title = cell_text(row[indexes["商品标题"]]) + pdd_goods_id = cell_text(row[indexes["货源ID"]]) + raw_pdd_url = cell_text(row[indexes["货源URL"]]) + if not goods_id or not title or not pdd_goods_id or not raw_pdd_url: + raise CatalogImportError( + f"{path.name} 第 {row_number} 行商品ID、标题、货源ID和货源URL不能为空" + ) + pdd_url = canonical_pdd_url( + pdd_goods_id, raw_pdd_url, path, row_number + ) + observed_at = parse_observed_at( + row[indexes["更新时间"]], path, row_number + ) + shopee_product = { + "goods_id": goods_id, + "title": title, + "status": cell_text(row[indexes["商品状态"]]), + "main_sku_code": cell_text(row[indexes["主货号"]]), + "image_url": cell_text(row[indexes["缩略图URL"]]), + "shop_name": cell_text(row[indexes["店铺显示名"]]), + } + pdd_product = { + "goods_id": pdd_goods_id, + "url": pdd_url, + "title": "", + "shop_name": cell_text(row[indexes["货源店铺名"]]), + "dimensions": [], + "skus": [], + } + previous_pdd = pdd_definitions.get(pdd_goods_id) + if previous_pdd is not None and previous_pdd != pdd_product: + raise CatalogImportError( + f"{path.name} 第 {row_number} 行 PDD 商品 {pdd_goods_id} 的店铺或URL不一致" + ) + pdd_definitions[pdd_goods_id] = pdd_product + group = groups.get(goods_id) + if group is None: + group = ProductGroup( + shopee_product=shopee_product, + pdd_product=pdd_product, + association={ + "shopee_goods_id": goods_id, + "pdd_goods_id": pdd_goods_id, + }, + updated_at=observed_at, + ) + groups[goods_id] = group + else: + if ( + group.shopee_product != shopee_product + or group.pdd_product != pdd_product + ): + raise CatalogImportError( + f"{path.name} 第 {row_number} 行商品 {goods_id} 的固定字段不一致" + ) + group.updated_at = max(group.updated_at, observed_at) + + spec_raw = combine_spec_raw( + row[indexes["变种属性值一"]], + row[indexes["变种属性值二"]], + 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) + group.shopee_skus.append( + { + "sku_id": None, + "goods_id": goods_id, + "spec_raw": spec_raw, + "color": "", + "size": "", + "advice": "", + "parse_ok": False, + "sku_code": "", + } + ) + if not groups: + raise CatalogImportError(f"{path.name} 没有数据行") + return list(groups.values()) + finally: + workbook.close() + + +def encode_payload(payload: dict[str, Any]) -> bytes: + return json.dumps( + payload, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + + +def make_payload( + batch_namespace: str, + batch_number: int, + groups: Sequence[ProductGroup], + update_policy: str, +) -> 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}" + 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_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], + } + return payload, encode_payload(payload) + + +def validate_batch(payload: dict[str, Any], body: bytes) -> None: + product_count = len(payload["shopee_products"]) + len(payload["pdd_products"]) + sku_count = len(payload["shopee_skus"]) + if product_count > MAX_CATALOG_PRODUCTS: + raise CatalogImportError( + f"批次 {payload['batch_id']} 商品总数 {product_count} 超过 {MAX_CATALOG_PRODUCTS}" + ) + if sku_count > MAX_CATALOG_SKUS: + raise CatalogImportError( + f"批次 {payload['batch_id']} SKU 数 {sku_count} 超过 {MAX_CATALOG_SKUS}" + ) + if len(body) > MAX_BODY_BYTES: + raise CatalogImportError( + f"批次 {payload['batch_id']} JSON 为 {len(body)} 字节，超过 {MAX_BODY_BYTES}" + ) + + +def prepare_batches( + path: Path, max_shopee_products: int, update_policy: str +) -> FilePlan: + if not 1 <= max_shopee_products <= MAX_CATALOG_PRODUCTS: + raise CatalogImportError("每批蝦皮商品数必须在 1 到 500 之间") + file_hash = stable_file_sha256(path) + batch_namespace = hashlib.sha256( + f"{file_hash}\0{max_shopee_products}\0{update_policy}".encode("utf-8") + ).hexdigest()[:16] + groups = read_product_groups(path) + 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 + ) + validate_batch(payload, body) + return PreparedBatch( + batch_id=payload["batch_id"], + body=body, + shopee_products=len(payload["shopee_products"]), + shopee_skus=len(payload["shopee_skus"]), + pdd_products=len(payload["pdd_products"]), + associations=len(payload["associations"]), + ) + + for group in groups: + candidate = [*current, group] + payload, body = make_payload( + batch_namespace, len(batches) + 1, candidate, update_policy + ) + exceeds = ( + len(candidate) > max_shopee_products + or len(payload["shopee_products"]) + len(payload["pdd_products"]) + > MAX_CATALOG_PRODUCTS + or len(payload["shopee_skus"]) > MAX_CATALOG_SKUS + or len(body) > MAX_BODY_BYTES + ) + if exceeds: + if not current: + validate_batch(payload, body) + raise CatalogImportError( + f"商品 {group.shopee_product['goods_id']} 单独一批仍超过限制" + ) + batches.append(finish_batch(current)) + current = [group] + single_payload, single_body = make_payload( + batch_namespace, len(batches) + 1, current, update_policy + ) + validate_batch(single_payload, single_body) + else: + current = candidate + if current: + batches.append(finish_batch(current)) + + return FilePlan( + path=path, + sha256=file_hash, + groups=len(groups), + unique_pdd_products=len( + {group.pdd_product["goods_id"] for group in groups} + ), + skus=sum(len(group.shopee_skus) for group in groups), + batches=tuple(batches), + ) + + +def catalog_endpoint(base_url: str) -> str: + cleaned = base_url.strip().rstrip("/") + if not cleaned: + raise CatalogImportError( + f"请用 --base-url 或 {CATALOG_BASE_URL_ENV} 指定 Admin 地址" + ) + parsed = urlparse(cleaned) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.netloc + or parsed.username is not None + or parsed.password is not None + ): + raise CatalogImportError("Admin 地址必须是完整的 HTTP/HTTPS URL") + if cleaned.endswith(CATALOG_PATH): + return cleaned + return cleaned + CATALOG_PATH + + +def decode_error(status: int, raw: bytes) -> CatalogHTTPError: + code = "HTTP_ERROR" + message = "接口返回错误" + retryable = status in {429, 500, 502, 503, 504} + try: + value = json.loads(raw.decode("utf-8")) + error = value.get("error", {}) + code = str(error.get("code") or code) + message = str(error.get("message") or message) + retryable = bool(error.get("retryable", retryable)) + except (UnicodeDecodeError, json.JSONDecodeError, AttributeError): + pass + return CatalogHTTPError(status, code, message, retryable) + + +def post_batch(endpoint: str, token: str, body: bytes, timeout: float) -> dict[str, Any]: + request = Request( + endpoint, + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + ) + try: + with urlopen(request, timeout=timeout) as response: + raw = response.read() + status = response.status + except HTTPError as exc: + raise decode_error(exc.code, exc.read()) from None + except URLError as exc: + # URLError 不包含请求头；不要把 Request 对象放进异常文本。 + raise CatalogHTTPError(0, "NETWORK_ERROR", str(exc.reason), True) from None + if status != 200: + raise decode_error(status, raw) + try: + value = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CatalogHTTPError(status, "INVALID_RESPONSE", "响应不是合法 JSON", False) from exc + if not isinstance(value, dict): + raise CatalogHTTPError(status, "INVALID_RESPONSE", "响应必须是 JSON 对象", False) + return value + + +def submit_with_retry( + endpoint: str, + token: str, + batch: PreparedBatch, + timeout: float, + max_retries: int, + sleep: Callable[[float], None] = time.sleep, +) -> dict[str, Any]: + for attempt in range(max_retries + 1): + try: + return post_batch(endpoint, token, batch.body, timeout) + except CatalogHTTPError as exc: + if not exc.retryable or attempt >= max_retries: + raise + delay = min(2**attempt, 30) + print( + f" {batch.batch_id} 暂时失败 {exc.code}，{delay} 秒后重试 " + f"({attempt + 1}/{max_retries})", + file=sys.stderr, + ) + sleep(delay) + raise AssertionError("unreachable") + + +def print_plan(plan: FilePlan) -> None: + max_body = max(len(batch.body) for batch in plan.batches) + print( + f"{plan.path.name}: 蝦皮商品 {plan.groups}，SKU {plan.skus}，" + f"PDD 商品 {plan.unique_pdd_products}，批次 {len(plan.batches)}，" + f"最大请求 {max_body} 字节" + ) + + +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) + for path in files + ] + for plan in plans: + print_plan(plan) + total_batches = sum(len(plan.batches) for plan in plans) + if args.dry_run: + print(f"预检完成：共 {total_batches} 个批次，未发送任何请求。") + return 0 + + token = os.environ.get(CATALOG_TOKEN_ENV, "").strip() + if not token: + raise CatalogImportError(f"缺少环境变量 {CATALOG_TOKEN_ENV}") + endpoint = catalog_endpoint( + args.base_url or os.environ.get(CATALOG_BASE_URL_ENV, "") + ) + completed = 0 + for plan in plans: + print(f"开始提交 {plan.path.name}") + for number, batch in enumerate(plan.batches, start=1): + response = submit_with_retry( + endpoint, + token, + batch, + args.timeout, + args.max_retries, + ) + if response.get("batch_id") != batch.batch_id or response.get("status") != "succeeded": + raise CatalogImportError( + f"批次 {batch.batch_id} 响应身份或状态不正确" + ) + completed += 1 + replayed = "（幂等重放）" if response.get("replayed") else "" + print( + f" [{number}/{len(plan.batches)}] {batch.batch_id} 成功{replayed}" + ) + print(f"提交完成：{completed}/{total_batches} 个批次成功。") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="把已处理的蝦皮 Excel 分批提交到 Admin 商品目录接口" + ) + parser.add_argument( + "files", + nargs="*", + help="要导入的 xlsx；省略时使用 raw_data 下三个已处理文件", + ) + parser.add_argument( + "--base-url", + default="", + help=f"Admin 根地址；也可用环境变量 {CATALOG_BASE_URL_ENV}", + ) + parser.add_argument( + "--batch-products", + type=int, + default=200, + help="每批最多几个蝦皮商品，默认 200", + ) + parser.add_argument( + "--update-policy", + choices=("insert_only", "fill_missing", "overwrite_same_source"), + default="fill_missing", + help="接口更新策略，默认 fill_missing", + ) + parser.add_argument("--timeout", type=float, default=60, help="单次请求超时秒数") + parser.add_argument( + "--max-retries", type=int, default=4, help="临时错误最大重试次数，默认 4" + ) + parser.add_argument( + "--dry-run", action="store_true", help="只解析和检查批次，不发送请求" + ) + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + if args.timeout <= 0 or args.max_retries < 0: + parser.error("timeout 必须大于 0，max-retries 不能为负数") + try: + return run(args) + except CatalogImportError as exc: + print(f"导入失败：{exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/requirements-catalog-import.txt b/tools/requirements-catalog-import.txt new file mode 100644 index 0000000..26ca078 --- /dev/null +++ b/tools/requirements-catalog-import.txt @@ -0,0 +1 @@ +openpyxl==3.1.3 diff --git a/tools/test_import_catalog_xlsx.py b/tools/test_import_catalog_xlsx.py new file mode 100644 index 0000000..8600881 --- /dev/null +++ b/tools/test_import_catalog_xlsx.py @@ -0,0 +1,160 @@ +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()