king
d3b06e1695
- 콤보 출고는 콤보 줄(set_movement)로만 표시. 분해된 구성 낱개는 숨김
· create_set_out 구성 낱개 ref_type 을 SET_COMPONENT_REF('set')로 고정
· 이동이력 낱개 목록에서 ref_type=SET_COMPONENT_REF 제외
- 직접 낱개 입력만 '낱개', 콤보 입력만 '콤보'로 표기(재고 계산엔 영향 없음)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
986 lines
41 KiB
Python
986 lines
41 KiB
Python
"""말레이시아 창고 재고관리 모듈 라우터.
|
|
|
|
- 경로: /malaysia
|
|
- 권한: 로그인 + `malaysia` 모듈 권한 (관리자는 항상 통과). 서버 측 검사.
|
|
- 데이터: MalaysiaStockStore (malaysia_stock_db / PostgreSQL) 전용.
|
|
MALAYSIA_STOCK_DB_URL 미설정 시 store 가 None 이며, 각 페이지는 "설정 필요" 안내.
|
|
- 세트 BOM: itemcode_db(set_components) 읽기 전용(MalaysiaItemcodeReader). 별도 관리 메뉴 없음.
|
|
|
|
입력 방식: 입출고/재고조사 모두 전체 아이템(+세트) 그리드에 수량만 키인 → 일괄 등록.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Body, Depends, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse
|
|
|
|
from app.timezone import today_kst
|
|
|
|
from . import store
|
|
|
|
router = APIRouter(prefix="/malaysia", tags=["malaysia"])
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# 공용 헬퍼
|
|
# ────────────────────────────────────────────────────────────
|
|
def _store(request: Request) -> Any:
|
|
return getattr(request.app.state, "malaysia_store", None)
|
|
|
|
|
|
def _reader(request: Request) -> Any:
|
|
return getattr(request.app.state, "malaysia_itemcode", None)
|
|
|
|
|
|
def _bom_map(request: Request) -> dict[str, list[dict[str, Any]]]:
|
|
"""itemcode_db(set_components)에서 MY- 세트 BOM 을 읽어 반환. 리더 없으면 {}."""
|
|
r = _reader(request)
|
|
return r.set_bom_map() if r else {}
|
|
|
|
|
|
_WEEKDAYS_KO = ("월", "화", "수", "목", "금", "토", "일")
|
|
|
|
|
|
def _date_label(iso: str | None) -> str:
|
|
"""'2026-06-11' → '2026-06-11 (목)'. 파싱 실패 시 원문."""
|
|
from datetime import date as _date # noqa: WPS433
|
|
|
|
s = (iso or "").strip()
|
|
try:
|
|
d = _date.fromisoformat(s[:10])
|
|
except (ValueError, TypeError):
|
|
return s
|
|
return f"{d.isoformat()} ({_WEEKDAYS_KO[d.weekday()]})"
|
|
|
|
|
|
def _require_user(request: Request) -> dict[str, Any]:
|
|
from app.main import get_current_user_record # noqa: WPS433
|
|
from app.store import has_module # noqa: WPS433
|
|
|
|
user = get_current_user_record(request)
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="로그인이 필요합니다.")
|
|
if not has_module(user, "malaysia"):
|
|
raise HTTPException(status_code=403, detail="말레이시아 재고관리 모듈 권한이 없습니다.")
|
|
return user
|
|
|
|
|
|
def _render_config_needed(request: Request, user: dict[str, Any]) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
return render_template(
|
|
request,
|
|
"denied.html",
|
|
{
|
|
"reason": "말레이시아 재고관리 모듈이 아직 설정되지 않았습니다. "
|
|
"MALAYSIA_STOCK_DB_URL 환경변수를 설정하고 "
|
|
"scripts/sql/malaysia_stock_db_init.sql 로 malaysia_stock_db 를 "
|
|
"초기화한 뒤 컨테이너를 재기동하세요.",
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="malaysia"),
|
|
},
|
|
status_code=503,
|
|
)
|
|
|
|
|
|
def _guard(request: Request) -> tuple[Any, dict[str, Any]] | HTMLResponse | RedirectResponse:
|
|
"""로그인+권한+store 점검. 페이지 핸들러 진입부에서 사용."""
|
|
from app.main import get_current_user_record, render_template # noqa: WPS433
|
|
from app.store import has_module, is_admin # noqa: WPS433
|
|
|
|
user = get_current_user_record(request)
|
|
if user is None:
|
|
return RedirectResponse(url="/login", status_code=303)
|
|
if not has_module(user, "malaysia"):
|
|
return render_template(
|
|
request,
|
|
"denied.html",
|
|
{"reason": "말레이시아 재고관리 모듈 접근 권한이 없습니다.", "is_admin": is_admin(user)},
|
|
status_code=403,
|
|
)
|
|
st = _store(request)
|
|
if st is None:
|
|
return _render_config_needed(request, user)
|
|
return st, user
|
|
|
|
|
|
def _selected_wh(request: Request, st: Any) -> str:
|
|
"""쿼리 ?wh= 또는 기본(첫 활성 창고)."""
|
|
wh = (request.query_params.get("wh") or "").strip()
|
|
codes = [w["warehouse_code"] for w in st.list_warehouses()]
|
|
if wh in codes:
|
|
return wh
|
|
return codes[0] if codes else "MY-WH-01"
|
|
|
|
|
|
def _base_ctx(request: Request, user: dict[str, Any], st: Any, *, wh: str) -> dict[str, Any]:
|
|
from app.main import build_erp_nav # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
reader = _reader(request)
|
|
ko_names = reader.name_map() if reader else {}
|
|
return {
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"is_super": bool(user.get("is_super_admin")),
|
|
"nav_items": build_erp_nav(user, active="malaysia"),
|
|
"warehouses": st.list_warehouses(),
|
|
"selected_wh": wh,
|
|
"ko_names": ko_names,
|
|
}
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 재고 현황 (메인)
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request) -> HTMLResponse:
|
|
from app.main import render_template # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
st, user = guard
|
|
wh = _selected_wh(request, st)
|
|
ctx = _base_ctx(request, user, st, wh=wh)
|
|
rows = st.stock_status(warehouse_code=wh)
|
|
# 마지막 재고조사(취소 제외) 의 낱개 기준 total 과 전산재고 차이
|
|
latest = st.latest_stocktake(warehouse_code=wh)
|
|
last_date = latest["stocktake_date"] if latest else None
|
|
last_map: dict[str, int] = {}
|
|
if latest:
|
|
comp = st.compute_result(stocktake_id=latest["id"], bom_map=_bom_map(request))
|
|
last_map = {r["item_code"]: r["total_qty"] for r in comp["rows"]}
|
|
for r in rows:
|
|
code = r["item_code"]
|
|
lq = last_map.get(code)
|
|
r["last_stocktake_date"] = last_date
|
|
r["last_qty"] = lq
|
|
r["diff"] = (lq - r["current_qty"]) if lq is not None else None
|
|
set_stock = st.latest_set_quantities(warehouse_code=wh)
|
|
ctx.update(
|
|
{
|
|
"page_title": "말레이시아 재고관리",
|
|
"page_subtitle": f"{wh} · 재고 현황",
|
|
"rows": rows,
|
|
"set_rows": set_stock["rows"],
|
|
"set_stocktake_date": set_stock["stocktake_date"],
|
|
}
|
|
)
|
|
return render_template(request, "malaysia/index.html", ctx)
|
|
|
|
|
|
@router.get("/rack", response_class=HTMLResponse)
|
|
async def rack_view(request: Request) -> HTMLResponse:
|
|
"""창고 랙 보기 — 최근(취소 제외) 재고조사의 랙 입력을 칸별로 한눈에.
|
|
|
|
읽기 전용. 각 칸에 어떤 상품이 몇 박스(× 입수량 = 총개수) 있는지 표시.
|
|
"""
|
|
from app.main import render_template # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
st, user = guard
|
|
wh = _selected_wh(request, st)
|
|
# 랙보기는 '랙을 가장 최근에 저장한' 조사 기준(조사 날짜 아님).
|
|
latest = st.latest_rack_stocktake(warehouse_code=wh)
|
|
rack_by_cell: dict[str, list[dict[str, Any]]] = {}
|
|
# 뚜껑(MD-)은 malaysia_items 에 없으므로 이름맵에 병합해 코드 대신 이름 표시.
|
|
names = {**st.item_name_map(), **store.lid_name_map()}
|
|
if latest:
|
|
for e in st.list_rack_entries(stocktake_id=latest["id"]):
|
|
e["item_name"] = names.get(e["sku_code"], e["sku_code"])
|
|
e["total"] = int(e["units_per_box"]) * int(e["box_count"])
|
|
rack_by_cell.setdefault(e["cell_code"], []).append(e)
|
|
ctx = _base_ctx(request, user, st, wh=wh)
|
|
ctx.update(
|
|
{
|
|
"page_title": "말레이시아 재고관리 — 창고 랙",
|
|
"page_subtitle": f"{wh} · 랙 보기",
|
|
"rack_layout": store.rack_layout(),
|
|
"rack_by_cell": rack_by_cell,
|
|
"stocktake": latest,
|
|
}
|
|
)
|
|
return render_template(request, "malaysia/rack_view.html", ctx)
|
|
|
|
|
|
@router.post("/rack/save")
|
|
async def rack_save(
|
|
request: Request, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""랙 보기에서 드래그로 재배치한 칸 구성을 저장(위치만 변경).
|
|
|
|
대상은 랙 보기가 보여주는 조사(latest_rack_stocktake). 수량 합계는
|
|
불변이며 확정본도 허용. 폼은 stocktake_detail 과 동일한 병렬배열
|
|
(rk_cell/rk_sku/rk_upb/rk_box) → _parse_rack_form 재사용.
|
|
"""
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
form = await request.form()
|
|
wh = str(form.get("warehouse_code") or "").strip() or _selected_wh(request, st)
|
|
latest = st.latest_rack_stocktake(warehouse_code=wh)
|
|
if not latest:
|
|
raise HTTPException(status_code=400, detail="저장할 랙 재고조사가 없습니다.")
|
|
entries = _parse_rack_form(form)
|
|
try:
|
|
st.reposition_rack(stocktake_id=latest["id"], entries=entries)
|
|
except (KeyError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/rack?wh={wh}", status_code=303)
|
|
|
|
|
|
@router.post("/reset")
|
|
async def reset_data(
|
|
request: Request, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""전체 창고 재고 수량 데이터 초기화 — 슈퍼관리자 전용.
|
|
|
|
stock_movement + daily_stocktake(+line) 전체 삭제(현재고/조사값 0).
|
|
마스터(warehouses, malaysia_items)는 보존.
|
|
"""
|
|
if not user.get("is_super_admin"):
|
|
raise HTTPException(status_code=403, detail="재고 데이터 초기화는 슈퍼관리자만 가능합니다.")
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
st.reset_all_data()
|
|
return RedirectResponse(url="/malaysia/", status_code=303)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 입고 / 출고 등록 (그리드 일괄)
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/movements", response_class=HTMLResponse)
|
|
async def movements_page(request: Request) -> HTMLResponse:
|
|
from app.main import render_template # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
st, user = guard
|
|
wh = _selected_wh(request, st)
|
|
mtype = (request.query_params.get("type") or "").strip().upper()
|
|
fdate = (request.query_params.get("date") or "").strip()
|
|
mt = mtype if mtype in store.MOVEMENT_TYPES else None
|
|
raw_singles = st.list_movements(
|
|
warehouse_code=wh, movement_type=mt,
|
|
date_from=fdate or None, date_to=fdate or None, limit=300,
|
|
)
|
|
# 콤보 분해로 생긴 구성 낱개(ref_type=SET_COMPONENT_REF)는 '낱개 입력'이 아니므로
|
|
# 이동이력에서 숨긴다. 콤보는 아래 set_movement 줄로 표시된다.
|
|
singles = [m for m in raw_singles if (m.get("ref_type") or "") != store.SET_COMPONENT_REF]
|
|
for m in singles:
|
|
m["kind"] = "individual"
|
|
# 콤보(세트) 출고 원장도 함께 보여준다(낱개/콤보 분리 표시).
|
|
combos = st.list_set_movements(
|
|
warehouse_code=wh, movement_type=mt,
|
|
date_from=fdate or None, date_to=fdate or None, limit=300,
|
|
)
|
|
for m in combos:
|
|
m["kind"] = "set"
|
|
m["item_code"] = m.get("set_code", "")
|
|
# 날짜(그다음 id) 내림차순 병합.
|
|
movements = sorted(
|
|
singles + combos,
|
|
key=lambda m: (str(m.get("movement_date") or ""), int(m.get("id") or 0)),
|
|
reverse=True,
|
|
)
|
|
for m in movements:
|
|
m["date_label"] = _date_label(m.get("movement_date"))
|
|
ctx = _base_ctx(request, user, st, wh=wh)
|
|
ctx.update(
|
|
{
|
|
"page_title": "말레이시아 재고관리 — 입출고",
|
|
"page_subtitle": f"{wh} · 입고/출고/조정 일괄 등록",
|
|
"individual_items": st.list_items(kind="individual"),
|
|
"set_items": st.list_items(kind="set"),
|
|
"movements": movements,
|
|
"today": today_kst().isoformat(),
|
|
"filter_type": mtype if mtype in store.MOVEMENT_TYPES else "",
|
|
"filter_date": fdate,
|
|
}
|
|
)
|
|
return render_template(request, "malaysia/movements.html", ctx)
|
|
|
|
|
|
@router.post("/movements/apply")
|
|
async def movements_apply(request: Request, user: dict[str, Any] = Depends(_require_user)) -> RedirectResponse:
|
|
"""낱개 + 세트 한 화면 일괄 적용 (버튼 하나).
|
|
|
|
form: movement_type(낱개용 IN/OUT/ADJUST), movement_date, warehouse_code,
|
|
ref_type, ref_no, memo, qty_<CODE>...
|
|
|
|
- 낱개(individual): 선택한 movement_type 으로 등록. IN/OUT 양수만, ADJUST ±(0 스킵).
|
|
- 세트(set): 항상 OUT 으로 처리 — BOM(itemcode_db) 분해해 구성품별 OUT 생성.
|
|
콤보는 입고 불가(낱개로 분리 입고). IN 선택 + 콤보 수량 입력 시 400 안내.
|
|
빈칸/0 은 스킵. 코드별 에러는 모아서 부분 적용 후 안내.
|
|
"""
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
form = await request.form()
|
|
mtype = str(form.get("movement_type") or "").strip().upper()
|
|
if mtype not in ("IN", "OUT", "ADJUST"):
|
|
raise HTTPException(status_code=400, detail="이동 종류 오류")
|
|
wh = str(form.get("warehouse_code") or "").strip()
|
|
mdate = str(form.get("movement_date") or "").strip()
|
|
ref_type = str(form.get("ref_type") or "").strip()
|
|
ref_no = str(form.get("ref_no") or "").strip()
|
|
memo = str(form.get("memo") or "").strip()
|
|
bmap = _bom_map(request)
|
|
|
|
# 콤보는 입고 불가 — IN 선택 + 세트 수량 입력 시 차단(낱개로 분리 입고 안내)
|
|
if mtype == "IN":
|
|
set_codes_keyed = [
|
|
it["item_code"]
|
|
for it in st.list_items(kind="set")
|
|
if str(form.get(f"qty_{it['item_code']}") or "").strip() not in ("", "0")
|
|
]
|
|
if set_codes_keyed:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="콤보 상품은 입고할 수 없습니다. 낱개로 분리해서 입고하세요. "
|
|
"(입고 불가 콤보: " + ", ".join(set_codes_keyed) + ")",
|
|
)
|
|
|
|
applied = 0
|
|
errors: list[str] = []
|
|
|
|
# 낱개 — 선택한 종류로
|
|
for it in st.list_items(kind="individual"):
|
|
code = it["item_code"]
|
|
raw = str(form.get(f"qty_{code}") or "").strip()
|
|
if raw == "":
|
|
continue
|
|
try:
|
|
n = int(raw)
|
|
except ValueError:
|
|
errors.append(f"{code}: 숫자 아님")
|
|
continue
|
|
if n == 0:
|
|
continue
|
|
try:
|
|
st.create_movement(
|
|
movement_date=mdate, warehouse_code=wh, item_code=code,
|
|
movement_type=mtype, qty=n, ref_type=ref_type, ref_no=ref_no,
|
|
memo=memo, created_by=user["email"],
|
|
)
|
|
applied += 1
|
|
except ValueError as exc:
|
|
errors.append(f"{code}: {exc}")
|
|
|
|
# 세트 — 항상 OUT(BOM 분해). 콤보 입고는 위에서 이미 차단됨.
|
|
for it in st.list_items(kind="set"):
|
|
code = it["item_code"]
|
|
raw = str(form.get(f"qty_{code}") or "").strip()
|
|
if raw == "":
|
|
continue
|
|
try:
|
|
n = int(raw)
|
|
except ValueError:
|
|
errors.append(f"{code}: 숫자 아님")
|
|
continue
|
|
if n <= 0:
|
|
continue
|
|
try:
|
|
st.create_set_out(
|
|
movement_date=mdate, warehouse_code=wh, set_code=code, set_qty=n,
|
|
bom_map=bmap, ref_type="set", ref_no=ref_no, memo=memo,
|
|
created_by=user["email"],
|
|
)
|
|
applied += 1
|
|
except ValueError as exc:
|
|
errors.append(f"{code}: {exc}")
|
|
|
|
if errors and applied == 0:
|
|
raise HTTPException(status_code=400, detail="; ".join(errors[:10]))
|
|
return RedirectResponse(url=f"/malaysia/movements?wh={wh}&type={mtype}", status_code=303)
|
|
|
|
|
|
@router.get("/movements/export")
|
|
async def movements_export(request: Request) -> StreamingResponse:
|
|
"""해당 일자/종류(IN/OUT)에 입출고된 낱개 수량을 엑셀로 내보낸다.
|
|
|
|
컬럼: A=사방넷 코드, B=수량, C=이름.
|
|
세트 OUT 은 생성 시 구성품 OUT 으로 분해 저장되므로 낱개 기준으로 합산된다.
|
|
"""
|
|
import io # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard # type: ignore[return-value]
|
|
st, _user = guard
|
|
wh = _selected_wh(request, st)
|
|
mtype = (request.query_params.get("type") or "OUT").strip().upper()
|
|
if mtype not in ("IN", "OUT"):
|
|
raise HTTPException(status_code=400, detail="type 은 IN 또는 OUT 이어야 합니다.")
|
|
mdate = (request.query_params.get("date") or today_kst().isoformat()).strip()
|
|
|
|
totals = st.daily_movement_totals(
|
|
warehouse_code=wh, movement_date=mdate, movement_type=mtype
|
|
)
|
|
reader = _reader(request)
|
|
sabang = reader.sabangnet_code_map() if reader else {}
|
|
names = reader.name_map() if reader else {}
|
|
item_names = st.item_name_map()
|
|
|
|
from openpyxl import Workbook # noqa: WPS433
|
|
from openpyxl.styles import Alignment, Font, PatternFill # noqa: WPS433
|
|
from openpyxl.utils import get_column_letter # noqa: WPS433
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = f"{mtype}_{mdate}"
|
|
header = ["사방넷코드", "수량", "이름"]
|
|
ws.append(header)
|
|
for code in sorted(totals.keys()):
|
|
qty = totals[code]
|
|
if qty == 0:
|
|
continue
|
|
ws.append([
|
|
sabang.get(code, ""),
|
|
int(qty),
|
|
names.get(code) or item_names.get(code) or code,
|
|
])
|
|
|
|
# 제목행(A1~C1): 파랑 배경 + 흰색 굵은글씨 + 가운데 정렬
|
|
fill = PatternFill(start_color="FF4472C4", end_color="FF4472C4", fill_type="solid")
|
|
font = Font(color="FFFFFFFF", bold=True)
|
|
center = Alignment(horizontal="center", vertical="center")
|
|
for col in range(1, len(header) + 1):
|
|
cell = ws.cell(row=1, column=col)
|
|
cell.fill = fill
|
|
cell.font = font
|
|
cell.alignment = center
|
|
|
|
# 열너비 자동(내용 길이 기준, 한글 폭 보정). 최소 8 ~ 최대 50.
|
|
for col in range(1, len(header) + 1):
|
|
letter = get_column_letter(col)
|
|
longest = 0
|
|
for cell in ws[letter]:
|
|
v = "" if cell.value is None else str(cell.value)
|
|
width = sum(2 if ord(ch) > 0x1100 else 1 for ch in v)
|
|
longest = max(longest, width)
|
|
ws.column_dimensions[letter].width = max(8, min(50, longest + 2))
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
fname = f"malaysia_{mtype}_{mdate}.xlsx"
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
|
|
)
|
|
|
|
|
|
@router.get("/rack/export")
|
|
async def rack_export(request: Request) -> StreamingResponse:
|
|
"""창고 랙 보기 데이터를 엑셀로 내보낸다.
|
|
|
|
랙 보기와 동일 기준(latest_rack_stocktake)의 모든 칸 항목을 한 행씩.
|
|
컬럼: 아이템 코드 / 이름 / 수량 / 랙번호. 표 전체에 자동필터,
|
|
각 열은 내용 길이에 맞춰 너비 자동.
|
|
"""
|
|
import io # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard # type: ignore[return-value]
|
|
st, _user = guard
|
|
wh = _selected_wh(request, st)
|
|
latest = st.latest_rack_stocktake(warehouse_code=wh)
|
|
names = {**st.item_name_map(), **store.lid_name_map()}
|
|
|
|
rows: list[tuple[str, str, int, str]] = []
|
|
if latest:
|
|
for e in st.list_rack_entries(stocktake_id=latest["id"]):
|
|
code = e["sku_code"]
|
|
qty = int(e["units_per_box"]) * int(e["box_count"])
|
|
rows.append((code, names.get(code, code), qty, e["cell_code"]))
|
|
# 랙번호 → 아이템코드 순 정렬(보기 흐름과 동일하게 칸 기준).
|
|
rows.sort(key=lambda r: (r[3], r[0]))
|
|
|
|
from openpyxl import Workbook # noqa: WPS433
|
|
from openpyxl.styles import Alignment, Font, PatternFill # noqa: WPS433
|
|
from openpyxl.utils import get_column_letter # noqa: WPS433
|
|
|
|
# 아이템별 합계(코드 기준 수량 합산). 코드→이름 보존, 코드순 정렬.
|
|
summary: dict[str, int] = {}
|
|
for code, _name, qty, _cell in rows:
|
|
summary[code] = summary.get(code, 0) + qty
|
|
summary_rows = sorted(
|
|
((code, names.get(code, code), total) for code, total in summary.items()),
|
|
key=lambda r: r[0],
|
|
)
|
|
|
|
fill = PatternFill(start_color="FF4472C4", end_color="FF4472C4", fill_type="solid")
|
|
font = Font(color="FFFFFFFF", bold=True)
|
|
center = Alignment(horizontal="center", vertical="center")
|
|
|
|
def _style_header(row_idx: int, ncols: int) -> None:
|
|
for col in range(1, ncols + 1):
|
|
cell = ws.cell(row=row_idx, column=col)
|
|
cell.fill = fill
|
|
cell.font = font
|
|
cell.alignment = center
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "rack"
|
|
|
|
# ── 상세 표 ──
|
|
header = ["아이템 코드", "이름", "수량", "랙번호"]
|
|
ws.append(header)
|
|
for code, name, qty, cell in rows:
|
|
ws.append([code, name, qty, cell])
|
|
_style_header(1, len(header))
|
|
detail_last = ws.max_row # 자동필터 범위(상세표만)
|
|
ws.auto_filter.ref = f"A1:{get_column_letter(len(header))}{detail_last}"
|
|
|
|
# ── 아이템별 합계 표 (상세표 아래, 한 줄 띄움) ──
|
|
# 빈 셀 행으로 한 줄 띄움(append([]) 은 셀이 없어 max_row 가 안 늘어 위치가
|
|
# 어긋나므로 빈 문자열 셀을 넣는다).
|
|
ws.append([""])
|
|
sum_header = ["아이템 코드", "이름", "합계 수량"]
|
|
ws.append(sum_header)
|
|
sum_header_row = ws.max_row # 방금 추가한 합계 제목행(셀이 있어 정확)
|
|
for code, name, total in summary_rows:
|
|
ws.append([code, name, total])
|
|
_style_header(sum_header_row, len(sum_header))
|
|
|
|
# 열너비 자동(내용 길이 기준, 한글 폭 보정). 최소 8 ~ 최대 50.
|
|
for col in range(1, len(header) + 1):
|
|
letter = get_column_letter(col)
|
|
longest = 0
|
|
for cell in ws[letter]:
|
|
v = "" if cell.value is None else str(cell.value)
|
|
width = sum(2 if ord(ch) > 0x1100 else 1 for ch in v)
|
|
longest = max(longest, width)
|
|
ws.column_dimensions[letter].width = max(8, min(50, longest + 2))
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
sdate = (latest or {}).get("stocktake_date", today_kst().isoformat())
|
|
fname = f"malaysia_rack_{wh}_{sdate}.xlsx"
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 일일 재고조사
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/stocktakes", response_class=HTMLResponse)
|
|
async def stocktakes_page(request: Request) -> HTMLResponse:
|
|
from app.main import render_template # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
st, user = guard
|
|
wh = _selected_wh(request, st)
|
|
ctx = _base_ctx(request, user, st, wh=wh)
|
|
ctx.update(
|
|
{
|
|
"page_title": "말레이시아 재고관리 — 일일 재고조사",
|
|
"page_subtitle": f"{wh} · 조사 목록",
|
|
"stocktakes": st.list_stocktakes(warehouse_code=wh),
|
|
"today": today_kst().isoformat(),
|
|
}
|
|
)
|
|
return render_template(request, "malaysia/stocktakes.html", ctx)
|
|
|
|
|
|
@router.post("/stocktakes")
|
|
async def stocktake_create(
|
|
request: Request,
|
|
stocktake_date: str = Form(...),
|
|
warehouse_code: str = Form(...),
|
|
memo: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
head = st.create_stocktake(
|
|
stocktake_date=stocktake_date, warehouse_code=warehouse_code,
|
|
memo=memo, created_by=user["email"],
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{head['id']}", status_code=303)
|
|
|
|
|
|
@router.get("/stocktakes/{stocktake_id:int}", response_class=HTMLResponse)
|
|
async def stocktake_detail(request: Request, stocktake_id: int) -> HTMLResponse:
|
|
from app.main import render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
st, user = guard
|
|
try:
|
|
computed = st.compute_result(stocktake_id=stocktake_id, bom_map=_bom_map(request))
|
|
except KeyError:
|
|
return render_template(
|
|
request, "denied.html",
|
|
{"reason": "재고조사를 찾을 수 없습니다.", "is_admin": is_admin(user)},
|
|
status_code=404,
|
|
)
|
|
head = computed["stocktake"]
|
|
wh = head["warehouse_code"]
|
|
# 입력된 라인 qty 를 코드별 dict 로 (그리드 기본값)
|
|
entered = {ln["sku_code"]: ln["qty"] for ln in head.get("lines", [])}
|
|
# 랙 입력 항목을 칸별로 묶어 전달(그림 안 입력행 복원용)
|
|
rack_by_cell: dict[str, list[dict[str, Any]]] = {}
|
|
for e in st.list_rack_entries(stocktake_id=stocktake_id):
|
|
rack_by_cell.setdefault(e["cell_code"], []).append(e)
|
|
is_draft = head["status"] == "draft"
|
|
# 작성중일 때만 '이전값 불러오기' 가능 여부 계산(직전 조사에 랙 입력이 있을 때)
|
|
has_prev_rack = bool(
|
|
is_draft
|
|
and st.previous_rack_entries(warehouse_code=wh, exclude_stocktake_id=stocktake_id)
|
|
)
|
|
ctx = _base_ctx(request, user, st, wh=wh)
|
|
ctx.update(
|
|
{
|
|
"page_title": f"재고조사 #{stocktake_id}",
|
|
"page_subtitle": f"{head['stocktake_date']} · {wh} · {head['status']}",
|
|
"stocktake": head,
|
|
"rack_layout": store.rack_layout(),
|
|
"rack_by_cell": rack_by_cell,
|
|
# 낱개·세트 + 뚜껑(MD-, 위치확인용). 뚜껑은 재고 집계엔 빠진다.
|
|
"rack_items": st.list_items() + list(store.LID_ITEMS),
|
|
"entered": entered,
|
|
"result_rows": computed["rows"],
|
|
"missing_bom": computed["missing_bom"],
|
|
"is_draft": is_draft,
|
|
"has_prev_rack": has_prev_rack,
|
|
}
|
|
)
|
|
return render_template(request, "malaysia/stocktake_detail.html", ctx)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/lines/bulk")
|
|
async def stocktake_lines_bulk(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""그리드 일괄 저장. form: qty_<SKUCODE>... (낱개+세트). 빈칸은 스킵, 0 허용."""
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
form = await request.form()
|
|
codes = [i["item_code"] for i in st.list_items(kind="individual")]
|
|
codes += [i["item_code"] for i in st.list_items(kind="set")]
|
|
errors: list[str] = []
|
|
for code in codes:
|
|
raw = str(form.get(f"qty_{code}") or "").strip()
|
|
if raw == "":
|
|
continue
|
|
try:
|
|
n = int(raw)
|
|
except ValueError:
|
|
errors.append(f"{code}: 숫자 아님")
|
|
continue
|
|
try:
|
|
st.upsert_line(stocktake_id=stocktake_id, sku_code=code, qty=n)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
errors.append(f"{code}: {exc}")
|
|
if errors:
|
|
raise HTTPException(status_code=400, detail="; ".join(errors[:10]))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
def _parse_rack_form(form: Any) -> list[dict[str, Any]]:
|
|
"""랙 폼 병렬배열(rk_cell/rk_sku/rk_upb/rk_box) → 입력 항목 리스트.
|
|
|
|
SKU 미선택/입수량·박스수 ≤ 0 인 행은 스킵(미입력 행 취급).
|
|
"""
|
|
cells = form.getlist("rk_cell")
|
|
skus = form.getlist("rk_sku")
|
|
upbs = form.getlist("rk_upb")
|
|
boxes = form.getlist("rk_box")
|
|
entries: list[dict[str, Any]] = []
|
|
for cell, sku, upb, box in zip(cells, skus, upbs, boxes):
|
|
code = str(sku or "").strip()
|
|
if not code:
|
|
continue
|
|
try:
|
|
u = int(str(upb).strip() or 0)
|
|
b = int(str(box).strip() or 0)
|
|
except ValueError:
|
|
continue
|
|
if u <= 0 or b <= 0:
|
|
continue
|
|
entries.append(
|
|
{"cell_code": cell, "sku_code": code, "units_per_box": u, "box_count": b}
|
|
)
|
|
return entries
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/rack/bulk")
|
|
async def stocktake_rack_bulk(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""랙 입력 일괄 저장.
|
|
|
|
저장 시 랙 항목 교체 + SKU별 집계로 daily_stocktake_line 재생성.
|
|
"""
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
form = await request.form()
|
|
entries = _parse_rack_form(form)
|
|
try:
|
|
st.replace_rack(stocktake_id=stocktake_id, entries=entries)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/rack/load-previous")
|
|
async def stocktake_rack_load_previous(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""직전 재고조사의 랙 저장값을 현재(작성중) 조사로 불러와 채운다.
|
|
|
|
이전 랙 항목을 그대로 replace_rack 으로 적용 → 화면에 복원되어 표시된다.
|
|
사용자는 이어서 수정 후 저장/확정한다. 현재 입력은 덮어쓰므로 화면에서 confirm.
|
|
"""
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
head = st.get_stocktake(stocktake_id=stocktake_id, with_lines=False)
|
|
if head is None:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
if head["status"] != "draft":
|
|
raise HTTPException(status_code=400, detail="작성중인 조사만 불러올 수 있습니다.")
|
|
entries = st.previous_rack_entries(
|
|
warehouse_code=head["warehouse_code"], exclude_stocktake_id=stocktake_id
|
|
)
|
|
if not entries:
|
|
raise HTTPException(status_code=400, detail="불러올 이전 재고조사 랙 데이터가 없습니다.")
|
|
try:
|
|
st.replace_rack(stocktake_id=stocktake_id, entries=entries)
|
|
except (KeyError, ValueError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/rack/finalize")
|
|
async def stocktake_rack_finalize(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""랙 입력 저장 + 즉시 확정(한 번에).
|
|
|
|
확정 버튼이 랙 폼과 별개라 랙 데이터가 누락되는 문제를 막기 위해,
|
|
화면의 랙 입력을 먼저 저장(replace_rack)한 뒤 finalize 한다.
|
|
"""
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
if not is_admin(user):
|
|
raise HTTPException(status_code=403, detail="전산 재고 반영은 관리자만 가능합니다.")
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
form = await request.form()
|
|
entries = _parse_rack_form(form)
|
|
try:
|
|
st.replace_rack(stocktake_id=stocktake_id, entries=entries)
|
|
st.finalize_stocktake(stocktake_id=stocktake_id, bom_map=_bom_map(request))
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/lines/{line_id:int}/delete")
|
|
async def stocktake_line_delete(
|
|
request: Request, stocktake_id: int, line_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
st.delete_line(stocktake_id=stocktake_id, line_id=line_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="라인을 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/finalize")
|
|
async def stocktake_finalize(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
if not is_admin(user):
|
|
raise HTTPException(status_code=403, detail="전산 재고 반영은 관리자만 가능합니다.")
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
st.finalize_stocktake(stocktake_id=stocktake_id, bom_map=_bom_map(request))
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/cancel")
|
|
async def stocktake_cancel(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
st.cancel_stocktake(stocktake_id=stocktake_id)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/malaysia/stocktakes/{stocktake_id}", status_code=303)
|
|
|
|
|
|
@router.post("/stocktakes/{stocktake_id:int}/delete")
|
|
async def stocktake_delete(
|
|
request: Request, stocktake_id: int, user: dict[str, Any] = Depends(_require_user)
|
|
) -> RedirectResponse:
|
|
"""재고조사 완전 삭제 — 슈퍼관리자 전용."""
|
|
if not user.get("is_super_admin"):
|
|
raise HTTPException(status_code=403, detail="재고조사 삭제는 슈퍼관리자만 가능합니다.")
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
wh = _selected_wh(request, st)
|
|
try:
|
|
st.delete_stocktake(stocktake_id=stocktake_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
return RedirectResponse(url=f"/malaysia/stocktakes?wh={wh}", status_code=303)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# JSON API
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/api/warehouses")
|
|
async def api_warehouses(request: Request, _: dict[str, Any] = Depends(_require_user)) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
return JSONResponse({"warehouses": st.list_warehouses()})
|
|
|
|
|
|
@router.get("/api/items")
|
|
async def api_items(request: Request, kind: str = "", _: dict[str, Any] = Depends(_require_user)) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
k = kind if kind in ("individual", "set") else None
|
|
return JSONResponse({"items": st.list_items(kind=k)})
|
|
|
|
|
|
@router.get("/api/bom")
|
|
async def api_bom(request: Request, _: dict[str, Any] = Depends(_require_user)) -> JSONResponse:
|
|
"""itemcode_db(set_components)에서 읽은 MY- 세트 BOM (읽기 전용)."""
|
|
return JSONResponse({"bom": _bom_map(request)})
|
|
|
|
|
|
@router.get("/api/stock")
|
|
async def api_stock(request: Request, wh: str = "", _: dict[str, Any] = Depends(_require_user)) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
wh = wh.strip() or _selected_wh(request, st)
|
|
return JSONResponse({"warehouse_code": wh, "rows": st.stock_status(warehouse_code=wh)})
|
|
|
|
|
|
@router.post("/api/movements")
|
|
async def api_movement_create(
|
|
request: Request, payload: dict[str, Any] = Body(...), user: dict[str, Any] = Depends(_require_user)
|
|
) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
row = st.create_movement(
|
|
movement_date=str(payload.get("movement_date") or ""),
|
|
warehouse_code=str(payload.get("warehouse_code") or ""),
|
|
item_code=str(payload.get("item_code") or ""),
|
|
movement_type=str(payload.get("movement_type") or ""),
|
|
qty=int(payload.get("qty") or 0),
|
|
ref_type=str(payload.get("ref_type") or ""),
|
|
ref_no=str(payload.get("ref_no") or ""),
|
|
memo=str(payload.get("memo") or ""),
|
|
created_by=user["email"],
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"ok": True, "movement": row})
|
|
|
|
|
|
@router.post("/api/movements/set-out")
|
|
async def api_set_out(
|
|
request: Request, payload: dict[str, Any] = Body(...), user: dict[str, Any] = Depends(_require_user)
|
|
) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
rows = st.create_set_out(
|
|
movement_date=str(payload.get("movement_date") or ""),
|
|
warehouse_code=str(payload.get("warehouse_code") or ""),
|
|
set_code=str(payload.get("set_code") or ""),
|
|
set_qty=int(payload.get("set_qty") or 0),
|
|
bom_map=_bom_map(request),
|
|
ref_type=str(payload.get("ref_type") or "set"),
|
|
ref_no=str(payload.get("ref_no") or ""),
|
|
memo=str(payload.get("memo") or ""),
|
|
created_by=user["email"],
|
|
)
|
|
except (ValueError, TypeError) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"ok": True, "movements": rows})
|
|
|
|
|
|
@router.get("/api/stocktakes/{stocktake_id:int}/result")
|
|
async def api_stocktake_result(
|
|
request: Request, stocktake_id: int, _: dict[str, Any] = Depends(_require_user)
|
|
) -> JSONResponse:
|
|
st = _store(request)
|
|
if st is None:
|
|
raise HTTPException(status_code=503, detail="malaysia_stock_db 미설정")
|
|
try:
|
|
computed = st.compute_result(stocktake_id=stocktake_id, bom_map=_bom_map(request))
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="재고조사를 찾을 수 없습니다.")
|
|
return JSONResponse(
|
|
{"stocktake": computed["stocktake"], "rows": computed["rows"], "missing_bom": computed["missing_bom"]}
|
|
)
|
|
|
|
|
|
@router.get("/health")
|
|
async def health() -> dict[str, str]:
|
|
return {"status": "ok", "module": "malaysia"}
|