king
4fce96b438
발주서에 있는 센터가 cupang_centers 에 없으면 자동 배분에서 제외됐다. 이제 발주서에 적힌 이름 그대로 센터를 등록(sort_order 는 기존 최대값 다음) 하고 계산에 포함한다. 경고문도 "등록되지 않은 센터" 대신 "새 입고센터 자동 등록" 으로 바꿨다. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1789 lines
69 KiB
Python
1789 lines
69 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 datetime import date as _date, timedelta as _timedelta
|
|
|
|
from app.timezone import today_kst
|
|
|
|
from fastapi import APIRouter, Body, Depends, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
|
|
from .holidays import holiday_name, 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 _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)
|
|
next_y, next_m = (year + 1, 1) if month == 12 else (year, month + 1)
|
|
prev_y, prev_m = (year - 1, 12) if month == 1 else (year, month - 1)
|
|
|
|
# 달력은 현재 달 + 다음 달 2개월을 함께 보여준다.
|
|
# 취소된 묶음은 달력·목록 어디에도 보이지 않는다(soft delete = 삭제로 취급).
|
|
merged: dict[Any, dict[str, Any]] = {}
|
|
for y, m in ((year, month), (next_y, next_m)):
|
|
for s in store.list_shipments(year=y, month=m):
|
|
if s.get("status") == "취소":
|
|
continue
|
|
merged[s["id"]] = s
|
|
shipments = list(merged.values())
|
|
|
|
# 달력 칸에는 출고 건수와 센터 수만 보여준다.
|
|
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:
|
|
in_view = (today.year, today.month) in ((year, month), (next_y, next_m))
|
|
sel = today.isoformat() if in_view 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) # 일요일 시작
|
|
|
|
def _build_month(y: int, m: int) -> dict[str, Any]:
|
|
weeks = [
|
|
[
|
|
{
|
|
"date": d.isoformat(),
|
|
"day": d.day,
|
|
"in_month": d.month == m,
|
|
"is_today": d == today,
|
|
"is_selected": d.isoformat() == sel,
|
|
"is_sunday": d.weekday() == 6,
|
|
"is_saturday": d.weekday() == 5,
|
|
"is_holiday": is_holiday(d),
|
|
"holiday_name": holiday_name(d),
|
|
"counts": counts.get(d.isoformat(), {}),
|
|
}
|
|
for d in week
|
|
]
|
|
for week in cal.monthdatescalendar(y, m)
|
|
]
|
|
ship_total = sum(
|
|
int(c.get("ship") or 0)
|
|
for day, c in counts.items()
|
|
if day[:7] == f"{y:04d}-{m:02d}"
|
|
)
|
|
return {
|
|
"year": y,
|
|
"month": m,
|
|
"label": f"{y}년 {m}월",
|
|
"weeks": weeks,
|
|
"ship_total": ship_total,
|
|
}
|
|
|
|
months = [_build_month(year, month), _build_month(next_y, next_m)]
|
|
|
|
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}월 · {next_y}년 {next_m}월 출고 일정",
|
|
"year": year,
|
|
"month": month,
|
|
"prev_y": prev_y, "prev_m": prev_m,
|
|
"next_y": next_y, "next_m": next_m,
|
|
"today_iso": today.isoformat(),
|
|
"weekdays": ["일", "월", "화", "수", "목", "금", "토"],
|
|
"months": months,
|
|
"selected_date": sel,
|
|
"sel_shipments": sel_shipments,
|
|
},
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 출고리스트 엑셀 — 확정한 출고일 기준 (쿠팡로켓 밀크런 양식)
|
|
# ════════════════════════════════════════════════════════════
|
|
def _shipments_for_date(store: Any, ship_date: str) -> list[dict[str, Any]]:
|
|
"""해당 출고일의 출고 묶음(라인 포함) — 취소 제외, 센터명 순."""
|
|
heads = [
|
|
h for h in store.list_shipments(date_from=ship_date, date_to=ship_date)
|
|
if h.get("status") != "취소"
|
|
]
|
|
out: list[dict[str, Any]] = []
|
|
for h in sorted(heads, key=lambda x: (x.get("center_name_snapshot") or "")):
|
|
full = store.get_shipment(shipment_id=h["id"])
|
|
if full:
|
|
out.append(full)
|
|
return out
|
|
|
|
|
|
def _sheet_target(store: Any) -> tuple[Any, str, dict[str, Any] | None]:
|
|
"""(writer, spreadsheet_id, skip 사유) — 설정이 없으면 skip 정보를 돌려준다."""
|
|
import os # noqa: WPS433
|
|
|
|
from app.integrations.google_sheets import get_writer # noqa: WPS433
|
|
|
|
spreadsheet_id = (os.getenv("CUPANG_SHEET_ID") or "").strip()
|
|
if not spreadsheet_id:
|
|
return None, "", {"ok": False, "skipped": True, "reason": "CUPANG_SHEET_ID 미설정"}
|
|
writer = get_writer()
|
|
if not writer.enabled:
|
|
return None, "", {"ok": False, "skipped": True, "reason": writer.reason}
|
|
return writer, spreadsheet_id, None
|
|
|
|
|
|
def _sync_google_sheet_after_delete(store: Any, ship_date: str) -> dict[str, Any]:
|
|
"""출고 삭제 후 구글 시트를 현재 상태에 맞춘다.
|
|
|
|
- 그날 남은 출고가 없으면 해당 날짜 시트를 삭제
|
|
- 남아 있으면 남은 내용으로 다시 기록
|
|
실패해도 DB 삭제는 이미 끝났으므로 예외를 밖으로 던지지 않는다.
|
|
"""
|
|
from .export import sheet_title # noqa: WPS433
|
|
|
|
writer, spreadsheet_id, skip = _sheet_target(store)
|
|
if skip:
|
|
return skip
|
|
|
|
if _shipments_for_date(store, ship_date):
|
|
return _push_to_google_sheet(store, ship_date)
|
|
|
|
try:
|
|
res = writer.delete_sheet(spreadsheet_id=spreadsheet_id, title=sheet_title(ship_date))
|
|
except Exception as exc: # noqa: BLE001 - 사유만 남긴다
|
|
return {"ok": False, "skipped": False, "reason": f"{type(exc).__name__}: {exc}"}
|
|
return {"ok": True, "skipped": False, **res}
|
|
|
|
|
|
def _log_sheet(action: str, ship_date: str, res: dict[str, Any]) -> None:
|
|
print( # noqa: T201 - 컨테이너 로그 확인용(비밀값 없음)
|
|
f"[cupang] sheet {action} date={ship_date} ok={res.get('ok')} "
|
|
f"skipped={res.get('skipped')} detail={res.get('action', '')} "
|
|
f"reason={res.get('reason', '')}",
|
|
flush=True,
|
|
)
|
|
|
|
|
|
def _push_to_google_sheet(store: Any, ship_date: str) -> dict[str, Any]:
|
|
"""확정한 출고일을 Google 스프레드시트에 시트 1장으로 기록.
|
|
|
|
- 대상 스프레드시트: 환경변수 `CUPANG_SHEET_ID`
|
|
- 인증: 서비스 계정(app/integrations/google_sheets.py). 미설정이면 조용히 skip.
|
|
- 실패해도 출고 묶음 저장은 이미 끝났으므로 예외를 밖으로 던지지 않는다.
|
|
"""
|
|
from .export import ( # noqa: WPS433
|
|
COL_FIRST, COL_LAST, FIRST_DATA_ROW, HEADER_BG, HEADER_ROW, PALLET_BG,
|
|
TITLE_ROW, WIDTHS_PX, build_table, sheet_title,
|
|
)
|
|
|
|
writer, spreadsheet_id, skip = _sheet_target(store)
|
|
if skip:
|
|
return skip
|
|
|
|
shipments = _shipments_for_date(store, ship_date)
|
|
if not shipments:
|
|
return {"ok": False, "skipped": True, "reason": "해당 출고일의 출고 묶음 없음"}
|
|
|
|
table = build_table(ship_date, shipments)
|
|
try:
|
|
res = writer.write_table(
|
|
spreadsheet_id=spreadsheet_id,
|
|
title=sheet_title(ship_date),
|
|
cells=table["cells"],
|
|
merges=table["merges"],
|
|
header_row=HEADER_ROW,
|
|
first_data_row=FIRST_DATA_ROW,
|
|
last_row=table["last_row"],
|
|
first_col=COL_FIRST,
|
|
last_col=COL_LAST,
|
|
title_row=TITLE_ROW,
|
|
widths=WIDTHS_PX,
|
|
header_bg=HEADER_BG,
|
|
row_highlights=[(r1, r2, PALLET_BG) for (r1, r2) in table["pallet_ranges"]],
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - 시트 기록 실패는 경고로만 알린다
|
|
return {"ok": False, "skipped": False, "reason": f"{type(exc).__name__}: {exc}"}
|
|
return {"ok": True, "skipped": False, **res}
|
|
|
|
|
|
@router.get("/export.xlsx")
|
|
async def export_shipments_xlsx(request: Request, date: str = "") -> Any:
|
|
"""출고일 기준 출고리스트 엑셀 다운로드. 시트명 = YYYYMMDD."""
|
|
from io import BytesIO # noqa: WPS433
|
|
|
|
from fastapi.responses import StreamingResponse # noqa: WPS433
|
|
|
|
from .export import build_workbook # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, _user = guard
|
|
|
|
ship_date = (date or "").strip()
|
|
try:
|
|
_date.fromisoformat(ship_date)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="출고일자(date=YYYY-MM-DD)가 필요합니다.")
|
|
|
|
shipments = _shipments_for_date(store, ship_date)
|
|
if not shipments:
|
|
raise HTTPException(status_code=404, detail="해당 출고일의 출고 묶음이 없습니다.")
|
|
|
|
wb = build_workbook(ship_date, shipments)
|
|
buf = BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
filename = f"{ship_date.replace('-', '')}_cupang_milkrun.xlsx"
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 출고 묶음 — 등록 / 수정 / 상세
|
|
# ════════════════════════════════════════════════════════════
|
|
@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:
|
|
"""출고 묶음 보기 — 상자 계산 화면과 같은 3열 구성(읽기 전용).
|
|
|
|
수정 기능은 없앴다. 잘못 만들었으면 달력에서 삭제하고 다시 확정한다.
|
|
"""
|
|
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,
|
|
)
|
|
|
|
lines = ship.get("lines") or []
|
|
items = [
|
|
{
|
|
"product_code": ln.get("product_code"),
|
|
"product_name": ln.get("product_name_snapshot") or ln.get("product_code"),
|
|
"quantity": int(ln.get("quantity") or 0),
|
|
}
|
|
for ln in lines
|
|
]
|
|
calc = _compute_boxes(store, items)
|
|
total_qty = sum(it["quantity"] for it in items)
|
|
|
|
return render_template(
|
|
request,
|
|
"cupang/view.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,
|
|
"items": items,
|
|
"calc": calc,
|
|
"box_list": _expand_box_list(ship, calc, items),
|
|
"total_qty": total_qty,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/{shipment_id:int}/boxes.xlsx")
|
|
async def export_box_list_xlsx(request: Request, shipment_id: int) -> Any:
|
|
"""상자 목록(상자번호·제품명·제품코드·수량) 엑셀 다운로드."""
|
|
from io import BytesIO # noqa: WPS433
|
|
|
|
from fastapi.responses import StreamingResponse # noqa: WPS433
|
|
|
|
from .export import build_box_list_workbook # 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:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
|
|
items = [
|
|
{
|
|
"product_code": ln.get("product_code"),
|
|
"product_name": ln.get("product_name_snapshot") or ln.get("product_code"),
|
|
"quantity": int(ln.get("quantity") or 0),
|
|
}
|
|
for ln in (ship.get("lines") or [])
|
|
]
|
|
box_list = _expand_box_list(ship, _compute_boxes(store, items), items)
|
|
|
|
wb = build_box_list_workbook(ship, box_list)
|
|
buf = BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
stamp = str(ship.get("ship_date") or "").replace("-", "")
|
|
filename = f"{stamp}_cupang_boxes_{shipment_id}.xlsx"
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
|
)
|
|
|
|
|
|
def _safe_next(raw: str, fallback: str) -> str:
|
|
"""열린 리다이렉트 방지 — /cupang/ 안쪽 경로만 허용."""
|
|
nxt = (raw or "").strip()
|
|
if nxt.startswith("/cupang/") and "//" not in nxt[1:]:
|
|
return nxt
|
|
return fallback
|
|
|
|
|
|
@router.post("/{shipment_id:int}/delete")
|
|
async def delete(
|
|
request: Request,
|
|
shipment_id: int,
|
|
next: str = Form(""),
|
|
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 미설정")
|
|
ship = store.get_shipment(shipment_id=shipment_id, with_lines=False)
|
|
ship_date = str((ship or {}).get("ship_date") or "")
|
|
try:
|
|
store.soft_delete(shipment_id=shipment_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
if ship_date:
|
|
_log_sheet("sync", ship_date, _sync_google_sheet_after_delete(store, ship_date))
|
|
return RedirectResponse(
|
|
url=_safe_next(next, f"/cupang/?date={ship_date}" if ship_date else "/cupang/"),
|
|
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 미설정")
|
|
ship = store.get_shipment(shipment_id=shipment_id, with_lines=False)
|
|
ship_date = str((ship or {}).get("ship_date") or "")
|
|
try:
|
|
store.hard_delete(shipment_id=shipment_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
|
if ship_date:
|
|
_log_sheet("sync", ship_date, _sync_google_sheet_after_delete(store, ship_date))
|
|
return RedirectResponse(url="/cupang/", status_code=303)
|
|
|
|
|
|
@router.post("/day-delete")
|
|
async def day_delete(
|
|
request: Request,
|
|
date: str = Form(...),
|
|
next: str = Form(""),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> RedirectResponse:
|
|
"""선택한 출고일의 묶음을 한 번에 취소 처리한다(soft delete)."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
|
|
day = (date or "").strip()
|
|
try:
|
|
_date.fromisoformat(day)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="날짜 형식이 올바르지 않습니다.")
|
|
|
|
for ship in store.list_shipments(date_from=day, date_to=day):
|
|
if ship.get("status") == "취소":
|
|
continue
|
|
try:
|
|
store.soft_delete(shipment_id=ship["id"])
|
|
except KeyError:
|
|
continue
|
|
# 그날 출고가 모두 사라졌으면 구글 시트의 해당 날짜 시트도 지운다.
|
|
_log_sheet("sync", day, _sync_google_sheet_after_delete(store, day))
|
|
return RedirectResponse(
|
|
url=_safe_next(next, f"/cupang/?date={day}"), status_code=303
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 쿠팡 로켓 매출 (cupang_sales)
|
|
# ════════════════════════════════════════════════════════════
|
|
def _sales_totals(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
|
"""조회 결과 합계 — 화면 KPI 와 엑셀 하단에 같이 쓴다."""
|
|
total = {
|
|
"count": len(rows),
|
|
"quantity": 0,
|
|
"supply_amount": 0.0,
|
|
"cost_amount": 0.0,
|
|
"logistics_total": 0.0,
|
|
"margin": 0.0,
|
|
}
|
|
for r in rows:
|
|
total["quantity"] += int(r.get("quantity") or 0)
|
|
for key in ("supply_amount", "cost_amount", "logistics_total", "margin"):
|
|
total[key] += float(r.get(key) or 0)
|
|
supply = total["supply_amount"]
|
|
total["margin_rate"] = round(total["margin"] / supply * 100, 2) if supply else 0.0
|
|
total["logistics_ratio"] = (
|
|
round(total["logistics_total"] / supply * 100, 2) if supply else 0.0
|
|
)
|
|
return total
|
|
|
|
|
|
def _sales_filters(request: Request) -> dict[str, str]:
|
|
q = request.query_params
|
|
return {
|
|
"date_from": (q.get("from") or "").strip(),
|
|
"date_to": (q.get("to") or "").strip(),
|
|
"center_name": (q.get("center") or "").strip(),
|
|
"keyword": (q.get("q") or "").strip(),
|
|
"po_type": (q.get("type") or "").strip(),
|
|
"week_label": (q.get("week") or "").strip(),
|
|
}
|
|
|
|
|
|
@router.get("/sales", response_class=HTMLResponse)
|
|
async def sales_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
|
|
|
|
filters = _sales_filters(request)
|
|
rows = store.list_sales(**filters)
|
|
weekly = {w["week_label"]: w for w in store.list_sales_weekly()}
|
|
|
|
# 주차별 광고비/할인/장려금은 조회에 걸린 주차만 더한다.
|
|
weeks = {r.get("week_label") for r in rows if r.get("week_label")}
|
|
extra = {"ad_cost": 0.0, "promo_discount": 0.0, "incentive": 0.0}
|
|
for label in weeks:
|
|
w = weekly.get(label)
|
|
if not w:
|
|
continue
|
|
for key in extra:
|
|
extra[key] += float(w.get(key) or 0)
|
|
|
|
# 줄무늬 — 기본은 입고센터가 바뀔 때마다, 센터로 검색 중이면 출고일이 바뀔 때마다.
|
|
band_key = "ship_date" if filters["center_name"] else "center_name"
|
|
band = 0
|
|
prev = None
|
|
for row in rows:
|
|
value = row.get(band_key)
|
|
if prev is not None and value != prev:
|
|
band ^= 1
|
|
prev = value
|
|
row["band"] = band
|
|
|
|
totals = _sales_totals(rows)
|
|
totals.update(extra)
|
|
totals["net_margin"] = (
|
|
totals["margin"] - extra["ad_cost"] - extra["promo_discount"] + extra["incentive"]
|
|
)
|
|
totals["net_margin_rate"] = (
|
|
round(totals["net_margin"] / totals["supply_amount"] * 100, 2)
|
|
if totals["supply_amount"] else 0.0
|
|
)
|
|
|
|
return render_template(
|
|
request,
|
|
"cupang/sales.html",
|
|
{
|
|
"user": user,
|
|
"is_admin": is_admin(user),
|
|
"nav_items": build_erp_nav(user, active="cupang"),
|
|
"page_title": "쿠팡 로켓 매출",
|
|
"page_subtitle": "발주 라인별 공급가·원가·물류비·마진",
|
|
"rows": rows,
|
|
"totals": totals,
|
|
"filters": filters,
|
|
"centers": store.sales_centers(),
|
|
"weeks": store.sales_weeks(),
|
|
"po_types": ["일반", "벤더플렉스"],
|
|
"weekly": sorted(weekly.values(), key=lambda w: w.get("week_from") or ""),
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/sales/api")
|
|
async def sales_create(
|
|
request: Request,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
row = store.create_sale(data=payload)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"ok": True, "row": row})
|
|
|
|
|
|
@router.post("/sales/api/{sale_id:int}")
|
|
async def sales_update(
|
|
request: Request,
|
|
sale_id: int,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
row = store.update_sale(sale_id=sale_id, data=payload)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="매출 행을 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"ok": True, "row": row})
|
|
|
|
|
|
@router.post("/sales/api/{sale_id:int}/delete")
|
|
async def sales_delete(
|
|
request: Request,
|
|
sale_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_sale(sale_id=sale_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="매출 행을 찾을 수 없습니다.")
|
|
return JSONResponse({"ok": True})
|
|
|
|
|
|
@router.post("/sales/upload")
|
|
async def sales_upload(
|
|
request: Request,
|
|
files: list[UploadFile] = File(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""구글 시트 양식 그대로의 xlsx/csv 를 올려 일괄 등록(같은 라인은 덮어씀)."""
|
|
from .sales_import import parse_upload # noqa: WPS433
|
|
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
|
|
saved = 0
|
|
weekly_saved = 0
|
|
names: list[str] = []
|
|
for up in files:
|
|
data = await up.read()
|
|
if not data:
|
|
continue
|
|
try:
|
|
parsed = parse_upload(up.filename or "", data)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=f"{up.filename}: {exc}")
|
|
except Exception as exc: # noqa: BLE001 - 파일 형식 문제를 사용자에게 알린다
|
|
raise HTTPException(
|
|
status_code=400, detail=f"{up.filename}: 읽을 수 없습니다 ({type(exc).__name__})"
|
|
)
|
|
if parsed["lines"]:
|
|
saved += store.upsert_sales_bulk(parsed["lines"])["saved"]
|
|
for wk in parsed["weeklies"]:
|
|
try:
|
|
store.upsert_sales_weekly(data=wk)
|
|
weekly_saved += 1
|
|
except ValueError:
|
|
continue
|
|
names.append(up.filename or "")
|
|
|
|
if not saved and not weekly_saved:
|
|
raise HTTPException(status_code=400, detail="읽어들인 매출 행이 없습니다.")
|
|
return JSONResponse(
|
|
{"ok": True, "saved": saved, "weekly": weekly_saved, "files": names}
|
|
)
|
|
|
|
|
|
@router.get("/sales/export.xlsx")
|
|
async def sales_export_xlsx(request: Request) -> Any:
|
|
"""현재 조회 조건 그대로 엑셀 다운로드."""
|
|
from io import BytesIO # noqa: WPS433
|
|
|
|
from fastapi.responses import StreamingResponse # noqa: WPS433
|
|
|
|
from .export import build_sales_workbook # noqa: WPS433
|
|
|
|
guard = _guard(request)
|
|
if not isinstance(guard, tuple):
|
|
return guard
|
|
store, _user = guard
|
|
|
|
filters = _sales_filters(request)
|
|
rows = store.list_sales(**filters)
|
|
wb = build_sales_workbook(rows, _sales_totals(rows))
|
|
buf = BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
stamp = (filters["date_from"] or "all").replace("-", "")
|
|
return StreamingResponse(
|
|
buf,
|
|
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
|
headers={"Content-Disposition": f'attachment; filename="cupang_sales_{stamp}.xlsx"'},
|
|
)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 상자 입수량 관리
|
|
# ════════════════════════════════════════════════════════════
|
|
@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 "")),
|
|
},
|
|
)
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────
|
|
# 쿠팡 발주 엑셀 업로드 — F13 입고예정일 / 22행부터 상품
|
|
# B열 = 쿠팡상품코드, F열 = 물류센터, G열 = 발주수량
|
|
# ────────────────────────────────────────────────────────────
|
|
PO_FIRST_ROW = 22 # 상품이 시작되는 행
|
|
PO_ARRIVAL_CELL = "F13" # 입고예정일시
|
|
|
|
|
|
def _po_cell_date(value: Any) -> _date | None:
|
|
"""F13 값(datetime / date / 문자열) → date. 인식 못 하면 None."""
|
|
from datetime import datetime as _dt # noqa: WPS433
|
|
|
|
if isinstance(value, _dt):
|
|
return value.date()
|
|
if isinstance(value, _date):
|
|
return value
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return None
|
|
text = text.split(" ")[0].replace(".", "-").replace("/", "-")
|
|
try:
|
|
return _date.fromisoformat(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _po_int(value: Any) -> int:
|
|
try:
|
|
return int(float(str(value).replace(",", "").strip()))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _parse_po_sheet(ws: Any) -> dict[str, Any]:
|
|
"""발주서 시트 1장 → {arrival_date, rows:[{coupang_item_code, center_name, quantity}]}."""
|
|
arrival = _po_cell_date(ws[PO_ARRIVAL_CELL].value)
|
|
rows: list[dict[str, Any]] = []
|
|
for r in range(PO_FIRST_ROW, (ws.max_row or PO_FIRST_ROW) + 1):
|
|
code = str(ws.cell(row=r, column=2).value or "").strip() # B
|
|
if not code or code in ("합계", "소계"):
|
|
continue
|
|
code = code.split(".")[0] if code.replace(".", "").isdigit() else code
|
|
center = str(ws.cell(row=r, column=6).value or "").strip() # F
|
|
qty = _po_int(ws.cell(row=r, column=7).value) # G
|
|
if not center or qty <= 0:
|
|
continue
|
|
rows.append({"coupang_item_code": code, "center_name": center, "quantity": qty})
|
|
return {"arrival_date": arrival, "rows": rows}
|
|
|
|
|
|
@router.post("/api/box-calc/upload")
|
|
async def box_calc_upload(
|
|
request: Request,
|
|
files: list[UploadFile] = File(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""쿠팡 발주 엑셀 여러 개 → 센터별 제품 수량 합산.
|
|
|
|
- 출고일 = F13(입고예정일)의 하루 전
|
|
- 쿠팡상품코드(B) → cupang_products.coupang_item_code 로 제품 매칭
|
|
- 센터명(F) → cupang_centers.name 으로 매칭(없으면 그 이름으로 자동 등록)
|
|
"""
|
|
from io import BytesIO # noqa: WPS433
|
|
from datetime import timedelta as _timedelta # noqa: WPS433
|
|
|
|
from openpyxl import load_workbook # noqa: WPS433
|
|
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
if not files:
|
|
raise HTTPException(status_code=400, detail="엑셀 파일을 선택하세요.")
|
|
|
|
products = store.list_products(include_inactive=True)
|
|
by_coupang = {
|
|
(p.get("coupang_item_code") or "").strip(): p
|
|
for p in products
|
|
if (p.get("coupang_item_code") or "").strip()
|
|
}
|
|
centers = store.list_centers(include_inactive=True)
|
|
by_center_name = {(c.get("name") or "").strip(): c for c in centers}
|
|
next_center_order = max((int(c.get("sort_order") or 0) for c in centers), default=0) + 1
|
|
rules = {r["product_code"]: r for r in store.list_box_rules()}
|
|
|
|
warnings: list[str] = []
|
|
file_infos: list[dict[str, Any]] = []
|
|
ship_dates: list[str] = []
|
|
# (출고일, 센터명, 제품코드) → 수량
|
|
agg: dict[tuple[str, str, str], dict[str, Any]] = {}
|
|
unknown_codes: set[str] = set()
|
|
added_centers: set[str] = set()
|
|
no_rule: set[str] = set()
|
|
|
|
for up in files:
|
|
raw = await up.read()
|
|
name = up.filename or "(이름 없음)"
|
|
try:
|
|
wb = load_workbook(BytesIO(raw), data_only=True)
|
|
except Exception: # noqa: BLE001 - 엑셀이 아니거나 손상
|
|
warnings.append(f"{name}: 엑셀 파일을 읽지 못했습니다.")
|
|
continue
|
|
parsed = _parse_po_sheet(wb.worksheets[0])
|
|
wb.close()
|
|
|
|
arrival = parsed["arrival_date"]
|
|
ship = (arrival - _timedelta(days=1)) if arrival else None
|
|
if ship:
|
|
ship_dates.append(ship.isoformat())
|
|
else:
|
|
warnings.append(f"{name}: F13 입고예정일을 읽지 못했습니다.")
|
|
|
|
ship_iso = ship.isoformat() if ship else ""
|
|
used = 0
|
|
for row in parsed["rows"]:
|
|
prod = by_coupang.get(row["coupang_item_code"])
|
|
if not prod:
|
|
unknown_codes.add(row["coupang_item_code"])
|
|
continue
|
|
cname = row["center_name"]
|
|
if cname not in by_center_name:
|
|
# 미등록 센터는 발주서에 적힌 이름 그대로 자동 등록해 계산에 포함한다.
|
|
try:
|
|
created = store.create_center(name=cname, sort_order=next_center_order)
|
|
next_center_order += 1
|
|
except Exception: # noqa: BLE001 - 등록 실패해도 나머지는 계산
|
|
warnings.append(f"센터 자동 등록 실패: {cname}")
|
|
continue
|
|
by_center_name[cname] = created
|
|
added_centers.add(cname)
|
|
code = prod["product_code"]
|
|
if code not in rules:
|
|
no_rule.add(f'{prod["product_name"]}({code})')
|
|
key = (ship_iso, cname, code)
|
|
cell = agg.setdefault(
|
|
key,
|
|
{
|
|
"ship_date": ship_iso,
|
|
"center_name": cname,
|
|
"product_code": code,
|
|
"product_name": prod["product_name"],
|
|
"coupang_item_code": row["coupang_item_code"],
|
|
"quantity": 0,
|
|
},
|
|
)
|
|
cell["quantity"] += row["quantity"]
|
|
used += row["quantity"]
|
|
|
|
file_infos.append(
|
|
{
|
|
"filename": name,
|
|
"arrival_date": arrival.isoformat() if arrival else "",
|
|
"ship_date": ship_iso,
|
|
"rows": len(parsed["rows"]),
|
|
"quantity": used,
|
|
}
|
|
)
|
|
|
|
if unknown_codes:
|
|
warnings.append(
|
|
"등록되지 않은 쿠팡상품코드 " + str(len(unknown_codes)) + "건 제외: "
|
|
+ ", ".join(sorted(unknown_codes)[:10])
|
|
+ (" 외" if len(unknown_codes) > 10 else "")
|
|
)
|
|
if added_centers:
|
|
warnings.append("새 입고센터 자동 등록: " + ", ".join(sorted(added_centers)))
|
|
if no_rule:
|
|
warnings.append("상자 입수량 미설정: " + ", ".join(sorted(no_rule)))
|
|
|
|
uniq_dates = sorted(set(ship_dates))
|
|
|
|
# 출고일 → 센터 → 품목. 업로드한 파일의 출고일이 다르면 날짜별로 나뉜다.
|
|
by_date: dict[str, dict[str, dict[str, Any]]] = {}
|
|
for (ship_iso, cname, _code), cell in agg.items():
|
|
centers_of_date = by_date.setdefault(ship_iso, {})
|
|
g = centers_of_date.setdefault(
|
|
cname,
|
|
{
|
|
"center_name": cname,
|
|
"center_id": (by_center_name.get(cname) or {}).get("id"),
|
|
"items": [],
|
|
},
|
|
)
|
|
g["items"].append(
|
|
{
|
|
"product_code": cell["product_code"],
|
|
"product_name": cell["product_name"],
|
|
"coupang_item_code": cell["coupang_item_code"],
|
|
"quantity": cell["quantity"],
|
|
}
|
|
)
|
|
|
|
groups: list[dict[str, Any]] = []
|
|
for ship_iso in sorted(by_date):
|
|
centers_of_date = by_date[ship_iso]
|
|
for g in centers_of_date.values():
|
|
g["items"].sort(key=lambda it: it["product_code"])
|
|
groups.append(
|
|
{
|
|
"ship_date": ship_iso,
|
|
"centers": sorted(centers_of_date.values(), key=lambda g: g["center_name"]),
|
|
"files": [f for f in file_infos if f["ship_date"] == ship_iso],
|
|
"quantity": sum(
|
|
it["quantity"]
|
|
for g in centers_of_date.values()
|
|
for it in g["items"]
|
|
),
|
|
}
|
|
)
|
|
|
|
return JSONResponse(
|
|
{
|
|
"ok": True,
|
|
"ship_date": uniq_dates[0] if uniq_dates else "",
|
|
"ship_dates": uniq_dates,
|
|
"files": file_infos,
|
|
"groups": groups,
|
|
"warnings": warnings,
|
|
}
|
|
)
|
|
|
|
|
|
@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 로 서버에서 계산한다.
|
|
"""
|
|
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 는 배열이어야 합니다.")
|
|
|
|
return JSONResponse(_compute_boxes(store, raw_items))
|
|
|
|
|
|
def _compute_boxes(store: Any, raw_items: list[Any]) -> dict[str, Any]:
|
|
"""[{product_code, quantity}] → 제품별 상자 + 자투리 혼합 상자 + 합계.
|
|
|
|
상자 계산 화면(API)과 출고 상세 보기가 같은 결과를 쓰도록 한 곳에 둔다.
|
|
"""
|
|
from .store import compute_boxes # noqa: WPS433
|
|
|
|
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 {
|
|
"results": results,
|
|
"totals": sorted(totals.values(), key=lambda t: t["box_name"]),
|
|
"mixes": mixes,
|
|
"grand_total_boxes": grand_total,
|
|
}
|
|
|
|
|
|
def _expand_box_list(
|
|
shipment: dict[str, Any], calc: dict[str, Any], items: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
"""전체 상자를 1번부터 번호 붙여 펼친다 — 상자마다 무엇이 몇 개 들었는지.
|
|
|
|
확정 당시 스냅샷(`box_plan`)이 있으면 그것을 그대로 펼치고,
|
|
없는 예전 데이터는 계산 결과(`calc`)로 대신 만든다.
|
|
"""
|
|
box_name_by_code = {
|
|
r["product_code"]: (r.get("box_name") or "") for r in calc.get("results", [])
|
|
}
|
|
out: list[dict[str, Any]] = []
|
|
|
|
plan = shipment.get("box_plan") or []
|
|
if plan:
|
|
for entry in plan:
|
|
count = int(entry.get("count") or 0)
|
|
units = int(entry.get("units") or 0)
|
|
if count <= 0:
|
|
continue
|
|
if entry.get("kind") == "mix":
|
|
contents = [
|
|
{
|
|
"product_name": it.get("product_name") or it.get("product_code") or "",
|
|
"product_code": it.get("product_code") or "",
|
|
"quantity": int(it.get("quantity") or 0),
|
|
}
|
|
for it in (entry.get("items") or [])
|
|
]
|
|
box_name = entry.get("box_type") or ""
|
|
for _ in range(count):
|
|
out.append({"kind": "mix", "box_name": box_name, "items": contents})
|
|
else:
|
|
code = str(entry.get("product_code") or "")
|
|
contents = [
|
|
{
|
|
"product_name": entry.get("name") or code,
|
|
"product_code": code,
|
|
"quantity": units,
|
|
}
|
|
]
|
|
box_name = box_name_by_code.get(code, "")
|
|
for _ in range(count):
|
|
out.append({"kind": "product", "box_name": box_name, "items": contents})
|
|
else:
|
|
for r in calc.get("results", []):
|
|
if not r.get("configured"):
|
|
continue
|
|
for _ in range(int(r.get("full_boxes") or 0)):
|
|
out.append(
|
|
{
|
|
"kind": "product",
|
|
"box_name": r.get("box_name") or "",
|
|
"items": [
|
|
{
|
|
"product_name": r["product_name"],
|
|
"product_code": r["product_code"],
|
|
"quantity": r["units_per_box"],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
for mix in calc.get("mixes", []):
|
|
for box in mix.get("boxes", []):
|
|
out.append(
|
|
{
|
|
"kind": "mix",
|
|
"box_name": mix.get("box_name") or "",
|
|
"items": [
|
|
{
|
|
"product_name": it.get("product_name") or "",
|
|
"product_code": it.get("product_code") or "",
|
|
"quantity": int(it.get("quantity") or 0),
|
|
}
|
|
for it in (box.get("items") or [])
|
|
],
|
|
}
|
|
)
|
|
|
|
for idx, box in enumerate(out, start=1):
|
|
box["no"] = idx
|
|
box["quantity"] = sum(int(it["quantity"]) for it in box["items"])
|
|
return out
|
|
|
|
|
|
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","coupang_item_code"}, ...]}
|
|
coupang_item_code 키가 없으면 기존 값을 유지한다.
|
|
"""
|
|
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()
|
|
cic = it.get("coupang_item_code")
|
|
if not code or not name:
|
|
continue
|
|
try:
|
|
store.upsert_product(
|
|
product_code=code,
|
|
product_name=name,
|
|
coupang_item_code=(None if cic is None else str(cic)),
|
|
)
|
|
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(...),
|
|
coupang_item_code: 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,
|
|
# 빈 값은 "미입력" 으로 보고 기존 쿠팡상품코드를 유지한다.
|
|
coupang_item_code=(coupang_item_code.strip() or None),
|
|
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("/api/products/{product_id:int}/edit")
|
|
async def product_edit(
|
|
request: Request,
|
|
product_id: int,
|
|
payload: dict[str, Any] = Body(...),
|
|
user: dict[str, Any] = Depends(_require_user),
|
|
) -> JSONResponse:
|
|
"""등록된 제품 수정. body: {product_name, product_code, coupang_item_code}."""
|
|
store = _store(request)
|
|
if store is None:
|
|
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
|
try:
|
|
product = store.update_product(
|
|
product_id=product_id,
|
|
product_name=str(payload.get("product_name") or ""),
|
|
product_code=str(payload.get("product_code") or ""),
|
|
coupang_item_code=str(payload.get("coupang_item_code") or ""),
|
|
)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="제품을 찾을 수 없습니다.")
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
return JSONResponse({"ok": True, "product": product})
|
|
|
|
|
|
@router.post("/api/products/{product_id:int}/toggle")
|
|
async def product_toggle_active(
|
|
request: Request,
|
|
product_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:
|
|
product = store.toggle_product_active(product_id=product_id)
|
|
except KeyError:
|
|
raise HTTPException(status_code=404, detail="제품을 찾을 수 없습니다.")
|
|
return JSONResponse({"ok": True, "active": bool(product["active"])})
|
|
|
|
|
|
@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)
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 분배 확정 — 센터별 출고 묶음 생성
|
|
# ════════════════════════════════════════════════════════════
|
|
def _clean_box_plan(raw: Any) -> list[dict[str, Any]]:
|
|
"""화면이 보낸 상자 구성을 표시용으로만 정리해서 저장한다.
|
|
|
|
수량·상자 수는 라인(제품 합계)이 정답이고, 이 값은 "어느 상자에 무엇이
|
|
담겼는지"를 보여주기 위한 스냅샷이다.
|
|
"""
|
|
if not isinstance(raw, list):
|
|
return []
|
|
out: list[dict[str, Any]] = []
|
|
for entry in raw[:200]:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
kind = "mix" if str(entry.get("kind")) == "mix" else "product"
|
|
try:
|
|
count = max(int(entry.get("count") or 0), 0)
|
|
units = max(int(entry.get("units") or 0), 0)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if count <= 0:
|
|
continue
|
|
item: dict[str, Any] = {
|
|
"kind": kind,
|
|
"name": str(entry.get("name") or "")[:120],
|
|
"count": count,
|
|
"units": units,
|
|
"quantity": count * units,
|
|
}
|
|
if kind == "product":
|
|
item["product_code"] = str(entry.get("product_code") or "")[:60]
|
|
else:
|
|
item["box_type"] = str(entry.get("box_type") or "")[:40]
|
|
contents = []
|
|
for it in (entry.get("items") or [])[:50]:
|
|
if not isinstance(it, dict):
|
|
continue
|
|
try:
|
|
qty = max(int(it.get("quantity") or 0), 0)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if qty <= 0:
|
|
continue
|
|
contents.append(
|
|
{
|
|
"product_name": str(it.get("product_name") or "")[:120],
|
|
"product_code": str(it.get("product_code") or "")[:60],
|
|
"quantity": qty,
|
|
}
|
|
)
|
|
item["items"] = contents
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
@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()
|
|
arrival_date = (_date.fromisoformat(ship_date) + _timedelta(days=1)).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())
|
|
# 화면에서 고른 상자 종류별 상자 수(예: "쿠팡상자41"). 출고리스트 엑셀의 "출고" 칸.
|
|
summary = str(raw.get("box_summary") or "").strip()
|
|
if not summary:
|
|
summary = f"{boxes}상자 · {pieces}개" if boxes else f"{pieces}개"
|
|
|
|
plans.append(
|
|
{
|
|
"center": center,
|
|
"method": method,
|
|
"lines": lines,
|
|
"summary": summary,
|
|
"box_plan": _clean_box_plan(raw.get("box_plan")),
|
|
}
|
|
)
|
|
|
|
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": arrival_date,
|
|
"center_id": center["id"],
|
|
"center_name_snapshot": center["name"],
|
|
"ship_method": plan["method"],
|
|
"outbound_summary": plan["summary"],
|
|
"worker": worker,
|
|
"status": "출고준비",
|
|
"memo": "상자 계산에서 분배 확정",
|
|
"box_plan": plan["box_plan"],
|
|
},
|
|
lines=plan["lines"],
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
created.append({"id": ship["id"], "center_name": center["name"]})
|
|
|
|
# 구글 스프레드시트에 출고일 시트 기록(설정돼 있을 때만).
|
|
sheet = _push_to_google_sheet(store, ship_date)
|
|
_log_sheet("push", ship_date, sheet)
|
|
|
|
return JSONResponse({"created": created, "ship_date": ship_date, "sheet": sheet})
|
|
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 상자 계산 임시 저장 (화면 상태 스냅샷)
|
|
# ════════════════════════════════════════════════════════════
|
|
@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"}
|