733 lines
27 KiB
Python
733 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
"""把已处理的蝦皮商品工作簿分批提交到 Admin 商品目录接口。
|
||
|
||
Token 只从 CMAUTOBUY_CATALOG_TOKEN 环境变量读取。相同文件和参数会生成完全
|
||
相同的请求，可直接重新运行：服务端会幂等重放已经成功的批次。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from collections import Counter, 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"
|
||
RAW_CONVERTER_VERSION = "xlsx-v1"
|
||
FORMALIZE_CONVERTER_VERSION = "xlsx-spec-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)
|
||
variant_values: list[tuple[str, str]] = 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
|
||
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。"""
|
||
|
||
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)
|
||
|
||
first_variant = cell_text(row[indexes["变种属性值一"]])
|
||
second_variant = cell_text(row[indexes["变种属性值二"]])
|
||
spec_raw = combine_spec_raw(
|
||
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,
|
||
"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,
|
||
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}"
|
||
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": (
|
||
[] 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": [] if sku_only else list(pdd_products.values()),
|
||
"associations": (
|
||
[] if sku_only else [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,
|
||
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"{converter_version}\0{file_hash}\0{max_shopee_products}\0{update_policy}".encode(
|
||
"utf-8"
|
||
)
|
||
).hexdigest()[:16]
|
||
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,
|
||
converter_version,
|
||
formalize_specs,
|
||
)
|
||
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,
|
||
converter_version,
|
||
formalize_specs,
|
||
)
|
||
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,
|
||
converter_version,
|
||
formalize_specs,
|
||
)
|
||
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(source_groups),
|
||
unique_pdd_products=len(
|
||
{group.pdd_product["goods_id"] for group in source_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),
|
||
)
|
||
|
||
|
||
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), 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,
|
||
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} 个批次，未发送任何请求。")
|
||
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="只解析和检查批次，不发送请求"
|
||
)
|
||
parser.add_argument(
|
||
"--formalize-specs",
|
||
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())
|