king
026a9ab526
센터 목록은 이름 등록·조회·관리가 목적이므로 "사용중" 표시를 없애고 활성/비활성/삭제만 남긴다. 좌우 2단(추가 카드 360px + 900px 고정 목록) 대신 상단 추가·검색 바 + 다중 컬럼 그리드로 바꿔 29개 이상 센터를 한 화면에서 훑을 수 있게 했다. - 행마다 폼 하나(formaction/name=active)로 저장·토글·삭제 처리 - 상태는 배지 대신 좌측 점(활성 초록 / 비활성 회색+흐리게) - 클라이언트 검색 필터, 카운트 실시간 갱신 - 라우터에서 center_in_use 조회 제거(화면에서 미사용) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
859 lines
32 KiB
Python
859 lines
32 KiB
Python
"""쿠팡 밀크런 모듈 라우터.
|
|
|
|
- 경로: /cupang
|
|
- 권한: 로그인 + `cupang` 모듈 권한 (관리자는 항상 통과). 서버 측 검사.
|
|
- 데이터: CupangDBStore (cupang_db / PostgreSQL) 전용.
|
|
CUPANG_DB_URL 미설정 시 store 가 None 이며, 각 페이지는 "설정 필요" 안내를 보여준다.
|
|
- 상품 검색: itemcode_db 읽기 전용(ItemcodeReader). 미설정 시 수동 입력 폴백.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import calendar as _calendar
|
|
import json
|
|
from fractions import Fraction
|
|
from typing import Any
|
|
|
|
from app.timezone import today_kst
|
|
|
|
from fastapi import APIRouter, Body, Depends, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
|
|
from .holidays import is_holiday
|
|
from .store import SHIP_METHODS
|
|
|
|
router = APIRouter(prefix="/cupang", tags=["cupang"])
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# 공용 헬퍼
|
|
# ────────────────────────────────────────────────────────────
|
|
def _store(request: Request) -> Any:
|
|
"""CupangDBStore 또는 None(CUPANG_DB_URL 미설정)."""
|
|
return getattr(request.app.state, "cupang_store", None)
|
|
|
|
|
|
def _itemcode(request: Request) -> Any:
|
|
return getattr(request.app.state, "itemcode_reader", None)
|
|
|
|
|
|
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, "cupang"):
|
|
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": "쿠팡 밀크런 모듈이 아직 설정되지 않았습니다. "
|
|
"CUPANG_DB_URL 환경변수를 설정하고 scripts/sql/cupang_db_init.sql 로 "
|
|
"cupang_db 를 초기화한 뒤 컨테이너를 재기동하세요.",
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
},
|
|
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, "cupang"):
|
|
return render_template(
|
|
request,
|
|
"denied.html",
|
|
{"reason": "쿠팡 밀크런 모듈 접근 권한이 없습니다.", "is_admin": is_admin(user)},
|
|
status_code=403,
|
|
)
|
|
store = _store(request)
|
|
if store is None:
|
|
return _render_config_needed(request, user)
|
|
return store, user
|
|
|
|
|
|
def _parse_lines(lines_json: str) -> list[dict[str, Any]]:
|
|
try:
|
|
data = json.loads(lines_json or "[]")
|
|
except (json.JSONDecodeError, TypeError):
|
|
raise HTTPException(status_code=400, detail="라인 데이터 형식 오류")
|
|
if not isinstance(data, list):
|
|
raise HTTPException(status_code=400, detail="라인 데이터는 배열이어야 합니다.")
|
|
return data
|
|
|
|
|
|
def _ym(request: Request) -> tuple[int, int]:
|
|
today = today_kst()
|
|
try:
|
|
year = int(request.query_params.get("year") or today.year)
|
|
month = int(request.query_params.get("month") or today.month)
|
|
except ValueError:
|
|
year, month = today.year, today.month
|
|
if not (1 <= month <= 12):
|
|
year, month = today.year, today.month
|
|
return year, month
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 메인 — 월간 달력 + 선택일 출고 리스트
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/", response_class=HTMLResponse)
|
|
async def index(request: Request) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
|
|
year, month = _ym(request)
|
|
counts = store.calendar_counts(year=year, month=month)
|
|
shipments = store.list_shipments(year=year, month=month)
|
|
|
|
# 선택 날짜 (기본: 오늘이 이번 달이면 오늘, 아니면 1일)
|
|
sel = request.query_params.get("date") or ""
|
|
today = today_kst()
|
|
if not sel:
|
|
sel = today.isoformat() if (today.year == year and today.month == month) else f"{year:04d}-{month:02d}-01"
|
|
|
|
# 선택일에 걸친 묶음(출고일 기준 우선, 작성/입고 포함)
|
|
sel_shipments = [
|
|
s for s in shipments
|
|
if sel in (s.get("ship_date"), s.get("document_date"), s.get("center_arrival_date"))
|
|
]
|
|
# 각 묶음에 품목 요약(제품명/수량) 첨부 — hover 툴팁용
|
|
for s in sel_shipments:
|
|
full = store.get_shipment(shipment_id=s["id"])
|
|
s["tip_items"] = [
|
|
{"name": ln.get("product_name_snapshot") or ln.get("product_code"),
|
|
"qty": ln.get("quantity", 0)}
|
|
for ln in (full.get("lines") if full else [])
|
|
]
|
|
|
|
cal = _calendar.Calendar(firstweekday=6) # 일요일 시작
|
|
weeks = cal.monthdatescalendar(year, month)
|
|
cal_weeks = [
|
|
[
|
|
{
|
|
"date": d.isoformat(),
|
|
"day": d.day,
|
|
"in_month": d.month == month,
|
|
"is_today": d == today,
|
|
"is_selected": d.isoformat() == sel,
|
|
"is_sunday": d.weekday() == 6,
|
|
"is_saturday": d.weekday() == 5,
|
|
"is_holiday": is_holiday(d),
|
|
"counts": counts.get(d.isoformat(), {}),
|
|
}
|
|
for d in week
|
|
]
|
|
for week in weeks
|
|
]
|
|
|
|
prev_y, prev_m = (year - 1, 12) if month == 1 else (year, month - 1)
|
|
next_y, next_m = (year + 1, 1) if month == 12 else (year, month + 1)
|
|
|
|
return render_template(
|
|
request,
|
|
"cupang/index.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 밀크런",
|
|
"page_subtitle": f"{year}년 {month}월 출고 일정",
|
|
"year": year,
|
|
"month": month,
|
|
"prev_y": prev_y, "prev_m": prev_m,
|
|
"next_y": next_y, "next_m": next_m,
|
|
"weekdays": ["일", "월", "화", "수", "목", "금", "토"],
|
|
"cal_weeks": cal_weeks,
|
|
"selected_date": sel,
|
|
"sel_shipments": sel_shipments,
|
|
},
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 출고 묶음 — 등록 / 수정 / 상세
|
|
# ════════════════════════════════════════════════════════════
|
|
def _form_context(request: Request, store: Any, user: dict[str, Any]) -> dict[str, Any]:
|
|
from app.main import build_erp_nav # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
reader = _itemcode(request)
|
|
return {
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"centers": sorted(store.list_centers(), key=lambda c: c["name"]),
|
|
"box_rules": store.list_box_rules(),
|
|
"products": store.list_products(),
|
|
"ship_methods": list(SHIP_METHODS),
|
|
"search_enabled": bool(reader and reader.enabled),
|
|
}
|
|
|
|
|
|
@router.get("/new", response_class=HTMLResponse)
|
|
async def new_form(request: Request) -> HTMLResponse:
|
|
from app.main import render_template # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
ctx = _form_context(request, store, user)
|
|
ctx.update(
|
|
{
|
|
"page_title": "쿠팡 밀크런 — 신규 등록",
|
|
"page_subtitle": "공통 헤더 1개 + 품목 라인",
|
|
"mode": "new",
|
|
"shipment": None,
|
|
"default_date": today_kst().isoformat(),
|
|
}
|
|
)
|
|
return render_template(request, "cupang/form.html", ctx)
|
|
|
|
|
|
@router.post("/new")
|
|
async def create(
|
|
request: Request,
|
|
lines_json: str = Form("[]"),
|
|
document_date: str = Form(...),
|
|
ship_date: str = Form(...),
|
|
center_arrival_date: str = Form(...),
|
|
center_id: str = Form(""),
|
|
center_name_snapshot: str = Form(""),
|
|
ship_method: str = Form("택배"),
|
|
outbound_summary: str = Form(""),
|
|
worker: str = Form(""),
|
|
memo: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
header = {
|
|
"document_date": document_date,
|
|
"ship_date": ship_date,
|
|
"center_arrival_date": center_arrival_date,
|
|
"center_id": center_id,
|
|
"center_name_snapshot": center_name_snapshot,
|
|
"ship_method": ship_method,
|
|
"outbound_summary": outbound_summary,
|
|
"worker": worker,
|
|
"memo": memo,
|
|
}
|
|
try:
|
|
ship = store.create_shipment(
|
|
created_by=user["email"], header=header, lines=_parse_lines(lines_json)
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/cupang/{ship['id']}", status_code=303)
|
|
|
|
|
|
@router.get("/{shipment_id:int}", response_class=HTMLResponse)
|
|
async def detail(request: Request, shipment_id: int) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
ship = store.get_shipment(shipment_id=shipment_id)
|
|
if not ship:
|
|
return render_template(
|
|
request, "denied.html",
|
|
{"reason": "출고 묶음을 찾을 수 없습니다.", "is_admin": is_admin(user)},
|
|
status_code=404,
|
|
)
|
|
return render_template(
|
|
request,
|
|
"cupang/detail.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": f"출고 #{ship['id']}",
|
|
"page_subtitle": f"{ship['ship_date']} · {ship['center_name_snapshot']}",
|
|
"shipment": ship,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/{shipment_id:int}/edit", response_class=HTMLResponse)
|
|
async def edit_form(request: Request, shipment_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
|
|
store, user = guard
|
|
ship = store.get_shipment(shipment_id=shipment_id)
|
|
if not ship:
|
|
return render_template(
|
|
request, "denied.html",
|
|
{"reason": "출고 묶음을 찾을 수 없습니다.", "is_admin": is_admin(user)},
|
|
status_code=404,
|
|
)
|
|
ctx = _form_context(request, store, user)
|
|
ctx.update(
|
|
{
|
|
"page_title": f"출고 #{ship['id']} 수정",
|
|
"page_subtitle": "헤더/라인 수정 후 저장",
|
|
"mode": "edit",
|
|
"shipment": ship,
|
|
"default_date": ship["document_date"],
|
|
}
|
|
)
|
|
return render_template(request, "cupang/form.html", ctx)
|
|
|
|
|
|
@router.post("/{shipment_id:int}/edit")
|
|
async def update(
|
|
request: Request,
|
|
shipment_id: int,
|
|
lines_json: str = Form("[]"),
|
|
document_date: str = Form(...),
|
|
ship_date: str = Form(...),
|
|
center_arrival_date: str = Form(...),
|
|
center_id: str = Form(""),
|
|
center_name_snapshot: str = Form(""),
|
|
ship_method: str = Form("택배"),
|
|
outbound_summary: str = Form(""),
|
|
worker: str = Form(""),
|
|
memo: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
header = {
|
|
"document_date": document_date,
|
|
"ship_date": ship_date,
|
|
"center_arrival_date": center_arrival_date,
|
|
"center_id": center_id,
|
|
"center_name_snapshot": center_name_snapshot,
|
|
"ship_method": ship_method,
|
|
"outbound_summary": outbound_summary,
|
|
"worker": worker,
|
|
"memo": memo,
|
|
}
|
|
try:
|
|
store.update_shipment(
|
|
shipment_id=shipment_id, header=header, lines=_parse_lines(lines_json)
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url=f"/cupang/{shipment_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{shipment_id:int}/delete")
|
|
async def delete(
|
|
request: Request,
|
|
shipment_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
"""운영 안전: 기본은 status='취소' soft delete."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.soft_delete(shipment_id=shipment_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
return RedirectResponse(url=f"/cupang/{shipment_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{shipment_id:int}/hard-delete")
|
|
async def hard_delete(
|
|
request: Request,
|
|
shipment_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
"""완전 삭제(헤더+라인 CASCADE). 달력으로 복귀."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.hard_delete(shipment_id=shipment_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
return RedirectResponse(url="/cupang/", status_code=303)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 입고센터 관리
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/centers", response_class=HTMLResponse)
|
|
async def centers_page(request: Request) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
centers = sorted(store.list_centers(include_inactive=True), key=lambda c: c["name"])
|
|
return render_template(
|
|
request,
|
|
"cupang/centers.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 밀크런 — 입고센터 관리",
|
|
"page_subtitle": "센터명 등록 · 수정 · 활성/비활성 · 삭제.",
|
|
"centers": centers,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/centers")
|
|
async def center_create(
|
|
request: Request,
|
|
name: str = Form(...),
|
|
sort_order: int = Form(0),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.create_center(name=name, sort_order=sort_order)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url="/cupang/centers", status_code=303)
|
|
|
|
|
|
@router.post("/centers/{center_id}/edit")
|
|
async def center_edit(
|
|
request: Request,
|
|
center_id: int,
|
|
name: str = Form(""),
|
|
active: str = Form(""),
|
|
sort_order: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
kwargs: dict[str, Any] = {"center_id": center_id}
|
|
if name.strip():
|
|
kwargs["name"] = name
|
|
if active != "":
|
|
kwargs["active"] = active in ("1", "true", "on", "True")
|
|
if sort_order.strip():
|
|
try:
|
|
kwargs["sort_order"] = int(sort_order)
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
store.update_center(**kwargs)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="센터를 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url="/cupang/centers", status_code=303)
|
|
|
|
|
|
@router.post("/centers/{center_id}/delete")
|
|
async def center_delete(
|
|
request: Request,
|
|
center_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.delete_center(center_id=center_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="센터를 찾을 수 없습니다.")
|
|
return RedirectResponse(url="/cupang/centers", status_code=303)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 박스 입수량 관리
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/box-rules", response_class=HTMLResponse)
|
|
async def box_rules_page(request: Request) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
return render_template(
|
|
request,
|
|
"cupang/box_rules.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 밀크런 — 박스 입수량",
|
|
"page_subtitle": "제품코드별 쿠팡박스 1박스당 입수량 설정",
|
|
"box_rules": store.list_box_rules(include_inactive=True),
|
|
"products": store.list_products(),
|
|
"search_enabled": bool((_itemcode(request)) and _itemcode(request).enabled),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/box-rules")
|
|
async def box_rule_upsert(
|
|
request: Request,
|
|
product_code: str = Form(...),
|
|
units_per_box: int = Form(...),
|
|
product_name_snapshot: str = Form(""),
|
|
box_name: str = Form("쿠팡박스"),
|
|
memo: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.upsert_box_rule(
|
|
product_code=product_code,
|
|
units_per_box=units_per_box,
|
|
product_name_snapshot=product_name_snapshot,
|
|
box_name=box_name,
|
|
memo=memo,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url="/cupang/box-rules", status_code=303)
|
|
|
|
|
|
@router.post("/box-rules/{rule_id}/delete")
|
|
async def box_rule_delete(
|
|
request: Request,
|
|
rule_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.delete_box_rule(rule_id=rule_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="규칙을 찾을 수 없습니다.")
|
|
return RedirectResponse(url="/cupang/box-rules", status_code=303)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 박스 계산기 — 제품명 + 수량 → 박스 수 / 남은 낱개
|
|
# 저장하지 않는 계산 전용 화면. 규칙은 cupang_box_rules 를 그대로 사용한다.
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/box-calc", response_class=HTMLResponse)
|
|
async def box_calc_page(request: Request) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
return render_template(
|
|
request,
|
|
"cupang/box_calc.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 밀크런 — 박스 계산",
|
|
"page_subtitle": "제품명과 수량을 넣으면 박스 수와 남은 낱개를 계산합니다.",
|
|
"box_rules": store.list_box_rules(),
|
|
# 센터 선택 드롭다운은 가나다순 (한글 음절은 코드포인트 순 = 가나다순)
|
|
"centers": sorted(store.list_centers(), key=lambda c: (c.get("name") or "")),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/api/box-calc")
|
|
async def box_calc_api(
|
|
request: Request,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""[{product_code, quantity}] → 제품별 박스 계산 + 박스명별 합계.
|
|
|
|
클라이언트 계산을 신뢰하지 않고 store.compute_boxes 로 서버에서 계산한다.
|
|
"""
|
|
from .store import compute_boxes # noqa: WPS433
|
|
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
|
|
raw_items = payload.get("items")
|
|
if not isinstance(raw_items, list):
|
|
raise HTTPException(status_code=400, detail="items 는 배열이어야 합니다.")
|
|
|
|
rules = {r["product_code"]: r for r in store.list_box_rules()}
|
|
|
|
results: list[dict[str, Any]] = []
|
|
totals: dict[str, dict[str, Any]] = {}
|
|
for raw in raw_items:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
code = str(raw.get("product_code") or "").strip()
|
|
if not code:
|
|
continue
|
|
try:
|
|
qty = int(raw.get("quantity") or 0)
|
|
except (TypeError, ValueError):
|
|
qty = 0
|
|
qty = max(qty, 0)
|
|
|
|
rule = rules.get(code)
|
|
upb = rule["units_per_box"] if rule else None
|
|
calc = compute_boxes(qty, upb)
|
|
box_name = (rule or {}).get("box_name") or ""
|
|
results.append(
|
|
{
|
|
"product_code": code,
|
|
"product_name": (rule or {}).get("product_name_snapshot") or code,
|
|
"box_name": box_name,
|
|
"units_per_box": calc["units_per_box"],
|
|
"quantity": qty,
|
|
"configured": calc["configured"],
|
|
"full_boxes": calc["full_boxes"],
|
|
"remainder_units": calc["remainder_units"],
|
|
"required_boxes": calc["required_boxes"],
|
|
}
|
|
)
|
|
if calc["configured"]:
|
|
agg = totals.setdefault(
|
|
box_name, {"box_name": box_name, "full_boxes": 0, "required_boxes": 0, "remainder_units": 0}
|
|
)
|
|
agg["full_boxes"] += calc["full_boxes"]
|
|
agg["required_boxes"] += calc["required_boxes"]
|
|
agg["remainder_units"] += calc["remainder_units"]
|
|
|
|
mixes = _pack_leftovers(results)
|
|
grand_total = sum(r["full_boxes"] or 0 for r in results if r["configured"])
|
|
grand_total += sum(m["box_count"] for m in mixes)
|
|
|
|
return JSONResponse(
|
|
{
|
|
"results": results,
|
|
"totals": sorted(totals.values(), key=lambda t: t["box_name"]),
|
|
"mixes": mixes,
|
|
"grand_total_boxes": grand_total,
|
|
}
|
|
)
|
|
|
|
|
|
def _pack_leftovers(results: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""제품별 자투리(remainder_units)를 같은 박스명끼리 모아 혼합 박스에 담는다.
|
|
|
|
- 한 박스의 용량을 1 로 두고, 제품 1개가 차지하는 부피를 1/units_per_box 로 본다.
|
|
(예: 쿠팡 2호 = 8개들이 → 1개 = 1/8 박스)
|
|
- 같은 박스명끼리만 섞는다. 서로 다른 입수량이 섞여도 부피 합으로 정확히 계산된다.
|
|
- 한 제품의 자투리가 두 박스에 나뉘어 담기는 것은 허용(그래야 박스 수가 최소).
|
|
- 오차 없이 계산하려고 float 대신 Fraction 을 쓴다.
|
|
"""
|
|
groups: dict[str, list[dict[str, Any]]] = {}
|
|
for r in results:
|
|
if not r["configured"] or not r["remainder_units"]:
|
|
continue
|
|
groups.setdefault(r["box_name"], []).append(r)
|
|
|
|
out: list[dict[str, Any]] = []
|
|
for box_name in sorted(groups):
|
|
# 자투리가 많은 제품부터 담아 박스 안 품목 수를 줄인다.
|
|
items = sorted(groups[box_name], key=lambda r: -r["remainder_units"])
|
|
boxes: list[dict[str, Any]] = []
|
|
cur: list[dict[str, Any]] = []
|
|
free = Fraction(1)
|
|
|
|
for r in items:
|
|
unit = Fraction(1, int(r["units_per_box"]))
|
|
left = int(r["remainder_units"])
|
|
while left > 0:
|
|
take = min(left, int(free / unit))
|
|
if take == 0: # 남은 자리 없음 → 새 박스
|
|
boxes.append({"items": cur, "fill_percent": float(round((1 - free) * 100, 1))})
|
|
cur, free = [], Fraction(1)
|
|
continue
|
|
cur.append(
|
|
{
|
|
"product_code": r["product_code"],
|
|
"product_name": r["product_name"],
|
|
"quantity": take,
|
|
}
|
|
)
|
|
free -= unit * take
|
|
left -= take
|
|
if cur:
|
|
boxes.append({"items": cur, "fill_percent": float(round((1 - free) * 100, 1))})
|
|
|
|
out.append(
|
|
{
|
|
"box_name": box_name,
|
|
"box_count": len(boxes),
|
|
"leftover_units": sum(int(r["remainder_units"]) for r in items),
|
|
"boxes": boxes,
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 설정 — 제품명 카탈로그 관리 (itemcode_db 에서 등록)
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/products", response_class=HTMLResponse)
|
|
async def products_page(request: Request) -> HTMLResponse:
|
|
from app.main import build_erp_nav, render_template # noqa: WPS433
|
|
from app.store import is_admin # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, user = guard
|
|
reader = _itemcode(request)
|
|
return render_template(
|
|
request,
|
|
"cupang/products.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 밀크런 — 설정 (제품명)",
|
|
"page_subtitle": "왼쪽 itemcode_db 목록에서 선택해 등록하면 폼 드롭다운에 노출됩니다.",
|
|
"products": store.list_products(include_inactive=True),
|
|
"registered_codes": [p["product_code"] for p in store.list_products(include_inactive=True)],
|
|
"search_enabled": bool(reader and reader.enabled),
|
|
"search_reason": (reader.reason if reader else ""),
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/api/products/all")
|
|
async def product_all(
|
|
request: Request, _: dict[str, Any] = Depends(_require_user)
|
|
) -> JSONResponse:
|
|
"""itemcode_db 전체 상품 목록(낱개+세트). 설정 화면 왼쪽 리스트 소스."""
|
|
reader = _itemcode(request)
|
|
results = reader.list_all() if reader else []
|
|
return JSONResponse(
|
|
{
|
|
"enabled": bool(reader and reader.enabled),
|
|
"reason": (reader.reason if reader else "itemcode 리더 미초기화"),
|
|
"error": (getattr(reader, "last_error", "") if reader else ""),
|
|
"count": len(results),
|
|
"results": results,
|
|
}
|
|
)
|
|
|
|
|
|
@router.post("/products/bulk")
|
|
async def product_bulk(
|
|
request: Request,
|
|
items: list[dict[str, Any]] = Body(..., embed=True),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""선택한 상품들을 일괄 등록(upsert). body: {"items":[{"code","name"}, ...]}."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
added = 0
|
|
for it in items:
|
|
code = str(it.get("code") or "").strip()
|
|
name = str(it.get("name") or "").strip()
|
|
if not code or not name:
|
|
continue
|
|
try:
|
|
store.upsert_product(product_code=code, product_name=name)
|
|
added += 1
|
|
except ValueError:
|
|
continue
|
|
return JSONResponse({"ok": True, "added": added})
|
|
|
|
|
|
@router.post("/products")
|
|
async def product_upsert(
|
|
request: Request,
|
|
product_code: str = Form(...),
|
|
product_name: str = Form(...),
|
|
sort_order: int = Form(0),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.upsert_product(
|
|
product_code=product_code, product_name=product_name, sort_order=sort_order
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return RedirectResponse(url="/cupang/products", status_code=303)
|
|
|
|
|
|
@router.post("/products/{product_id:int}/active")
|
|
async def product_set_active(
|
|
request: Request,
|
|
product_id: int,
|
|
active: str = Form(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.set_product_active(
|
|
product_id=product_id, active=active in ("1", "true", "on", "True")
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="제품을 찾을 수 없습니다.")
|
|
return RedirectResponse(url="/cupang/products", status_code=303)
|
|
|
|
|
|
@router.post("/products/{product_id:int}/delete")
|
|
async def product_delete(
|
|
request: Request,
|
|
product_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
"""완전 삭제(hard delete)."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.delete_product(product_id=product_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="제품을 찾을 수 없습니다.")
|
|
return RedirectResponse(url="/cupang/products", status_code=303)
|
|
|
|
|
|
@router.get("/health")
|
|
async def health() -> dict[str, str]:
|
|
return {"status": "ok", "module": "cupang"}
|