king
fa396bd237
Side B 다음에 Ground 영역을 추가하고 단일 파렛트 6칸(G1~G6)을 한 줄로 배치. rack_cell_codes 검증 목록과 rack_layout 렌더 구조에 반영. 면 제목을 side.label 로 바꿔 A/B는 'Side A/B', Ground는 'Ground'로 표기. 입력/이동 저장 검증 모두 G1~G6 허용. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
297 lines
11 KiB
Python
297 lines
11 KiB
Python
"""말레이시아 창고 재고관리 모듈 — 상수 및 순수 계산/검증 헬퍼.
|
|
|
|
- 데이터 저장은 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)
|
|
# Side B 아래 바닥(Ground) 파렛트 6칸. 코드: 'G1' ~ 'G6'.
|
|
RACK_GROUND: tuple[str, ...] = ("G1", "G2", "G3", "G4", "G5", "G6")
|
|
|
|
|
|
def rack_cell_codes() -> list[str]:
|
|
"""모든 랙 칸 코드 목록(검증용). 예: 'A3-1-1' ... 'B1-3-2', 'G1'~'G6'."""
|
|
codes = [
|
|
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
|
|
]
|
|
codes.extend(RACK_GROUND)
|
|
return codes
|
|
|
|
|
|
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, "label": f"Side {side}", "rows": rows})
|
|
# Side B 아래 Ground: 6칸(G1~G6)을 한 줄에. 각 칸은 분할 없는 단일 파렛트.
|
|
out.append({
|
|
"side": "Ground",
|
|
"label": "Ground",
|
|
"rows": [{"row": 0, "cols": [[code] for code in RACK_GROUND]}],
|
|
})
|
|
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
|