dbx-main/app/modules/cupang/sales_import.py

"""쿠팡 로켓 매출 시트 파서 (xlsx / csv).

구글 시트 "쿠팡 로켓 매출" 양식을 그대로 읽는다.

    1~3행  병합 머리글
    4행~   발주 라인 1건 = 1행
           · 주차 소계 행: 광고비(CPC)/할인 프로모션/장려금이 여기에만 있다
           · 월계/총계 행: 저장하지 않는다(화면에서 합산)

열 순서(0-based)
    0 구분  1 순번  2 발주번호  3 발주유형  4 발주일  5 출고일  6 센터입고일
    7 SKUID  8 바코드  9 품목  10 수량  11 입고센터
    12 공급단가  13 공급가  14 원가(단가)  15 원가(발주량)
    16 피킹비 단가  17 피킹비 합계  18 밀크런/쉽먼트  19 물류비 합계  20 물류비중(%)
    21 마진  22 마진율  23 광고비(CPC)  24 할인 프로모션  25 장려금
    26 순마진  27 순마진율  28 사방넷 재고차감일
"""

from __future__ import annotations

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

COL = {
    "week": 0, "seq": 1, "po_no": 2, "po_type": 3, "order_date": 4,
    "ship_date": 5, "center_arrival_date": 6, "sku_id": 7, "barcode": 8,
    "item_name": 9, "quantity": 10, "center_name": 11,
    "supply_unit_price": 12, "supply_amount": 13,
    "cost_unit_price": 14, "cost_amount": 15,
    "picking_unit_price": 16, "picking_amount": 17, "milkrun_amount": 18,
    "logistics_total": 19, "logistics_ratio": 20,
    "margin": 21, "margin_rate": 22,
    "ad_cost": 23, "promo_discount": 24, "incentive": 25,
    "net_margin": 26, "net_margin_rate": 27, "stock_deduct_memo": 28,
}

BAD = {"", "-", "—", "#DIV/0!", "#N/A", "#VALUE!", "#REF!"}
HEADER_ROWS = 3


def _cell(row: list[Any], key: str) -> str:
    idx = COL[key]
    if idx >= len(row):
        return ""
    value = row[idx]
    if value is None:
        return ""
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()
    return str(value).strip()


def _flat(text: str) -> str:
    """여러 줄 라벨을 한 줄로 — "\\n8월2주차\\n(8/11~8/17)" → "8월2주차(8/11~8/17)"."""
    return re.sub(r"\s+", "", (text or "").strip())


def parse_rows(rows: list[list[Any]]) -> dict[str, Any]:
    """표 전체 → {lines, weeklies, skipped}. 값 정규화는 DB 계층이 한 번 더 한다."""
    lines: list[dict[str, Any]] = []
    weeklies: list[dict[str, Any]] = []
    week_label = ""
    skipped = 0

    for row in rows[HEADER_ROWS:]:
        if not row or not any(str(c or "").strip() for c in row):
            continue
        week = _flat(_cell(row, "week"))
        seq = _cell(row, "seq")
        po_no = _cell(row, "po_no")

        # 데이터 행 판정 — 순번이 비어 있는 행도 시트에 있으므로 발주번호+품목으로 본다.
        sku = _cell(row, "sku_id")
        item = _cell(row, "item_name")
        po_is_num = po_no.replace(".0", "").replace("-", "").isdigit()
        if po_no not in BAD and po_is_num and (sku not in BAD or item not in BAD):
            if week:
                week_label = week
            line = {
                "week_label": week_label,
                "seq": seq.replace(".0", "") if seq.replace(".0", "").isdigit() else "",
            }
            for field in (
                "po_no", "po_type", "order_date", "ship_date", "center_arrival_date",
                "sku_id", "barcode", "item_name", "quantity", "center_name",
                "supply_unit_price", "supply_amount", "cost_unit_price", "cost_amount",
                "picking_unit_price", "picking_amount", "milkrun_amount",
                "logistics_total", "logistics_ratio", "margin", "margin_rate",
            ):
                line[field] = _cell(row, field)
            line["stock_deduct_memo"] = re.sub(
                r"\s+", " ", _cell(row, "stock_deduct_memo")
            ).strip()
            # SKU/바코드가 숫자로 읽히면 소수점이 붙는다 → 정수 표기로
            for key in ("sku_id", "barcode", "po_no"):
                if line[key].endswith(".0"):
                    line[key] = line[key][:-2]
            lines.append(line)
            continue

        # 주차 소계 — 광고비/할인/장려금만 가져온다
        if week and "주차" in week:
            weeklies.append({
                "week_label": week_label,
                "ad_cost": _cell(row, "ad_cost"),
                "promo_discount": _cell(row, "promo_discount"),
                "incentive": _cell(row, "incentive"),
            })
            continue
        skipped += 1

    # 주차 기간은 그 주차 라인들의 출고일 최소/최대로 채운다
    span: dict[str, tuple[str, str]] = {}
    for line in lines:
        d = str(line.get("ship_date") or "")
        if not d or d in BAD:
            continue
        lo, hi = span.get(line["week_label"], (d, d))
        span[line["week_label"]] = (min(lo, d), max(hi, d))

    seen: set[str] = set()
    weekly_rows: list[dict[str, Any]] = []
    for wk in weeklies:
        label = wk["week_label"]
        if not label or label in seen:
            continue
        seen.add(label)
        lo, hi = span.get(label, ("", ""))
        wk["week_from"], wk["week_to"] = lo, hi
        weekly_rows.append(wk)

    return {"lines": lines, "weeklies": weekly_rows, "skipped": skipped}


def parse_xlsx(data: bytes) -> dict[str, Any]:
    from openpyxl import load_workbook  # noqa: WPS433

    wb = load_workbook(io.BytesIO(data), data_only=True, read_only=True)
    ws = wb[wb.sheetnames[0]]
    rows = [list(r) for r in ws.iter_rows(values_only=True)]
    wb.close()
    return parse_rows(rows)


def parse_csv(data: bytes) -> dict[str, Any]:
    for encoding in ("utf-8-sig", "utf-8", "cp949"):
        try:
            text = data.decode(encoding)
            break
        except UnicodeDecodeError:
            continue
    else:
        raise ValueError("CSV 인코딩을 알 수 없습니다(UTF-8 또는 CP949).")
    return parse_rows([row for row in csv.reader(io.StringIO(text))])


def parse_upload(filename: str, data: bytes) -> dict[str, Any]:
    name = (filename or "").lower()
    if name.endswith(".csv"):
        return parse_csv(data)
    if name.endswith((".xlsx", ".xlsm")):
        return parse_xlsx(data)
    raise ValueError("xlsx 또는 csv 파일만 올릴 수 있습니다.")