feat(cupang): 쿠팡 밀크런 모듈 추가 (출고 달력/박스계산/입고센터)
- app/modules/cupang: router/db/store/itemcode + 템플릿 5종 - cupang_db 전용(JSON 폴백 없음). CUPANG_DB_URL 미설정 시 안내 페이지 - 월간 달력 + 출고 묶음(헤더+라인), 박스 입수량 자동계산(서버 재계산) - 입고센터 관리(사용중 soft delete), 박스규칙 upsert, 엑셀 내보내기 - 상품 검색은 itemcode_db 읽기 전용(미설정 시 수동 입력) - MODULE_KEYS/메뉴/아이콘/라벨 동기화, scripts/sql/cupang_db_init.sql(멱등) - app/data/ gitignore 추가(PII) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,726 @@
|
||||
"""쿠팡 밀크런 모듈 라우터.
|
||||
|
||||
- 경로: /cupang
|
||||
- 권한: 로그인 + `cupang` 모듈 권한 (관리자는 항상 통과). 서버 측 검사.
|
||||
- 데이터: CupangDBStore (cupang_db / PostgreSQL) 전용.
|
||||
CUPANG_DB_URL 미설정 시 store 가 None 이며, 각 페이지는 "설정 필요" 안내를 보여준다.
|
||||
- 상품 검색: itemcode_db 읽기 전용(ItemcodeReader). 미설정 시 수동 입력 폴백.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import calendar as _calendar
|
||||
import io
|
||||
import json
|
||||
from datetime import date, datetime
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
||||
from fastapi.responses import (
|
||||
HTMLResponse,
|
||||
JSONResponse,
|
||||
RedirectResponse,
|
||||
StreamingResponse,
|
||||
)
|
||||
|
||||
from .store import DEFAULT_CENTERS, SHEET_COLUMNS, SHIP_METHODS, STATUSES, compute_boxes
|
||||
|
||||
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 = date.today()
|
||||
try:
|
||||
year = int(request.query_params.get("year") or today.year)
|
||||
month = int(request.query_params.get("month") or today.month)
|
||||
except ValueError:
|
||||
year, month = today.year, today.month
|
||||
if not (1 <= month <= 12):
|
||||
year, month = today.year, today.month
|
||||
return year, month
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 메인 — 월간 달력 + 선택일 출고 리스트
|
||||
# ════════════════════════════════════════════════════════════
|
||||
@router.get("/", response_class=HTMLResponse)
|
||||
async def index(request: Request) -> HTMLResponse:
|
||||
from app.main import build_erp_nav, render_template # noqa: WPS433
|
||||
from app.store import is_admin # noqa: WPS433
|
||||
|
||||
guard = _guard(request)
|
||||
if not isinstance(guard, tuple):
|
||||
return guard
|
||||
store, user = guard
|
||||
|
||||
year, month = _ym(request)
|
||||
counts = store.calendar_counts(year=year, month=month)
|
||||
shipments = store.list_shipments(year=year, month=month)
|
||||
|
||||
# 선택 날짜 (기본: 오늘이 이번 달이면 오늘, 아니면 1일)
|
||||
sel = request.query_params.get("date") or ""
|
||||
today = date.today()
|
||||
if not sel:
|
||||
sel = today.isoformat() if (today.year == year and today.month == month) else f"{year:04d}-{month:02d}-01"
|
||||
|
||||
# 선택일에 걸친 묶음(출고일 기준 우선, 작성/입고 포함)
|
||||
sel_shipments = [
|
||||
s for s in shipments
|
||||
if sel in (s.get("ship_date"), s.get("document_date"), s.get("center_arrival_date"))
|
||||
]
|
||||
|
||||
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,
|
||||
"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,
|
||||
"statuses": list(STATUSES),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/api/calendar")
|
||||
async def api_calendar(
|
||||
request: Request, _: dict[str, Any] = Depends(_require_user)
|
||||
) -> JSONResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
year, month = _ym(request)
|
||||
return JSONResponse(
|
||||
{"year": year, "month": month, "counts": store.calendar_counts(year=year, month=month)}
|
||||
)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 출고 묶음 — 등록 / 수정 / 상세
|
||||
# ════════════════════════════════════════════════════════════
|
||||
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": store.list_centers(),
|
||||
"box_rules": store.list_box_rules(),
|
||||
"ship_methods": list(SHIP_METHODS),
|
||||
"statuses": list(STATUSES),
|
||||
"search_enabled": bool(reader and reader.enabled),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/new", response_class=HTMLResponse)
|
||||
async def new_form(request: Request) -> HTMLResponse:
|
||||
from app.main import render_template # noqa: WPS433
|
||||
|
||||
guard = _guard(request)
|
||||
if not isinstance(guard, tuple):
|
||||
return guard
|
||||
store, user = guard
|
||||
ctx = _form_context(request, store, user)
|
||||
ctx.update(
|
||||
{
|
||||
"page_title": "쿠팡 밀크런 — 신규 등록",
|
||||
"page_subtitle": "공통 헤더 1개 + 품목 라인",
|
||||
"mode": "new",
|
||||
"shipment": None,
|
||||
"default_date": date.today().isoformat(),
|
||||
}
|
||||
)
|
||||
return render_template(request, "cupang/form.html", ctx)
|
||||
|
||||
|
||||
@router.post("/new")
|
||||
async def create(
|
||||
request: Request,
|
||||
lines_json: str = Form("[]"),
|
||||
document_no: 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(""),
|
||||
status: 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_no": document_no,
|
||||
"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,
|
||||
"status": status,
|
||||
"memo": memo,
|
||||
}
|
||||
try:
|
||||
ship = store.create_shipment(
|
||||
created_by=user["email"], header=header, lines=_parse_lines(lines_json)
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc))
|
||||
return RedirectResponse(url=f"/cupang/{ship['id']}", status_code=303)
|
||||
|
||||
|
||||
@router.get("/{shipment_id:int}", response_class=HTMLResponse)
|
||||
async def detail(request: Request, shipment_id: int) -> HTMLResponse:
|
||||
from app.main import build_erp_nav, render_template # noqa: WPS433
|
||||
from app.store import is_admin # noqa: WPS433
|
||||
|
||||
guard = _guard(request)
|
||||
if not isinstance(guard, tuple):
|
||||
return guard
|
||||
store, user = guard
|
||||
ship = store.get_shipment(shipment_id=shipment_id)
|
||||
if not ship:
|
||||
return render_template(
|
||||
request, "denied.html",
|
||||
{"reason": "출고 묶음을 찾을 수 없습니다.", "is_admin": is_admin(user)},
|
||||
status_code=404,
|
||||
)
|
||||
return render_template(
|
||||
request,
|
||||
"cupang/detail.html",
|
||||
{
|
||||
"user": user,
|
||||
"is_admin": is_admin(user),
|
||||
"nav_items": build_erp_nav(user, active="cupang"),
|
||||
"page_title": f"출고 #{ship['id']}",
|
||||
"page_subtitle": f"{ship['ship_date']} · {ship['center_name_snapshot']}",
|
||||
"shipment": ship,
|
||||
"statuses": list(STATUSES),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@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_no: 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_no": document_no,
|
||||
"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}/status")
|
||||
async def change_status(
|
||||
request: Request,
|
||||
shipment_id: int,
|
||||
status: 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_status(shipment_id=shipment_id, status=status)
|
||||
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.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 = store.list_centers(include_inactive=True)
|
||||
# 사용 중 여부 표시
|
||||
for c in centers:
|
||||
c["in_use"] = store.center_in_use(center_id=c["id"])
|
||||
return render_template(
|
||||
request,
|
||||
"cupang/centers.html",
|
||||
{
|
||||
"user": user,
|
||||
"is_admin": is_admin(user),
|
||||
"nav_items": build_erp_nav(user, active="cupang"),
|
||||
"page_title": "쿠팡 밀크런 — 입고센터 관리",
|
||||
"page_subtitle": "추가 · 수정 · 비활성화. 사용 중 센터는 삭제되지 않고 비활성화됩니다.",
|
||||
"centers": centers,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/centers")
|
||||
async def center_create(
|
||||
request: Request,
|
||||
name: str = Form(...),
|
||||
sort_order: int = Form(0),
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> RedirectResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
try:
|
||||
store.create_center(name=name, sort_order=sort_order)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc))
|
||||
return RedirectResponse(url="/cupang/centers", status_code=303)
|
||||
|
||||
|
||||
@router.post("/centers/{center_id}/edit")
|
||||
async def center_edit(
|
||||
request: Request,
|
||||
center_id: int,
|
||||
name: str = Form(""),
|
||||
active: str = Form(""),
|
||||
sort_order: str = Form(""),
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> RedirectResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
kwargs: dict[str, Any] = {"center_id": center_id}
|
||||
if name.strip():
|
||||
kwargs["name"] = name
|
||||
if active != "":
|
||||
kwargs["active"] = active in ("1", "true", "on", "True")
|
||||
if sort_order.strip():
|
||||
try:
|
||||
kwargs["sort_order"] = int(sort_order)
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
store.update_center(**kwargs)
|
||||
except KeyError:
|
||||
raise HTTPException(status_code=404, detail="센터를 찾을 수 없습니다.")
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc))
|
||||
return RedirectResponse(url="/cupang/centers", status_code=303)
|
||||
|
||||
|
||||
@router.post("/centers/{center_id}/delete")
|
||||
async def center_delete(
|
||||
request: Request,
|
||||
center_id: int,
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> RedirectResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
try:
|
||||
store.delete_center(center_id=center_id)
|
||||
except KeyError:
|
||||
raise HTTPException(status_code=404, detail="센터를 찾을 수 없습니다.")
|
||||
return RedirectResponse(url="/cupang/centers", status_code=303)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 박스 입수량 관리
|
||||
# ════════════════════════════════════════════════════════════
|
||||
@router.get("/box-rules", response_class=HTMLResponse)
|
||||
async def box_rules_page(request: Request) -> HTMLResponse:
|
||||
from app.main import build_erp_nav, render_template # noqa: WPS433
|
||||
from app.store import is_admin # noqa: WPS433
|
||||
|
||||
guard = _guard(request)
|
||||
if not isinstance(guard, tuple):
|
||||
return guard
|
||||
store, user = guard
|
||||
return render_template(
|
||||
request,
|
||||
"cupang/box_rules.html",
|
||||
{
|
||||
"user": user,
|
||||
"is_admin": is_admin(user),
|
||||
"nav_items": build_erp_nav(user, active="cupang"),
|
||||
"page_title": "쿠팡 밀크런 — 박스 입수량",
|
||||
"page_subtitle": "제품코드별 쿠팡박스 1박스당 입수량 설정",
|
||||
"box_rules": store.list_box_rules(include_inactive=True),
|
||||
"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.deactivate_box_rule(rule_id=rule_id)
|
||||
except KeyError:
|
||||
raise HTTPException(status_code=404, detail="규칙을 찾을 수 없습니다.")
|
||||
return RedirectResponse(url="/cupang/box-rules", status_code=303)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 상품 검색 (itemcode_db 읽기 전용) + 박스 계산 미리보기
|
||||
# ════════════════════════════════════════════════════════════
|
||||
@router.get("/api/products/search")
|
||||
async def product_search(
|
||||
request: Request,
|
||||
q: str = "",
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> JSONResponse:
|
||||
reader = _itemcode(request)
|
||||
store = _store(request)
|
||||
results = reader.search(q) if reader else []
|
||||
# box_rule 의 units_per_box 를 결과에 병합 (있으면)
|
||||
rule_map = store.box_rule_map() if store else {}
|
||||
for r in results:
|
||||
rule = rule_map.get(r["code"])
|
||||
r["units_per_box"] = rule["units_per_box"] if rule else None
|
||||
r["box_rule_id"] = rule["id"] if rule else None
|
||||
return JSONResponse(
|
||||
{
|
||||
"enabled": bool(reader and reader.enabled),
|
||||
"reason": (reader.reason if reader else "itemcode 리더 미초기화"),
|
||||
"results": results,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.get("/api/box-calc")
|
||||
async def box_calc(
|
||||
request: Request,
|
||||
product_code: str = "",
|
||||
quantity: int = 0,
|
||||
units_per_box: str = "",
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> JSONResponse:
|
||||
store = _store(request)
|
||||
upb: int | None
|
||||
if units_per_box.strip():
|
||||
try:
|
||||
upb = int(units_per_box)
|
||||
except ValueError:
|
||||
upb = None
|
||||
else:
|
||||
rule = store.get_box_rule(product_code=product_code) if store else None
|
||||
upb = rule["units_per_box"] if rule else None
|
||||
return JSONResponse(compute_boxes(quantity, upb))
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 엑셀 내보내기 (구글시트 컬럼 순서 유지)
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def _flatten_for_sheet(shipments: list[dict[str, Any]]) -> list[list[Any]]:
|
||||
"""묶음 → 시트 행. 같은 묶음 2번째 라인부터 공통 헤더 칸은 비운다."""
|
||||
rows: list[list[Any]] = []
|
||||
seq = 0
|
||||
for s in shipments:
|
||||
lines = s.get("lines") or []
|
||||
for i, ln in enumerate(lines):
|
||||
seq += 1
|
||||
if i == 0:
|
||||
rows.append([
|
||||
seq,
|
||||
s.get("document_date", ""),
|
||||
s.get("ship_date", ""),
|
||||
s.get("center_arrival_date", ""),
|
||||
s.get("center_name_snapshot", ""),
|
||||
s.get("ship_method", ""),
|
||||
ln.get("product_code", ""),
|
||||
ln.get("product_name_snapshot", ""),
|
||||
ln.get("quantity", 0),
|
||||
s.get("outbound_summary", ""),
|
||||
s.get("worker", ""),
|
||||
])
|
||||
else:
|
||||
# 공통 헤더 칸 비움 (구분/제품/수량만 채움)
|
||||
rows.append([
|
||||
seq, "", "", "", "", "",
|
||||
ln.get("product_code", ""),
|
||||
ln.get("product_name_snapshot", ""),
|
||||
ln.get("quantity", 0),
|
||||
"", "",
|
||||
])
|
||||
return rows
|
||||
|
||||
|
||||
@router.get("/export")
|
||||
async def export_month(
|
||||
request: Request,
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> StreamingResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
year, month = _ym(request)
|
||||
shipments = store.shipments_for_export(year=year, month=month)
|
||||
fname = f"cupang_milkrun_{year:04d}{month:02d}.xlsx"
|
||||
return _build_xlsx(shipments, fname, sheet_title=f"{year}-{month:02d}")
|
||||
|
||||
|
||||
@router.get("/{shipment_id:int}/export")
|
||||
async def export_one(
|
||||
request: Request,
|
||||
shipment_id: int,
|
||||
user: dict[str, Any] = Depends(_require_user),
|
||||
) -> StreamingResponse:
|
||||
store = _store(request)
|
||||
if store is None:
|
||||
raise HTTPException(status_code=503, detail="cupang_db 미설정")
|
||||
shipments = store.shipments_for_export(shipment_id=shipment_id)
|
||||
if not shipments:
|
||||
raise HTTPException(status_code=404, detail="출고 묶음을 찾을 수 없습니다.")
|
||||
fname = f"cupang_milkrun_{shipment_id}.xlsx"
|
||||
return _build_xlsx(shipments, fname, sheet_title=f"출고{shipment_id}")
|
||||
|
||||
|
||||
def _build_xlsx(shipments: list[dict[str, Any]], fname: str, *, sheet_title: str) -> StreamingResponse:
|
||||
from openpyxl import Workbook # 지연 import
|
||||
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
ws.title = sheet_title[:31] or "cupang"
|
||||
ws.append(list(SHEET_COLUMNS))
|
||||
for row in _flatten_for_sheet(shipments):
|
||||
ws.append(row)
|
||||
widths = [6, 12, 12, 12, 12, 10, 14, 26, 8, 24, 12]
|
||||
for col, w in enumerate(widths, start=1):
|
||||
ws.column_dimensions[ws.cell(row=1, column=col).column_letter].width = w
|
||||
|
||||
buf = io.BytesIO()
|
||||
wb.save(buf)
|
||||
buf.seek(0)
|
||||
return StreamingResponse(
|
||||
buf,
|
||||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||||
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/health")
|
||||
async def health() -> dict[str, str]:
|
||||
return {"status": "ok", "module": "cupang"}
|
||||
Reference in New Issue
Block a user