"""출고 XLSX 파서 (openpyxl) — TikTok / Shopee 공용. pandas 대신 레포에 이미 있는 openpyxl 로 읽어 의존성을 가볍게 유지한다 (동작: 헤더 정규화 → 표준키 매핑 → 박스 묶기). - TikTok: 03_TikTok_Order_Export.xlsx, Shopee: Packing List.Doorstep Delivery.xlsx. 두 포맷 모두 같은 표준키로 매핑된다(store.COLUMN_ALIASES). 헤더가 첫 행이 아닐 수 있어(Shopee 는 제목 행이 위에 붙는 경우가 있다) 상단 몇 행을 훑어 필수 컬럼이 잡히는 첫 행을 헤더로 본다. - 받는 사람(이름/전화/주소) 컬럼은 매핑되면 보존한다(작업 카드 표시 + 출고 엑셀). - Tracking ID 는 숫자로 읽혀도 문자열로 보존(store.clean_text). - Quantity 가 비면 1, NaN/None 은 빈 문자열로 처리. - 필수 컬럼이 없으면 어떤 컬럼이 없는지 ParseError 로 알린다. """ from __future__ import annotations from typing import Any from . import store class ParseError(Exception): """엑셀 파싱 실패. message 는 사용자에게 그대로 보여줄 한국어 메시지.""" # 헤더가 첫 행이 아닐 수 있어 상단에서 헤더 행을 찾을 때 훑는 최대 행 수. _HEADER_SCAN_ROWS = 15 # Shopee SPX 송장번호 접두사 → 배송사. xlsx 에 배송사 컬럼이 없어 송장으로 추론. _SHOPEE_DEFAULT_PROVIDER = "SPX" def parse_export(path: str, platform: str = "") -> dict[str, Any]: """출고 XLSX 경로 → {"parcels": [...], "row_count": int}. parcels 는 store.group_parcels 결과(1박스=1항목). 실패 시 ParseError. - Shopee: SKU/수량이 product_info 한 셀에 묶여 있어 별도 경로. - Manual: 받는 사람/상품/수량이 첫 시트 한 행에 있는 수동 오더(1행=1박스). """ p = (platform or "").strip() if p == store.PLATFORM_SHOPEE: return _parse_shopee(path) if p == store.PLATFORM_MANUAL: return _parse_manual(path) return _parse_generic(path) # 메뉴얼 오더 엑셀 컬럼 별칭(헤더 정규화 후 비교). 받는 사람 정보가 엑셀에 직접 있다. _MANUAL_ALIASES: dict[str, tuple[str, ...]] = { "recipient_name": ("receiver name", "recipient name", "recipient", "받는 사람", "받는사람", "수취인", "이름", "name"), "recipient_phone": ("phone", "phone number", "phone #", "contact", "전화", "전화번호", "연락처"), "recipient_address": ("address", "delivery address", "주소", "배송지", "배송주소"), "product_name": ("product name", "item name", "product", "상품명", "상품이름"), "seller_sku": ("item code", "seller sku", "sku", "아이템코드", "아이템 코드", "상품코드", "코드"), "quantity": ("quantity", "qty", "수량", "주문수량"), } def _parse_manual(path: str) -> dict[str, Any]: """메뉴얼 오더 엑셀(첫 시트). 1행 = 1박스. 컬럼: Receiver Name / Phone / Address / Product Name / Item Code / Quantity. 받는 사람 정보가 엑셀에 직접 있어 PDF 없이 카드/엑셀에 표시된다. 묶음키 (Order/Tracking/Package)가 없으므로 행마다 별도 박스(seq)로 만든다. """ all_rows = _load_rows(path) header_idx = None mapping: dict[str, int] = {} for idx, raw in enumerate(all_rows[:_HEADER_SCAN_ROWS]): if raw is None: continue m = { key: store.find_column(raw, aliases) for key, aliases in _MANUAL_ALIASES.items() } # 최소: 아이템코드 컬럼이 잡혀야 한다. if m.get("seller_sku") is not None: header_idx = idx mapping = {k: v for k, v in m.items() if v is not None} break if header_idx is None: raise ParseError( "메뉴얼 오더 엑셀에서 컬럼을 찾지 못했습니다. " "필요 컬럼: Item Code(필수), Receiver Name, Phone, Address, Product Name, Quantity" ) def cell(raw: Any, key: str) -> str: idx = mapping.get(key) if idx is None or idx >= len(raw): return "" return store.clean_text(raw[idx]) parcels: list[dict[str, Any]] = [] seq = 0 for raw in all_rows[header_idx + 1:]: if raw is None: continue sku = cell(raw, "seller_sku") name = cell(raw, "recipient_name") if not sku and not name: continue # 빈 행 if not sku: continue # 아이템코드 없으면 출고 대상 아님 seq += 1 parcels.append( { "seq": seq, "order_id": "", "package_id": "", "tracking_id": "", "shipping_provider": "", "recipient_name": name, "recipient_phone": cell(raw, "recipient_phone"), "recipient_address": cell(raw, "recipient_address"), "items": [ { "seller_sku": sku, "product_name": cell(raw, "product_name"), "quantity": store.normalize_quantity(cell(raw, "quantity")), } ], } ) return {"parcels": parcels, "row_count": len(parcels)} 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 raise ParseError("openpyxl 이 설치되어 있지 않습니다.") from exc # read_only=True 는 쓰지 않는다: TikTok export 는 워크시트 메타가 # 부실해 read_only 모드가 A열만 읽고 끊기는 openpyxl 버그가 있다(실제 54열인데 # 1열만 반환). 일일 출고량(수백~수천 행)이라 일반 모드로 충분하다. try: wb = load_workbook(path, data_only=True) except Exception as exc: # noqa: BLE001 raise ParseError(f"엑셀 파일을 열 수 없습니다: {type(exc).__name__}") from exc try: 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)} # 하위 호환 별칭(기존 호출부/테스트가 쓰던 이름). parse_order_export = parse_export def _find_header(rows: list) -> tuple[int | None, dict[str, int]]: """상단 행들을 훑어 필수 컬럼이 잡히는 첫 행을 헤더로 본다. 반환: (헤더 행 인덱스, {표준키: 열 인덱스}). 못 찾으면 (None, {}). """ for idx, raw in enumerate(rows[:_HEADER_SCAN_ROWS]): if raw is None: continue mapping = store.resolve_columns(raw) if not store.missing_required(mapping): return idx, mapping return None, {} def _is_description_row(record: dict[str, str]) -> bool: """헤더 아래 '컬럼 설명' 행 판별. Order/Tracking/Package ID 는 공백 없는 식별자다. 채워진 ID 값 중 하나라도 공백을 포함하면(예: 'Platform unique order ID.') 데이터가 아닌 설명 행이다. """ for key in ("order_id", "tracking_id", "package_id"): val = (record.get(key) or "").strip() if val and any(ch.isspace() for ch in val): return True return False