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

"""출고 배치 → 취합 엑셀(openpyxl) 생성.

다운로드 zip 에 업로드 원본과 함께 넣는 '출고 엑셀'을 만든다. 1박스 안에 여러
SKU 가 있으면 SKU 당 1행으로 펼치고 박스 단위(받는 사람/주문/택배) 정보는 반복한다.

파일명 형식: 2026.06.22(Mon)_tictoc.xlsx  (Shopee 는 끝이 _shopee).
  - TikTok 은 운영 요청 철자에 맞춰 'tictoc' 을 쓴다.
"""

from __future__ import annotations

import io
import re
from datetime import date, datetime
from typing import Any

# 엑셀 컬럼(요청 순서). (헤더, 박스/배치/상품 키)
_COLUMNS: tuple[tuple[str, str], ...] = (
    ("발송 날짜", "dispatch_date"),
    ("고객 이름", "recipient_name"),
    ("전화번호", "recipient_phone"),
    ("주소", "recipient_address"),
    ("상품이름", "product_name"),
    ("아이템 코드", "seller_sku"),
    ("주문 수량", "quantity"),
    ("오더번호", "order_id"),
    ("택배사", "shipping_provider"),
    ("택배송장번호", "tracking_id"),
    ("주문처", "platform"),
)

# 플랫폼 → 파일명 접미사.
_PLATFORM_SUFFIX: dict[str, str] = {
    "TikTok": "tictoc",
    "Shopee": "shopee",
}


def export_filename(*, dispatch_date: str, platform: str) -> str:
    """2026.06.22(Mon)_tictoc.xlsx 형식 파일명. 날짜 파싱 실패 시 원문 사용."""
    d = _parse_date(dispatch_date)
    if d is not None:
        stamp = d.strftime("%Y.%m.%d(%a)")
    else:
        stamp = (dispatch_date or "").strip() or "dispatch"
    suffix = _PLATFORM_SUFFIX.get((platform or "").strip(), (platform or "dispatch").strip().lower())
    return f"{stamp}_{suffix}.xlsx"


def split_sku(seller_sku: str) -> tuple[str, int]:
    """seller_sku → (아이템코드, 배수).

    끝의 `_<숫자>` 는 '같은 상품 N개'를 뜻한다(예: MT-0320_3 → 코드 MT-0320, 3개).
    접미사가 없으면 배수 1.
    """
    s = (seller_sku or "").strip()
    m = re.search(r"_(\d+)$", s)
    if m:
        return s[: m.start()], int(m.group(1))
    return s, 1


def resolve_product_name(seller_sku: str, name_map: dict[str, str], fallback: str = "") -> str:
    """아이템코드(seller_sku)로 itemcode_db 상품명을 찾는다.

    seller_sku 는 변형 접미사가 붙을 수 있다(예: MT-0320_3 → DB 는 MT-0320).
    정확 일치 → 접미사(_숫자) 제거 후 일치 → 없으면 fallback(파싱된 상품명).
    """
    key = (seller_sku or "").strip().upper()
    if not key:
        return fallback
    if key in name_map:
        return name_map[key]
    base = re.sub(r"_\d+$", "", key)
    if base != key and base in name_map:
        return name_map[base]
    return fallback


def build_workbook_bytes(
    *,
    batch: dict[str, Any],
    parcels: list[dict[str, Any]],
    name_map: dict[str, str] | None = None,
) -> bytes:
    """배치 + 박스(상품 포함) → xlsx 바이트.

    parcels 각 항목은 db.list_parcels 결과(items 포함, recipient_* 포함).
    name_map: itemcode_db 의 {코드(대문자): 상품명}. 있으면 아이템코드로 상품명을
    조회해 채운다(없으면 파싱된 상품명 사용).
    """
    name_map = name_map or {}
    from openpyxl import Workbook  # noqa: WPS433 — 지연 import
    from openpyxl.styles import Font

    wb = Workbook()
    ws = wb.active
    ws.title = "출고"

    headers = [h for h, _ in _COLUMNS]
    ws.append(headers)
    for cell in ws[1]:
        cell.font = Font(bold=True)

    dispatch_date = batch.get("dispatch_date", "") or ""
    platform = batch.get("platform", "") or ""

    for p in parcels:
        box = {
            "dispatch_date": dispatch_date,
            "platform": platform,
            "recipient_name": p.get("recipient_name", "") or "",
            "recipient_phone": p.get("recipient_phone", "") or "",
            "recipient_address": p.get("recipient_address", "") or "",
            "order_id": p.get("order_id", "") or "",
            "shipping_provider": p.get("shipping_provider", "") or "",
            "tracking_id": p.get("tracking_id", "") or "",
        }
        items = p.get("items") or [{}]
        for it in items:
            row_data = dict(box)
            sku = it.get("seller_sku", "") or ""
            base_code, mult = split_sku(sku)
            row_data["product_name"] = resolve_product_name(
                base_code, name_map, fallback=it.get("product_name", "") or ""
            )
            row_data["seller_sku"] = base_code  # 접미사(_N) 제거한 실제 아이템코드
            qty = it.get("quantity")
            row_data["quantity"] = (int(qty) * mult) if qty is not None else ""
            ws.append([_cell(row_data.get(key, "")) for _, key in _COLUMNS])

    _autosize(ws, headers)

    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


# ── 사방넷 재고 업로드 엑셀(낱개 분해) ──
_STOCK_HEADERS: tuple[str, ...] = ("상품코드[필수]", "가용수량", "불용수량", "바코드")
_LID_PREFIX = "MD-"  # 뚜껑 — 재고 집계 제외
_SET_PREFIX = "MY-"  # 콤보(세트)


def split_singles_combos(
    rows: list[dict[str, Any]],
) -> tuple[dict[str, int], dict[str, int]]:
    """SKU별 합산 행 → (낱개 {code: qty}, 콤보 {code: qty}).

    _N 배수 반영, MD-(뚜껑) 제외. 콤보는 MY- 접두사. 분해는 하지 않는다
    (말레이시아 재고관리가 세트 OUT 시 BOM 으로 분해하므로 콤보 그대로 넘긴다).
    """
    singles: dict[str, int] = {}
    combos: dict[str, int] = {}
    for r in rows:
        base, mult = split_sku(r.get("seller_sku", ""))
        base = base.upper()
        if not base:
            continue
        pieces = int(r.get("qty", 0)) * mult
        if pieces <= 0 or base.startswith(_LID_PREFIX):
            continue
        if base.startswith(_SET_PREFIX):
            combos[base] = combos.get(base, 0) + pieces
        else:
            singles[base] = singles.get(base, 0) + pieces
    return singles, combos


def explode_to_singles(
    rows: list[dict[str, Any]],
    set_bom: dict[str, list[dict[str, Any]]],
) -> dict[str, int]:
    """SKU별 합산 행 → 낱개 코드별 수량.

    - seller_sku 의 _N 접미사는 배수(같은 상품 N개)로 곱한다.
    - 콤보(세트, set_bom 에 존재)는 구성 낱개로 분해해 (배수 × 구성수량) 더한다.
    - MD-(뚜껑)은 재고 집계에서 제외한다.
    반환: {낱개코드(대문자): 수량}.
    """
    singles: dict[str, int] = {}
    for r in rows:
        base, mult = split_sku(r.get("seller_sku", ""))
        base = base.upper()
        if not base:
            continue
        pieces = int(r.get("qty", 0)) * mult
        if pieces <= 0 or base.startswith(_LID_PREFIX):
            continue
        comps = set_bom.get(base)
        if comps:  # 콤보 → 낱개 분해
            for c in comps:
                cc = str(c.get("component_code", "")).strip().upper()
                if not cc or cc.startswith(_LID_PREFIX):
                    continue
                singles[cc] = singles.get(cc, 0) + pieces * int(c.get("component_qty", 0) or 0)
        else:
            singles[base] = singles.get(base, 0) + pieces
    return {k: v for k, v in singles.items() if v > 0}


def build_stock_xlsx(
    *,
    rows: list[dict[str, Any]],
    set_bom: dict[str, list[dict[str, Any]]],
    sabangnet_map: dict[str, str],
) -> bytes:
    """사방넷 재고 업로드용 xlsx(낱개 분해). 4열: 상품코드[필수]/가용수량/불용수량/바코드.

    A=사방넷코드, B=수량, C=0(불용), D=아이템코드(바코드). 코드 오름차순.
    """
    from openpyxl import Workbook  # noqa: WPS433
    from openpyxl.styles import Alignment, Font, PatternFill

    singles = explode_to_singles(rows, set_bom)

    wb = Workbook()
    ws = wb.active
    ws.title = "재고"
    ws.append(list(_STOCK_HEADERS))
    head_fill = PatternFill("solid", fgColor="C55A11")  # 사방넷 템플릿 주황
    head_font = Font(bold=True, color="FFFFFF")
    for cell in ws[1]:
        cell.fill = head_fill
        cell.font = head_font
        cell.alignment = Alignment(horizontal="center", vertical="center")

    for code in sorted(singles):
        sabangnet = sabangnet_map.get(code, "")
        ws.append([sabangnet, singles[code], 0, code])

    widths = [22, 12, 12, 18]
    from openpyxl.utils import get_column_letter
    for i, w in enumerate(widths, start=1):
        ws.column_dimensions[get_column_letter(i)].width = w

    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


def stock_filename(dispatch_date: str) -> str:
    """사방넷 출고 엑셀 파일명: 2026.06.22(Mon)_sabangnet_outbound.xlsx."""
    d = _parse_date(dispatch_date)
    stamp = d.strftime("%Y.%m.%d(%a)") if d else (dispatch_date or "outbound").strip()
    return f"{stamp}_sabangnet_outbound.xlsx"


def _cell(value: Any) -> Any:
    """송장/전화번호 등이 숫자로 보이면 엑셀이 과학표기/반올림 하므로 문자열 보존."""
    if isinstance(value, int):
        return value
    return "" if value is None else str(value)


def _autosize(ws: Any, headers: list[str]) -> None:
    """대략적인 컬럼 폭. 주소는 길어 상한을 둔다."""
    from openpyxl.utils import get_column_letter

    widths: list[int] = [len(h) for h in headers]
    for row in ws.iter_rows(min_row=2, values_only=True):
        for i, val in enumerate(row):
            ln = len(str(val)) if val is not None else 0
            if ln > widths[i]:
                widths[i] = ln
    for i, w in enumerate(widths, start=1):
        ws.column_dimensions[get_column_letter(i)].width = min(max(w + 2, 8), 50)


def _parse_date(value: str) -> date | None:
    s = (value or "").strip()
    if not s:
        return None
    for fmt in ("%Y-%m-%d", "%Y.%m.%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(s, fmt).date()
        except ValueError:
            continue
    return None