feat(cupang): 쿠팡 로켓 매출 화면 + cupang_sales 테이블

- scripts/sql/cupang_db_006_sales.sql: cupang_sales(발주 라인) / cupang_sales_weekly(광고비·할인·장려금) 생성
- scripts/sql/cupang_sales_seed.sql: 구글 시트 초기 데이터 2,218행 + 주차비용 100건
  (공급가 합계가 시트 2024/2025 총계와 일치)
- /cupang/sales: 기간·센터·발주유형·검색 조회, 합계 KPI(공급가/원가/물류비/마진/순마진),
  행 추가·수정·삭제, 시트(xlsx·csv) 업로드 일괄 등록, 조회 조건 그대로 엑셀 다운로드
- 달력 상단에 "쿠팡 로켓 매출" 버튼 추가
- sales_import.py: 병합 머리글 3행 건너뛰고 라인/주차 소계 분리, 월계·총계는 저장하지 않음

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-02 16:03:08 +09:00
parent 15de38cb74
commit 0979a89ed9
15 changed files with 3551 additions and 7 deletions
+161
View File
@@ -0,0 +1,161 @@
"""쿠팡 로켓 매출 시트 파서 (xlsx / csv).
구글 시트 "쿠팡 로켓 매출" 양식을 그대로 읽는다.
1~3행 병합 머리글
4행~ 발주 라인 1건 = 1행
· 주차 소계 행: 광고비(CPC)/할인 프로모션/장려금이 여기에만 있다
· 월계/총계 행: 저장하지 않는다(화면에서 합산)
열 순서(0-based)
0 구분 1 순번 2 발주번호 3 발주유형 4 발주일 5 출고일 6 센터입고일
7 SKUID 8 바코드 9 품목 10 수량 11 입고센터
12 공급단가 13 공급가 14 원가(단가) 15 원가(발주량)
16 피킹비 단가 17 피킹비 합계 18 밀크런/쉽먼트 19 물류비 합계 20 물류비중(%)
21 마진 22 마진율 23 광고비(CPC) 24 할인 프로모션 25 장려금
26 순마진 27 순마진율 28 사방넷 재고차감일
"""
from __future__ import annotations
import csv
import io
import re
from datetime import date, datetime
from typing import Any
COL = {
"week": 0, "seq": 1, "po_no": 2, "po_type": 3, "order_date": 4,
"ship_date": 5, "center_arrival_date": 6, "sku_id": 7, "barcode": 8,
"item_name": 9, "quantity": 10, "center_name": 11,
"supply_unit_price": 12, "supply_amount": 13,
"cost_unit_price": 14, "cost_amount": 15,
"picking_unit_price": 16, "picking_amount": 17, "milkrun_amount": 18,
"logistics_total": 19, "logistics_ratio": 20,
"margin": 21, "margin_rate": 22,
"ad_cost": 23, "promo_discount": 24, "incentive": 25,
"net_margin": 26, "net_margin_rate": 27, "stock_deduct_memo": 28,
}
BAD = {"", "-", "", "#DIV/0!", "#N/A", "#VALUE!", "#REF!"}
HEADER_ROWS = 3
def _cell(row: list[Any], key: str) -> str:
idx = COL[key]
if idx >= len(row):
return ""
value = row[idx]
if value is None:
return ""
if isinstance(value, datetime):
return value.date().isoformat()
if isinstance(value, date):
return value.isoformat()
return str(value).strip()
def _flat(text: str) -> str:
"""여러 줄 라벨을 한 줄로 — "\\n8월2주차\\n(8/11~8/17)""8월2주차(8/11~8/17)"."""
return re.sub(r"\s+", "", (text or "").strip())
def parse_rows(rows: list[list[Any]]) -> dict[str, Any]:
"""표 전체 → {lines, weeklies, skipped}. 값 정규화는 DB 계층이 한 번 더 한다."""
lines: list[dict[str, Any]] = []
weeklies: list[dict[str, Any]] = []
week_label = ""
skipped = 0
for row in rows[HEADER_ROWS:]:
if not row or not any(str(c or "").strip() for c in row):
continue
week = _flat(_cell(row, "week"))
seq = _cell(row, "seq")
po_no = _cell(row, "po_no")
if seq.replace(".0", "").isdigit() and po_no and po_no not in BAD:
if week:
week_label = week
line = {"week_label": week_label, "seq": seq.replace(".0", "")}
for field in (
"po_no", "po_type", "order_date", "ship_date", "center_arrival_date",
"sku_id", "barcode", "item_name", "quantity", "center_name",
"supply_unit_price", "supply_amount", "cost_unit_price", "cost_amount",
"picking_unit_price", "picking_amount", "milkrun_amount",
"logistics_total", "logistics_ratio", "margin", "margin_rate",
):
line[field] = _cell(row, field)
line["stock_deduct_memo"] = re.sub(
r"\s+", " ", _cell(row, "stock_deduct_memo")
).strip()
# SKU/바코드가 숫자로 읽히면 소수점이 붙는다 → 정수 표기로
for key in ("sku_id", "barcode", "po_no"):
if line[key].endswith(".0"):
line[key] = line[key][:-2]
lines.append(line)
continue
# 주차 소계 — 광고비/할인/장려금만 가져온다
if week and "주차" in week:
weeklies.append({
"week_label": week_label,
"ad_cost": _cell(row, "ad_cost"),
"promo_discount": _cell(row, "promo_discount"),
"incentive": _cell(row, "incentive"),
})
continue
skipped += 1
# 주차 기간은 그 주차 라인들의 출고일 최소/최대로 채운다
span: dict[str, tuple[str, str]] = {}
for line in lines:
d = str(line.get("ship_date") or "")
if not d or d in BAD:
continue
lo, hi = span.get(line["week_label"], (d, d))
span[line["week_label"]] = (min(lo, d), max(hi, d))
seen: set[str] = set()
weekly_rows: list[dict[str, Any]] = []
for wk in weeklies:
label = wk["week_label"]
if not label or label in seen:
continue
seen.add(label)
lo, hi = span.get(label, ("", ""))
wk["week_from"], wk["week_to"] = lo, hi
weekly_rows.append(wk)
return {"lines": lines, "weeklies": weekly_rows, "skipped": skipped}
def parse_xlsx(data: bytes) -> dict[str, Any]:
from openpyxl import load_workbook # noqa: WPS433
wb = load_workbook(io.BytesIO(data), data_only=True, read_only=True)
ws = wb[wb.sheetnames[0]]
rows = [list(r) for r in ws.iter_rows(values_only=True)]
wb.close()
return parse_rows(rows)
def parse_csv(data: bytes) -> dict[str, Any]:
for encoding in ("utf-8-sig", "utf-8", "cp949"):
try:
text = data.decode(encoding)
break
except UnicodeDecodeError:
continue
else:
raise ValueError("CSV 인코딩을 알 수 없습니다(UTF-8 또는 CP949).")
return parse_rows([row for row in csv.reader(io.StringIO(text))])
def parse_upload(filename: str, data: bytes) -> dict[str, Any]:
name = (filename or "").lower()
if name.endswith(".csv"):
return parse_csv(data)
if name.endswith((".xlsx", ".xlsm")):
return parse_xlsx(data)
raise ValueError("xlsx 또는 csv 파일만 올릴 수 있습니다.")