feat: 增加商品目录 Excel 分批提交脚本 (#146)

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
+```

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())

tools/requirements-catalog-import.txt

@@ -0,0 +1 @@
+openpyxl==3.1.3

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()