feat(cupang): 쿠팡 발주 엑셀 업로드 → 센터별 박스 자동 계산
- ① 카드를 "쿠팡 발주 업로드"로 개편, xlsx 다중 선택 업로드 - POST /cupang/api/box-calc/upload (openpyxl): F13 입고예정일 −1일 = 출고일, 22행부터 B=쿠팡상품코드 / F=센터명 / G=수량을 읽어 파일들을 합산 - 쿠팡상품코드 → cupang_products.coupang_item_code 로 제품 매칭, 센터명 → cupang_centers.name 매칭. 미매칭 코드/센터/박스규칙은 경고로 표시 - 박스는 센터 단위로 포장되므로 센터마다 박스 계산을 돌려 ② 요약에 합치고 ③ 센터에 자동 배분(자투리 혼합 박스는 소속 센터를 라벨에 표시) - 확정 대화상자 기본 출고일도 업로드한 발주서 날짜로 채움 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -20,7 +20,7 @@ from datetime import date as _date
|
||||
|
||||
from app.timezone import today_kst
|
||||
|
||||
from fastapi import APIRouter, Body, Depends, Form, HTTPException, Request
|
||||
from fastapi import APIRouter, Body, Depends, File, Form, HTTPException, Request, UploadFile
|
||||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||||
|
||||
from .holidays import is_holiday
|
||||
@@ -674,8 +674,8 @@ async def box_calc_page(request: Request) -> HTMLResponse:
|
||||
"user": user,
|
||||
"is_admin": is_admin(user),
|
||||
"nav_items": build_erp_nav(user, active="cupang"),
|
||||
"page_title": "쿠팡 밀크런 — 박스 계산",
|
||||
"page_subtitle": "제품명과 수량을 넣으면 박스 수와 남은 낱개를 계산합니다.",
|
||||
"page_title": "쿠팡 밀크런 — 쿠팡 발주 업로드",
|
||||
"page_subtitle": "쿠팡 발주 엑셀을 올리면 센터별 수량을 합산해 박스를 계산합니다.",
|
||||
"box_rules": store.list_box_rules(),
|
||||
# 센터 선택 드롭다운은 가나다순 (한글 음절은 코드포인트 순 = 가나다순)
|
||||
"centers": sorted(store.list_centers(), key=lambda c: (c.get("name") or "")),
|
||||
@@ -683,6 +683,200 @@ async def box_calc_page(request: Request) -> HTMLResponse:
|
||||
)
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────
|
||||
# 쿠팡 발주 엑셀 업로드 — 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}
|
||||
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], dict[str, Any]] = {}
|
||||
unknown_codes: set[str] = set()
|
||||
unknown_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 입고예정일을 읽지 못했습니다.")
|
||||
|
||||
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:
|
||||
unknown_centers.add(cname)
|
||||
code = prod["product_code"]
|
||||
if code not in rules:
|
||||
no_rule.add(f'{prod["product_name"]}({code})')
|
||||
key = (cname, code)
|
||||
cell = agg.setdefault(
|
||||
key,
|
||||
{
|
||||
"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.isoformat() if ship else "",
|
||||
"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 unknown_centers:
|
||||
warnings.append("등록되지 않은 센터: " + ", ".join(sorted(unknown_centers)))
|
||||
if no_rule:
|
||||
warnings.append("박스 입수량 미설정: " + ", ".join(sorted(no_rule)))
|
||||
|
||||
uniq_dates = sorted(set(ship_dates))
|
||||
if len(uniq_dates) > 1:
|
||||
warnings.append("파일마다 출고일이 다릅니다: " + ", ".join(uniq_dates))
|
||||
|
||||
grouped: dict[str, dict[str, Any]] = {}
|
||||
for (cname, _code), cell in agg.items():
|
||||
g = grouped.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"],
|
||||
}
|
||||
)
|
||||
for g in grouped.values():
|
||||
g["items"].sort(key=lambda it: it["product_code"])
|
||||
|
||||
return JSONResponse(
|
||||
{
|
||||
"ok": True,
|
||||
"ship_date": uniq_dates[0] if uniq_dates else "",
|
||||
"ship_dates": uniq_dates,
|
||||
"files": file_infos,
|
||||
"centers": sorted(grouped.values(), key=lambda g: g["center_name"]),
|
||||
"warnings": warnings,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@router.post("/api/box-calc")
|
||||
async def box_calc_api(
|
||||
request: Request,
|
||||
|
||||
Reference in New Issue
Block a user