dbx-main/app/modules/malaysia/store.py

"""말레이시아 창고 재고관리 모듈 — 상수 및 순수 계산/검증 헬퍼.

- 데이터 저장은 malaysia_stock_db(PostgreSQL) 전용이다(`db.py`).
  운영 데이터가 JSON 과 DB 로 갈라지는 것을 막기 위해 JSON 폴백을 두지 않는다.
  MALAYSIA_STOCK_DB_URL 미설정 시 라우터가 "설정 필요" 안내 페이지를 보여준다.
- 이 모듈에는 DB 의존이 없는 상수와 순수 함수만 둔다(테스트 용이).
  prefix 검증, 세트 BOM 분해, 재고조사 집계가 모두 여기 있다.

코드 prefix 규칙(요구사항 8 검증):
  - 낱개(개별) 아이템 : MT-, MX-, MZ-   → 입고/출고/조정 movement 가능
  - 세트 아이템        : MY-            → movement 불가, 재고조사/BOM 만 가능
  - 뚜껑(제외)         : MD-            → 어디에도 사용 불가
"""

from __future__ import annotations

from typing import Any, Iterable

# ── 이동(movement) 종류 ──
MOVEMENT_TYPES: tuple[str, ...] = ("IN", "OUT", "ADJUST", "STOCKTAKE")

# ── 재고조사 상태 ──
STOCKTAKE_STATUSES: tuple[str, ...] = ("draft", "finalized", "cancelled")

# ── 코드 prefix ──
INDIVIDUAL_PREFIXES: tuple[str, ...] = ("MT-", "MX-", "MZ-")
SET_PREFIX: str = "MY-"
BLOCKED_PREFIX: str = "MD-"  # 뚜껑 — 재고관리 대상에서 제외

# ── 창고 랙 레이아웃 (재고조사 랙 입력용) ──
# 그림 고정: A/B 두 면, 각 면 3행 × 3열, 각 열은 2분할 → 면당 18칸, 총 36칸.
# 행은 위→아래로 3,2,1 (그림과 동일). 칸 코드: '{면}{행}-{열}-{분할}' 예) A3-1-1.
RACK_SIDES: tuple[str, ...] = ("A", "B")
RACK_ROWS: tuple[int, ...] = (3, 2, 1)
RACK_COLS: tuple[int, ...] = (1, 2, 3)
RACK_SUBS: tuple[int, ...] = (1, 2)


def rack_cell_codes() -> list[str]:
    """모든 랙 칸 코드 목록(검증용). 예: 'A3-1-1' ... 'B1-3-2'."""
    return [
        f"{side}{row}-{col}-{sub}"
        for side in RACK_SIDES
        for row in RACK_ROWS
        for col in RACK_COLS
        for sub in RACK_SUBS
    ]


def rack_layout() -> list[dict[str, Any]]:
    """템플릿 렌더용 중첩 구조.

    [{"side":"A","rows":[{"row":3,"cols":[["A3-1-1","A3-1-2"], ...]}, ...]}, ...]
    """
    out: list[dict[str, Any]] = []
    for side in RACK_SIDES:
        rows: list[dict[str, Any]] = []
        for row in RACK_ROWS:
            cols = [
                [f"{side}{row}-{col}-{sub}" for sub in RACK_SUBS]
                for col in RACK_COLS
            ]
            rows.append({"row": row, "cols": cols})
        out.append({"side": side, "rows": rows})
    return out


def aggregate_rack(entries: Iterable[dict[str, Any]]) -> dict[str, int]:
    """랙 입력 항목을 SKU 코드별 총수량으로 집계.

    각 항목: {"sku_code", "units_per_box", "box_count"}.
    qty = SUM(units_per_box * box_count). 입수량/박스수 ≤ 0 또는 코드 공백은 스킵.
    반환: { sku_code: total_qty }  (낱개·세트 코드 모두 포함 가능)
    """
    totals: dict[str, int] = {}
    for e in entries:
        code = _norm(e.get("sku_code"))
        if not code:
            continue
        try:
            upb = int(e.get("units_per_box") or 0)
            box = int(e.get("box_count") or 0)
        except (TypeError, ValueError):
            continue
        if upb <= 0 or box <= 0:
            continue
        totals[code] = totals.get(code, 0) + upb * box
    return totals


# ════════════════════════════════════════════════════════════
# 코드 분류 / 검증 (순수 함수)
# ════════════════════════════════════════════════════════════
def _norm(code: str) -> str:
    return (code or "").strip().upper()


def is_individual_code(code: str) -> bool:
    """낱개 아이템(MT-/MX-/MZ-) 여부."""
    return _norm(code).startswith(INDIVIDUAL_PREFIXES)


def is_set_code(code: str) -> bool:
    """세트 아이템(MY-) 여부."""
    return _norm(code).startswith(SET_PREFIX)


def is_blocked_code(code: str) -> bool:
    """제외 대상(MD- 뚜껑) 여부."""
    return _norm(code).startswith(BLOCKED_PREFIX)


def classify_code(code: str) -> str:
    """코드 종류를 반환. 'individual' | 'set' | 'blocked' | 'unknown'."""
    if is_blocked_code(code):
        return "blocked"
    if is_individual_code(code):
        return "individual"
    if is_set_code(code):
        return "set"
    return "unknown"


def validate_movement_code(code: str) -> str:
    """입고/출고/조정 movement 에 쓸 수 있는 코드인지 검증.

    낱개(MT-/MX-/MZ-)만 허용. MY-/MD-/기타는 ValueError.
    반환: 정규화된 코드(대문자/trim).
    """
    c = _norm(code)
    if not c:
        raise ValueError("아이템 코드가 비어 있습니다.")
    kind = classify_code(c)
    if kind == "blocked":
        raise ValueError(f"뚜껑 코드(MD-)는 재고관리 대상이 아닙니다: {c}")
    if kind == "set":
        raise ValueError(f"세트 코드(MY-)는 입고/출고에 직접 쓸 수 없습니다: {c}")
    if kind != "individual":
        raise ValueError(f"허용되지 않는 코드입니다(MT-/MX-/MZ- 만 가능): {c}")
    return c


def validate_stocktake_code(code: str) -> str:
    """재고조사 라인(daily_stocktake_line)에 쓸 수 있는 코드인지 검증.

    낱개(MT-/MX-/MZ-) 또는 세트(MY-) 허용. MD-/기타는 ValueError.
    """
    c = _norm(code)
    if not c:
        raise ValueError("SKU 코드가 비어 있습니다.")
    kind = classify_code(c)
    if kind == "blocked":
        raise ValueError(f"뚜껑 코드(MD-)는 재고조사 대상이 아닙니다: {c}")
    if kind not in ("individual", "set"):
        raise ValueError(f"허용되지 않는 코드입니다(MT-/MX-/MZ-/MY- 만 가능): {c}")
    return c


def validate_bom_set_code(code: str) -> str:
    """set_bom 의 set_code 검증 — MY- 만 허용."""
    c = _norm(code)
    if not is_set_code(c):
        raise ValueError(f"세트 코드는 MY- 로 시작해야 합니다: {c}")
    return c


def validate_bom_component_code(code: str) -> str:
    """set_bom 의 component_code 검증 — MT-/MX-/MZ- 만 허용(MY-/MD- 금지)."""
    c = _norm(code)
    if is_blocked_code(c):
        raise ValueError(f"뚜껑 코드(MD-)는 세트 구성품이 될 수 없습니다: {c}")
    if is_set_code(c):
        raise ValueError(f"세트(MY-)가 세트의 구성품이 될 수 없습니다(중첩 불가): {c}")
    if not is_individual_code(c):
        raise ValueError(f"구성품은 MT-/MX-/MZ- 만 가능합니다: {c}")
    return c


def validate_qty(qty: Any, *, allow_zero: bool = True, allow_negative: bool = False) -> int:
    """수량 검증/정규화. 정수 변환 + 음수/0 정책 적용."""
    try:
        n = int(qty)
    except (TypeError, ValueError):
        raise ValueError(f"수량은 정수여야 합니다: {qty!r}")
    if not allow_negative and n < 0:
        raise ValueError(f"수량은 음수가 될 수 없습니다: {n}")
    if not allow_zero and n == 0:
        raise ValueError("수량은 0이 될 수 없습니다.")
    return n


# ════════════════════════════════════════════════════════════
# 세트 BOM 분해 / 재고조사 집계 (순수 함수 — DB 무관, 테스트 용이)
# ════════════════════════════════════════════════════════════
def explode_set(
    set_code: str, set_qty: int, bom_map: dict[str, list[dict[str, Any]]]
) -> dict[str, int]:
    """세트 1종 × 수량을 BOM 기준으로 낱개 아이템 수량으로 분해.

    bom_map: { set_code: [{"component_code": str, "component_qty": int}, ...] }
    반환: { component_code: 분해수량 }
    BOM 이 없으면 빈 dict(호출측에서 경고 처리).
    """
    out: dict[str, int] = {}
    for comp in bom_map.get(_norm(set_code), []):
        code = _norm(comp.get("component_code"))
        try:
            per = int(comp.get("component_qty") or 0)
        except (TypeError, ValueError):
            per = 0
        if not code or per <= 0:
            continue
        out[code] = out.get(code, 0) + per * int(set_qty)
    return out


def explode_stocktake(
    lines: Iterable[dict[str, Any]],
    bom_map: dict[str, list[dict[str, Any]]],
) -> dict[str, dict[str, Any]]:
    """재고조사 라인을 낱개 아이템 기준 최종 수량으로 집계.

    입력 라인: [{"sku_code": str, "qty": int}, ...]
      - 낱개 코드(MT/MX/MZ) qty → direct_qty 로 누적
      - 세트 코드(MY) qty       → set_bom 으로 분해해 from_set_qty 로 누적

    반환(낱개 아이템 코드 기준):
      { item_code: {
          "direct_qty":   직접 조사 수량 합,
          "from_set_qty": 세트 분해 수량 합,
          "total_qty":    direct_qty + from_set_qty,
        } }

    추가로 결과에 포함되는 진단 키는 없다(순수 수량만). 세트 BOM 누락
    경고는 `missing_bom_sets()` 로 별도 확인한다.
    """
    result: dict[str, dict[str, int]] = {}

    def _bucket(code: str) -> dict[str, int]:
        return result.setdefault(code, {"direct_qty": 0, "from_set_qty": 0})

    for ln in lines:
        code = _norm(ln.get("sku_code"))
        try:
            qty = int(ln.get("qty") or 0)
        except (TypeError, ValueError):
            qty = 0
        if not code or qty < 0:
            continue
        if is_individual_code(code):
            _bucket(code)["direct_qty"] += qty
        elif is_set_code(code):
            for comp_code, comp_qty in explode_set(code, qty, bom_map).items():
                _bucket(comp_code)["from_set_qty"] += comp_qty
        # blocked/unknown 은 무시(검증 단계에서 이미 막힘)

    out: dict[str, dict[str, Any]] = {}
    for code, v in result.items():
        total = v["direct_qty"] + v["from_set_qty"]
        out[code] = {
            "item_code": code,
            "direct_qty": v["direct_qty"],
            "from_set_qty": v["from_set_qty"],
            "total_qty": total,
        }
    return out


def missing_bom_sets(
    lines: Iterable[dict[str, Any]],
    bom_map: dict[str, list[dict[str, Any]]],
) -> list[str]:
    """재고조사에 입력된 MY- 세트 중 BOM 구성이 없는 코드 목록.

    요구사항 8: 'BOM 이 없는 MY- 코드가 재고조사에 입력되면 경고/오류'.
    """
    missing: list[str] = []
    seen: set[str] = set()
    for ln in lines:
        code = _norm(ln.get("sku_code"))
        if not is_set_code(code) or code in seen:
            continue
        seen.add(code)
        if not bom_map.get(code):
            missing.append(code)
    return missing