dbx-main/app/modules/dispatch/parser.py

"""출고 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


def parse_export(path: str) -> dict[str, Any]:
    """출고 XLSX 경로 → {"parcels": [...], "row_count": int}.

    parcels 는 store.group_parcels 결과(1박스=1항목). 실패 시 ParseError.
    TikTok·Shopee 공용(표준키 매핑이 흡수한다).
    """
    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 는 워크시트 <dimension> 메타가
    # 부실해 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:
        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)
    finally:
        wb.close()

    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