fix(dispatch): Shopee Packing List xlsx 파싱 추가
- 헤더 별칭에 underscore형(order_sn, tracking_number) 추가 - Shopee 는 SKU/수량이 product_info 한 셀에 묶여 있어 전용 파서 추가 · store.parse_shopee_product_info: [n] 단위 상품 분해, SKU/상품명/수량 추출 · parser._parse_shopee: 1행=1주문=1박스, 상품 펼쳐 박스로 묶음, 배송사 SPX 추론 - parse_export(platform) 로 Shopee/일반 경로 분기, 라우터가 platform 전달 - 이름/주소/전화는 xlsx 에 없음 → 라벨 PDF 단계에서 보강 예정 - 테스트 3개 추가 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -27,13 +27,23 @@ class ParseError(Exception):
|
||||
# 헤더가 첫 행이 아닐 수 있어 상단에서 헤더 행을 찾을 때 훑는 최대 행 수.
|
||||
_HEADER_SCAN_ROWS = 15
|
||||
|
||||
# Shopee SPX 송장번호 접두사 → 배송사. xlsx 에 배송사 컬럼이 없어 송장으로 추론.
|
||||
_SHOPEE_DEFAULT_PROVIDER = "SPX"
|
||||
|
||||
def parse_export(path: str) -> dict[str, Any]:
|
||||
|
||||
def parse_export(path: str, platform: str = "") -> dict[str, Any]:
|
||||
"""출고 XLSX 경로 → {"parcels": [...], "row_count": int}.
|
||||
|
||||
parcels 는 store.group_parcels 결과(1박스=1항목). 실패 시 ParseError.
|
||||
TikTok·Shopee 공용(표준키 매핑이 흡수한다).
|
||||
Shopee 는 SKU/수량이 product_info 한 셀에 묶여 있어 별도 경로로 파싱한다.
|
||||
"""
|
||||
if (platform or "").strip() == store.PLATFORM_SHOPEE:
|
||||
return _parse_shopee(path)
|
||||
return _parse_generic(path)
|
||||
|
||||
|
||||
def _load_rows(path: str) -> list:
|
||||
"""첫 시트 전체 행(values_only). 열기/빈 파일 예외는 ParseError 로."""
|
||||
try:
|
||||
from openpyxl import load_workbook # noqa: WPS433 — 지연 import
|
||||
except ImportError as exc: # pragma: no cover
|
||||
@@ -46,13 +56,18 @@ def parse_export(path: str) -> dict[str, Any]:
|
||||
wb = load_workbook(path, data_only=True)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise ParseError(f"엑셀 파일을 열 수 없습니다: {type(exc).__name__}") from exc
|
||||
|
||||
try:
|
||||
ws = wb.worksheets[0]
|
||||
all_rows = list(ws.iter_rows(values_only=True))
|
||||
if not all_rows:
|
||||
rows = list(wb.worksheets[0].iter_rows(values_only=True))
|
||||
finally:
|
||||
wb.close()
|
||||
if not rows:
|
||||
raise ParseError("엑셀에 데이터가 없습니다(빈 파일).")
|
||||
return rows
|
||||
|
||||
|
||||
def _parse_generic(path: str) -> dict[str, Any]:
|
||||
"""TikTok 등 표준 컬럼형 엑셀: 헤더 행 자동탐지 → 표준키 매핑 → 박스 묶기."""
|
||||
all_rows = _load_rows(path)
|
||||
header_idx, mapping = _find_header(all_rows)
|
||||
if header_idx is None:
|
||||
missing = store.missing_required({})
|
||||
@@ -76,8 +91,57 @@ def parse_export(path: str) -> dict[str, Any]:
|
||||
if _is_description_row(record):
|
||||
continue
|
||||
norm_rows.append(record)
|
||||
finally:
|
||||
wb.close()
|
||||
|
||||
parcels = store.group_parcels(norm_rows)
|
||||
return {"parcels": parcels, "row_count": len(norm_rows)}
|
||||
|
||||
|
||||
def _parse_shopee(path: str) -> dict[str, Any]:
|
||||
"""Shopee Packing List.Doorstep Delivery.xlsx 파싱.
|
||||
|
||||
컬럼: order_sn, tracking_number, product_info(상품 N개가 한 셀에 묶임).
|
||||
1행 = 1주문 = 1박스. product_info 를 펼쳐 상품별 행을 만든 뒤 박스로 묶는다.
|
||||
이름/주소/전화는 xlsx 에 없으므로 라벨 PDF 단계에서 채운다(여기선 빈 값).
|
||||
"""
|
||||
all_rows = _load_rows(path)
|
||||
|
||||
header_idx = order_idx = track_idx = pinfo_idx = None
|
||||
for idx, raw in enumerate(all_rows[:_HEADER_SCAN_ROWS]):
|
||||
if raw is None:
|
||||
continue
|
||||
oi = store.find_column(raw, store.COLUMN_ALIASES["order_id"])
|
||||
ti = store.find_column(raw, store.COLUMN_ALIASES["tracking_id"])
|
||||
pi = store.find_column(raw, store.PRODUCT_INFO_ALIASES)
|
||||
if pi is not None and (oi is not None or ti is not None):
|
||||
header_idx, order_idx, track_idx, pinfo_idx = idx, oi, ti, pi
|
||||
break
|
||||
|
||||
if header_idx is None:
|
||||
raise ParseError(
|
||||
"Shopee 엑셀에서 필요한 컬럼을 찾지 못했습니다: "
|
||||
"product_info + (order_sn 또는 tracking_number)"
|
||||
)
|
||||
|
||||
norm_rows: list[dict[str, str]] = []
|
||||
for raw in all_rows[header_idx + 1:]:
|
||||
if raw is None:
|
||||
continue
|
||||
order = store.clean_text(raw[order_idx]) if order_idx is not None and order_idx < len(raw) else ""
|
||||
tracking = store.clean_text(raw[track_idx]) if track_idx is not None and track_idx < len(raw) else ""
|
||||
blob = raw[pinfo_idx] if pinfo_idx < len(raw) else ""
|
||||
if not (order or tracking):
|
||||
continue
|
||||
for it in store.parse_shopee_product_info(blob):
|
||||
norm_rows.append(
|
||||
{
|
||||
"order_id": order,
|
||||
"tracking_id": tracking,
|
||||
"shipping_provider": _SHOPEE_DEFAULT_PROVIDER,
|
||||
"seller_sku": it["seller_sku"],
|
||||
"product_name": it["product_name"],
|
||||
"quantity": it["quantity"],
|
||||
}
|
||||
)
|
||||
|
||||
parcels = store.group_parcels(norm_rows)
|
||||
return {"parcels": parcels, "row_count": len(norm_rows)}
|
||||
|
||||
@@ -240,7 +240,7 @@ async def batch_create(
|
||||
|
||||
# 엑셀 파싱 — 실패 시 사용자에게 한국어 메시지 그대로 보여준다.
|
||||
try:
|
||||
parsed = parse_export(order_path)
|
||||
parsed = parse_export(order_path, platform=platform)
|
||||
except ParseError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc))
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
@@ -50,9 +51,9 @@ PLATFORMS: tuple[str, ...] = (PLATFORM_TIKTOK, PLATFORM_SHOPEE)
|
||||
# 해당 표준키 줄에 별칭 1개만 추가하면 된다.
|
||||
# 받는 사람(recipient_*) 정보는 작업 카드 표시 + 출고 엑셀 생성을 위해 보관한다.
|
||||
COLUMN_ALIASES: dict[str, tuple[str, ...]] = {
|
||||
"order_id": ("order id", "order_id", "order sn", "ordersn", "order no.", "order number", "주문번호"),
|
||||
"order_id": ("order id", "order_id", "order sn", "order_sn", "ordersn", "order no.", "order number", "주문번호"),
|
||||
"package_id": ("package id", "package_id", "package number", "패키지번호"),
|
||||
"tracking_id": ("tracking id", "tracking_id", "tracking number", "tracking no", "tracking no.", "awb", "awb no.", "송장번호"),
|
||||
"tracking_id": ("tracking id", "tracking_id", "tracking number", "tracking_number", "tracking no", "tracking no.", "awb", "awb no.", "송장번호"),
|
||||
"shipping_provider": ("shipping provider name", "shipping provider", "shipping_provider_name",
|
||||
"shipping option", "courier", "carrier", "logistics", "배송사", "택배사"),
|
||||
"seller_sku": ("seller sku", "seller_sku", "sku", "sku reference no.", "sku reference no",
|
||||
@@ -83,6 +84,66 @@ COLUMN_DISPLAY: dict[str, str] = {
|
||||
# 박스 단위로 보존하는 받는 사람 필드(같은 박스의 첫 비어있지 않은 값 사용).
|
||||
RECIPIENT_FIELDS: tuple[str, ...] = ("recipient_name", "recipient_phone", "recipient_address")
|
||||
|
||||
# Shopee Packing List 의 상품 정보 컬럼(헤더 정규화 후 비교).
|
||||
PRODUCT_INFO_ALIASES: tuple[str, ...] = ("product_info", "product info", "상품정보")
|
||||
|
||||
# Shopee product_info 한 셀에 여러 상품이 [1]…[2]… 로 들어온다. 각 상품은
|
||||
# "Key:Value" 들이 ';' 로 구분된다(예: Product Name:…; Variation Name:…; Price:…;
|
||||
# Quantity:1; SKU Reference No.: MT-0320_3;).
|
||||
_SHOPEE_ITEM_SPLIT = re.compile(r"\[\d+\]\s*")
|
||||
_SHOPEE_KEY_MAP: dict[str, str] = {
|
||||
"product name": "product_name",
|
||||
"variation name": "variation",
|
||||
"quantity": "quantity",
|
||||
"sku reference no": "seller_sku", # 끝의 '.' 은 비교 전에 제거
|
||||
}
|
||||
|
||||
|
||||
def parse_shopee_product_info(blob: Any) -> list[dict[str, Any]]:
|
||||
"""Shopee product_info 셀 → 상품 리스트.
|
||||
|
||||
반환 각 항목: {seller_sku, product_name, variation, quantity}.
|
||||
상품명 안에 ';' 이 들어가면 필드 분리가 깨질 수 있으나 Shopee 상품명에는
|
||||
드물어 허용한다(필요 시 별도 처리).
|
||||
"""
|
||||
s = clean_text(blob)
|
||||
if not s:
|
||||
return []
|
||||
items: list[dict[str, Any]] = []
|
||||
for chunk in _SHOPEE_ITEM_SPLIT.split(s):
|
||||
chunk = chunk.strip()
|
||||
if not chunk:
|
||||
continue
|
||||
fields: dict[str, str] = {}
|
||||
for part in chunk.split(";"):
|
||||
if ":" not in part:
|
||||
continue
|
||||
raw_key, _, value = part.partition(":")
|
||||
key = raw_key.strip().lower().rstrip(".").strip()
|
||||
std = _SHOPEE_KEY_MAP.get(key)
|
||||
if std:
|
||||
fields[std] = value.strip()
|
||||
sku = fields.get("seller_sku", "")
|
||||
if not sku and not fields.get("product_name"):
|
||||
continue # 식별 불가한 빈 항목 스킵
|
||||
items.append(
|
||||
{
|
||||
"seller_sku": sku,
|
||||
"product_name": fields.get("product_name", ""),
|
||||
"variation": fields.get("variation", ""),
|
||||
"quantity": normalize_quantity(fields.get("quantity")),
|
||||
}
|
||||
)
|
||||
return items
|
||||
|
||||
|
||||
def find_column(headers: Iterable[Any], aliases: tuple[str, ...]) -> int | None:
|
||||
"""헤더에서 별칭에 맞는 첫 열 인덱스. 없으면 None."""
|
||||
for idx, h in enumerate(headers):
|
||||
if normalize_header(h) in aliases:
|
||||
return idx
|
||||
return None
|
||||
|
||||
# 박스 묶음 기준 우선순위. 앞에서부터 비어있지 않은 첫 값으로 묶는다.
|
||||
BOX_KEY_PRIORITY: tuple[str, ...] = ("package_id", "tracking_id", "order_id")
|
||||
|
||||
|
||||
@@ -81,6 +81,34 @@ def test_column_alias_strip_and_case():
|
||||
assert "seller_sku" in mapping
|
||||
|
||||
|
||||
def test_shopee_product_info_single():
|
||||
blob = ("[1] Product Name:[1+1=3] Mira's Kitchen Korean Grade Miracle Food "
|
||||
"Container Launch PROMO; Variation Name:3x 320ml; Price: RM 20.00; "
|
||||
"Quantity: 1; SKU Reference No.: MT-0320_3;")
|
||||
items = store.parse_shopee_product_info(blob)
|
||||
assert len(items) == 1, items
|
||||
it = items[0]
|
||||
assert it["seller_sku"] == "MT-0320_3", it
|
||||
assert it["quantity"] == 1, it
|
||||
assert it["variation"] == "3x 320ml", it
|
||||
assert "Mira's Kitchen" in it["product_name"], it
|
||||
|
||||
|
||||
def test_shopee_product_info_multi():
|
||||
blob = ("[1] Product Name:Item A; Variation Name:S; Quantity: 2; "
|
||||
"SKU Reference No.: MT-0001; "
|
||||
"[2] Product Name:Item B; Variation Name:L; Quantity: 3; "
|
||||
"SKU Reference No.: MX-0002;")
|
||||
items = store.parse_shopee_product_info(blob)
|
||||
skus = {i["seller_sku"]: i["quantity"] for i in items}
|
||||
assert skus == {"MT-0001": 2, "MX-0002": 3}, skus
|
||||
|
||||
|
||||
def test_shopee_product_info_empty():
|
||||
assert store.parse_shopee_product_info("") == []
|
||||
assert store.parse_shopee_product_info(None) == []
|
||||
|
||||
|
||||
def _run_all():
|
||||
fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||||
for fn in fns:
|
||||
|
||||
Reference in New Issue
Block a user