diff --git a/app/modules/dispatch/parser.py b/app/modules/dispatch/parser.py index 2cc40b7..7c7b16d 100644 --- a/app/modules/dispatch/parser.py +++ b/app/modules/dispatch/parser.py @@ -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,38 +56,92 @@ 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: - raise ParseError("엑셀에 데이터가 없습니다(빈 파일).") - - header_idx, mapping = _find_header(all_rows) - if header_idx is None: - missing = store.missing_required({}) - raise ParseError("엑셀에서 필요한 컬럼을 찾지 못했습니다: " + ", ".join(missing)) - - norm_rows: list[dict[str, str]] = [] - for raw in all_rows[header_idx + 1:]: - if raw is None: - continue - # 표준키만 추출(매핑 안 된 열은 버린다). - record = { - key: store.clean_text(raw[idx]) if idx < len(raw) else "" - for key, idx in mapping.items() - } - # 완전 빈 행 스킵 - if not any(record.get(k) for k in ("order_id", "package_id", "tracking_id", "seller_sku")): - continue - # 헤더 바로 아래 '컬럼 설명' 행 스킵(TikTok export 2번째 행). - # 주문/송장/패키지 ID 는 공백을 포함하지 않는다. 설명 문장은 공백을 - # 포함하므로, 채워진 ID 값에 공백이 있으면 데이터가 아니라 설명 행이다. - if _is_description_row(record): - continue - norm_rows.append(record) + 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({}) + raise ParseError("엑셀에서 필요한 컬럼을 찾지 못했습니다: " + ", ".join(missing)) + + norm_rows: list[dict[str, str]] = [] + for raw in all_rows[header_idx + 1:]: + if raw is None: + continue + # 표준키만 추출(매핑 안 된 열은 버린다). + record = { + key: store.clean_text(raw[idx]) if idx < len(raw) else "" + for key, idx in mapping.items() + } + # 완전 빈 행 스킵 + if not any(record.get(k) for k in ("order_id", "package_id", "tracking_id", "seller_sku")): + continue + # 헤더 바로 아래 '컬럼 설명' 행 스킵(TikTok export 2번째 행). + # 주문/송장/패키지 ID 는 공백을 포함하지 않는다. 설명 문장은 공백을 + # 포함하므로, 채워진 ID 값에 공백이 있으면 데이터가 아니라 설명 행이다. + if _is_description_row(record): + continue + norm_rows.append(record) + + 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)} diff --git a/app/modules/dispatch/router.py b/app/modules/dispatch/router.py index 1e97771..d68e697 100644 --- a/app/modules/dispatch/router.py +++ b/app/modules/dispatch/router.py @@ -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)) diff --git a/app/modules/dispatch/store.py b/app/modules/dispatch/store.py index f303315..374b208 100644 --- a/app/modules/dispatch/store.py +++ b/app/modules/dispatch/store.py @@ -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") diff --git a/app/modules/dispatch/tests/test_grouping.py b/app/modules/dispatch/tests/test_grouping.py index ec710e3..425f155 100644 --- a/app/modules/dispatch/tests/test_grouping.py +++ b/app/modules/dispatch/tests/test_grouping.py @@ -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: