king
7a01c140b0
출고 묶음을 만드는 길이 둘(신규 등록 폼 / 박스 계산 분배 확정)이라 헷갈렸다. 폼 경로를 없애고 [+ 신규 등록] 이 박스 계산을 열게 한다. 중복이던 오른쪽 [박스 계산] 버튼은 제거. /cupang/new 는 북마크가 깨지지 않게 박스 계산으로 리다이렉트하고 POST /new(생성)는 삭제했다. 기존 묶음 수정은 그대로 동작한다. 달력 칸은 작성/출고/입고 세 배지를 늘어놓아 읽기 어려웠다. 출고 기준으로 "출고 N건 · 센터 M곳"만 남긴다. 오른쪽 상세도 같은 기준(출고일)으로 맞추고 센터별 출고방식·총 박스/개수와 상품별 수량·박스를 카드 안에 펼친다. 품목이 바로 보이므로 hover 툴팁은 제거. 임시 저장 기본 제목은 "2026.08.31(월) 오후 04시 50분" 형식으로.
1085 lines
42 KiB
Python
1085 lines
42 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
|
|
import re
|
|
from fractions import Fraction
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from datetime import date as _date
|
|
|
|
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)
|
|
shipments = store.list_shipments(year=year, month=month)
|
|
|
|
# 달력 칸에는 출고 건수와 센터 수만 보여준다.
|
|
counts: dict[str, dict[str, Any]] = {}
|
|
for s in shipments:
|
|
d = s.get("ship_date")
|
|
if not d:
|
|
continue
|
|
cell = counts.setdefault(str(d), {"ship": 0, "centers": set()})
|
|
cell["ship"] += 1
|
|
cell["centers"].add(s.get("center_id") or s.get("center_name_snapshot") or "")
|
|
for cell in counts.values():
|
|
cell["centers"] = len(cell["centers"])
|
|
|
|
# 선택 날짜 (기본: 오늘이 이번 달이면 오늘, 아니면 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 s.get("ship_date") == sel]
|
|
|
|
# 오른쪽 상세: 센터별 상품 목록·수량·박스 수·출고방식
|
|
for s in sel_shipments:
|
|
full = store.get_shipment(shipment_id=s["id"])
|
|
lines = (full.get("lines") if full else []) or []
|
|
s["items"] = [
|
|
{
|
|
"name": ln.get("product_name_snapshot") or ln.get("product_code"),
|
|
"qty": int(ln.get("quantity") or 0),
|
|
"boxes": int(ln.get("calculated_boxes") or 0),
|
|
}
|
|
for ln in lines
|
|
]
|
|
s["total_qty"] = sum(it["qty"] for it in s["items"])
|
|
s["total_boxes"] = sum(it["boxes"] for it in s["items"])
|
|
|
|
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")
|
|
async def new_form(request: Request) -> RedirectResponse:
|
|
"""신규 등록 폼은 없앴다. 출고 묶음은 박스 계산의 [분배 확정] 으로만 만든다.
|
|
|
|
예전 링크/북마크가 404 나지 않도록 박스 계산으로 보낸다.
|
|
"""
|
|
return RedirectResponse(url="/cupang/box-calc", 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)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 입고센터 관리
|
|
# ════════════════════════════════════════════════════════════
|
|
_CENTER_PREFIX_RE = re.compile(r"^\D*")
|
|
|
|
|
|
def _center_group(name: str) -> str:
|
|
"""센터명 앞부분(숫자 전까지)을 지역 그룹 키로 쓴다. 예) 인천14 → 인천."""
|
|
name = (name or "").strip()
|
|
prefix = _CENTER_PREFIX_RE.match(name).group(0).strip()
|
|
return prefix or name or "기타"
|
|
|
|
|
|
def _center_sort_key(name: str) -> list[Any]:
|
|
"""숫자를 숫자로 비교하는 자연 정렬. 예) 인천4 < 인천14."""
|
|
parts = re.split(r"(\d+)", (name or "").strip())
|
|
return [(1, int(p), "") if p.isdigit() else (0, 0, p.lower()) for p in parts]
|
|
|
|
|
|
def _group_centers(centers: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
buckets: dict[str, list[dict[str, Any]]] = {}
|
|
for c in centers:
|
|
buckets.setdefault(_center_group(c["name"]), []).append(c)
|
|
groups = []
|
|
for key, items in buckets.items():
|
|
items.sort(key=lambda c: _center_sort_key(c["name"]))
|
|
groups.append({"name": key, "items": items, "count": len(items)})
|
|
groups.sort(key=lambda g: _center_sort_key(g["name"]))
|
|
return groups
|
|
|
|
|
|
@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: _center_sort_key(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,
|
|
"center_groups": _group_centers(centers),
|
|
"flash": request.query_params.get("msg", ""),
|
|
"flash_name": request.query_params.get("name", ""),
|
|
},
|
|
)
|
|
|
|
|
|
@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 미설정")
|
|
name = (name or "").strip()
|
|
if not name:
|
|
return RedirectResponse(url="/cupang/centers", status_code=303)
|
|
|
|
# 같은 이름이 이미 있으면 새로 만들지 않고 알림만 돌려준다.
|
|
existing = next(
|
|
(
|
|
c
|
|
for c in store.list_centers(include_inactive=True)
|
|
if (c.get("name") or "").strip().lower() == name.lower()
|
|
),
|
|
None,
|
|
)
|
|
if existing is not None:
|
|
msg = "dup" if existing.get("active") else "dup_inactive"
|
|
return RedirectResponse(
|
|
url=f"/cupang/centers?msg={msg}&name={quote(name)}", status_code=303
|
|
)
|
|
|
|
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=f"/cupang/centers?msg=added&name={quote(name)}", status_code=303
|
|
)
|
|
|
|
|
|
@router.post("/centers/{center_id}/edit")
|
|
async def center_edit(
|
|
request: Request,
|
|
center_id: int,
|
|
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 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.post("/api/box-calc/confirm")
|
|
async def box_calc_confirm(
|
|
request: Request,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""{ship_date, centers:[{center_id, ship_method, boxes, items:[...]}]} → 센터마다 출고 묶음 1건 생성.
|
|
|
|
화면에서 보낸 박스 수는 요약 표시용이고, 라인의 박스 수는 저장 시
|
|
서버(store.compute_boxes)가 수량과 입수량으로 다시 계산한다.
|
|
"""
|
|
from .store import SHIP_METHODS as _METHODS # noqa: WPS433
|
|
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
|
|
ship_date = str(payload.get("ship_date") or "").strip()
|
|
try:
|
|
_date.fromisoformat(ship_date)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="출고일자를 올바르게 선택하세요.")
|
|
|
|
raw_centers = payload.get("centers")
|
|
if not isinstance(raw_centers, list) or not raw_centers:
|
|
raise HTTPException(status_code=400, detail="확정할 센터가 없습니다.")
|
|
|
|
rules = {r["product_code"]: r for r in store.list_box_rules()}
|
|
known_centers = {str(c["id"]): c for c in store.list_centers(include_inactive=True)}
|
|
today = today_kst().isoformat()
|
|
worker = str(user.get("name") or user.get("email") or "")
|
|
|
|
plans: list[dict[str, Any]] = []
|
|
for raw in raw_centers:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
cid = str(raw.get("center_id") or "").strip()
|
|
center = known_centers.get(cid)
|
|
if center is None:
|
|
raise HTTPException(status_code=400, detail=f"알 수 없는 센터입니다: {cid}")
|
|
|
|
method = str(raw.get("ship_method") or "").strip()
|
|
if method not in _METHODS:
|
|
raise HTTPException(
|
|
status_code=400, detail=f"{center['name']} 의 출고방식을 선택하세요."
|
|
)
|
|
|
|
# 같은 제품이 여러 박스로 나뉘어 담겼을 수 있으므로 제품코드로 합친다.
|
|
merged: dict[str, int] = {}
|
|
for it in raw.get("items") or []:
|
|
if not isinstance(it, dict):
|
|
continue
|
|
code = str(it.get("product_code") or "").strip()
|
|
if not code:
|
|
continue
|
|
try:
|
|
qty = int(it.get("quantity") or 0)
|
|
except (TypeError, ValueError):
|
|
qty = 0
|
|
if qty <= 0:
|
|
continue
|
|
merged[code] = merged.get(code, 0) + qty
|
|
|
|
if not merged:
|
|
continue
|
|
|
|
lines = []
|
|
for code, qty in merged.items():
|
|
rule = rules.get(code)
|
|
lines.append(
|
|
{
|
|
"product_code": code,
|
|
"product_name_snapshot": (rule or {}).get("product_name_snapshot") or code,
|
|
"quantity": qty,
|
|
"units_per_box": (rule or {}).get("units_per_box"),
|
|
"box_rule_id": (rule or {}).get("id"),
|
|
}
|
|
)
|
|
|
|
try:
|
|
boxes = max(int(raw.get("boxes") or 0), 0)
|
|
except (TypeError, ValueError):
|
|
boxes = 0
|
|
pieces = sum(merged.values())
|
|
summary = f"{boxes}박스 · {pieces}개" if boxes else f"{pieces}개"
|
|
|
|
plans.append({"center": center, "method": method, "lines": lines, "summary": summary})
|
|
|
|
if not plans:
|
|
raise HTTPException(status_code=400, detail="담긴 품목이 없습니다.")
|
|
|
|
created: list[dict[str, Any]] = []
|
|
for plan in plans:
|
|
center = plan["center"]
|
|
try:
|
|
ship = store.create_shipment(
|
|
created_by=str(user.get("email") or ""),
|
|
header={
|
|
"document_date": today,
|
|
"ship_date": ship_date,
|
|
# 센터입고일은 출고일과 같게 두고, 필요하면 출고 상세에서 고친다.
|
|
"center_arrival_date": ship_date,
|
|
"center_id": center["id"],
|
|
"center_name_snapshot": center["name"],
|
|
"ship_method": plan["method"],
|
|
"outbound_summary": plan["summary"],
|
|
"worker": worker,
|
|
"status": "출고준비",
|
|
"memo": "박스 계산에서 분배 확정",
|
|
},
|
|
lines=plan["lines"],
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
created.append({"id": ship["id"], "center_name": center["name"]})
|
|
|
|
return JSONResponse({"created": created, "ship_date": ship_date})
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 박스 계산 임시 저장 (화면 상태 스냅샷)
|
|
# ════════════════════════════════════════════════════════════
|
|
@router.get("/api/box-calc/drafts")
|
|
async def box_calc_drafts_list(
|
|
request: Request,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
return JSONResponse({"drafts": store.list_box_calc_drafts()})
|
|
|
|
|
|
@router.get("/api/box-calc/drafts/{draft_id:int}")
|
|
async def box_calc_draft_get(
|
|
request: Request,
|
|
draft_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
draft = store.get_box_calc_draft(draft_id=draft_id)
|
|
if draft is None:
|
|
raise HTTPException(status_code=404, detail="저장된 내용을 찾을 수 없습니다.")
|
|
return JSONResponse({"draft": draft})
|
|
|
|
|
|
@router.post("/api/box-calc/drafts")
|
|
async def box_calc_draft_save(
|
|
request: Request,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""{title, payload, draft_id?} → 저장(같은 제목이면 덮어쓰기)."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
|
|
title = str(payload.get("title") or "").strip()
|
|
if not title:
|
|
raise HTTPException(status_code=400, detail="제목을 입력하세요.")
|
|
if len(title) > 100:
|
|
title = title[:100]
|
|
snapshot = payload.get("payload")
|
|
if not isinstance(snapshot, dict):
|
|
raise HTTPException(status_code=400, detail="payload 는 객체여야 합니다.")
|
|
|
|
raw_id = payload.get("draft_id")
|
|
draft_id = int(raw_id) if isinstance(raw_id, int) or (isinstance(raw_id, str) and raw_id.isdigit()) else None
|
|
|
|
try:
|
|
draft = store.save_box_calc_draft(
|
|
title=title,
|
|
payload=snapshot,
|
|
created_by=str(user.get("name") or user.get("email") or ""),
|
|
draft_id=draft_id,
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="저장된 내용을 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"draft": draft})
|
|
|
|
|
|
@router.delete("/api/box-calc/drafts/{draft_id:int}")
|
|
async def box_calc_draft_delete(
|
|
request: Request,
|
|
draft_id: int,
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
store.delete_box_calc_draft(draft_id=draft_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="저장된 내용을 찾을 수 없습니다.")
|
|
return JSONResponse({"deleted": True})
|
|
|
|
|
|
@router.get("/health")
|
|
async def health() -> dict[str, str]:
|
|
return {"status": "ok", "module": "cupang"}
|