Files
dbx-main/app/modules/cupang/db.py
T
king 9ae27bce8a style(cupang): 매출 조회 조건 1행 고정 + 주차 필터, 상단 합계 KPI 제거
- 조회 조건은 flex-wrap: nowrap 로 한 줄 유지(좁은 화면은 가로 스크롤)
- 주차(week_label) 필터 추가 — 최근 주차부터 목록
- 공급가/원가/물류비/마진/순마진 합계 카드 삭제, 표 영역 높이 확대

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-02 16:13:49 +09:00

1125 lines
49 KiB
Python

"""cupang_db PostgreSQL 저장소.
- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — expense 모듈과 동일 패턴.
- 연결 정보: 환경변수 `CUPANG_DB_URL`
(예: postgresql://cupang_app:<pwd>@postgres-db:5432/cupang_db)
- 스키마(테이블/인덱스/트리거/seed)는 앱이 만들지 않는다.
`scripts/sql/cupang_db_init.sql` 을 superuser 가 사전 적용한다.
앱 계정(cupang_app)은 SELECT/INSERT/UPDATE/DELETE 권한만 받는다.
- 연결 풀은 lazy open — 부팅 시 DB 가 잠시 끊겨도 컨테이너가 죽지 않게.
상자 수 계산은 서버에서 `store.compute_boxes` 로 재계산하여 저장한다.
클라이언트가 보낸 상자 수는 신뢰하지 않는다.
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from typing import Any
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from app.timezone import KST
from .store import STATUSES, compute_boxes
class CupangDBStore:
def __init__(self, dsn: str, *, min_size: int = 1, max_size: int = 5):
self._pool = ConnectionPool(
conninfo=dsn,
min_size=min_size,
max_size=max_size,
kwargs={"row_factory": dict_row, "autocommit": True},
open=False,
)
self._pool.open(wait=False)
def close(self) -> None:
self._pool.close()
# ════════════════════════════════════════════════════════════
# 입고센터 (cupang_centers)
# ════════════════════════════════════════════════════════════
def list_centers(self, *, include_inactive: bool = False) -> list[dict[str, Any]]:
where = "" if include_inactive else "WHERE active = TRUE"
with self._pool.connection() as conn:
rows = conn.execute(
f"SELECT * FROM cupang_centers {where} "
"ORDER BY active DESC, sort_order ASC, name ASC"
).fetchall()
return [self._center_serialize(r) for r in rows]
def get_center(self, *, center_id: int) -> dict[str, Any] | None:
with self._pool.connection() as conn:
row = conn.execute(
"SELECT * FROM cupang_centers WHERE id = %s", (center_id,)
).fetchone()
return self._center_serialize(row) if row else None
def create_center(self, *, name: str, sort_order: int = 0) -> dict[str, Any]:
name = (name or "").strip()
if not name:
raise ValueError("센터명 필수")
with self._pool.connection() as conn:
row = conn.execute(
"""
INSERT INTO cupang_centers (name, sort_order)
VALUES (%s, %s)
ON CONFLICT (name) DO UPDATE
SET active = TRUE, sort_order = EXCLUDED.sort_order
RETURNING *
""",
(name, sort_order),
).fetchone()
return self._center_serialize(row)
def update_center(
self,
*,
center_id: int,
name: str | None = None,
active: bool | None = None,
sort_order: int | None = None,
) -> dict[str, Any]:
sets: list[str] = []
params: list[Any] = []
if name is not None:
n = name.strip()
if not n:
raise ValueError("센터명은 비울 수 없습니다.")
sets.append("name = %s")
params.append(n)
if active is not None:
sets.append("active = %s")
params.append(bool(active))
if sort_order is not None:
sets.append("sort_order = %s")
params.append(int(sort_order))
if not sets:
raise ValueError("변경할 값이 없습니다.")
params.append(center_id)
with self._pool.connection() as conn:
row = conn.execute(
f"UPDATE cupang_centers SET {', '.join(sets)} "
"WHERE id = %s RETURNING *",
params,
).fetchone()
if not row:
raise KeyError(center_id)
return self._center_serialize(row)
def center_in_use(self, *, center_id: int) -> bool:
with self._pool.connection() as conn:
row = conn.execute(
"SELECT 1 FROM cupang_shipments WHERE center_id = %s LIMIT 1",
(center_id,),
).fetchone()
return row is not None
def delete_center(self, *, center_id: int) -> dict[str, Any]:
"""사용 중이면 hard delete 하지 않고 active=false 로 비활성화한다.
반환: {"deleted": bool, "deactivated": bool}
"""
if self.center_in_use(center_id=center_id):
self.update_center(center_id=center_id, active=False)
return {"deleted": False, "deactivated": True}
with self._pool.connection() as conn:
cur = conn.execute(
"DELETE FROM cupang_centers WHERE id = %s", (center_id,)
)
if cur.rowcount == 0:
raise KeyError(center_id)
return {"deleted": True, "deactivated": False}
# ════════════════════════════════════════════════════════════
# 상자 입수량 규칙 (cupang_box_rules)
# ════════════════════════════════════════════════════════════
def list_box_rules(self, *, include_inactive: bool = False) -> list[dict[str, Any]]:
where = "" if include_inactive else "WHERE active = TRUE"
with self._pool.connection() as conn:
rows = conn.execute(
f"SELECT * FROM cupang_box_rules {where} "
"ORDER BY product_code ASC"
).fetchall()
return [self._rule_serialize(r) for r in rows]
def upsert_box_rule(
self,
*,
product_code: str,
units_per_box: int,
product_name_snapshot: str = "",
box_name: str = "쿠팡상자",
memo: str = "",
) -> dict[str, Any]:
code = (product_code or "").strip()
if not code:
raise ValueError("제품코드 필수")
upb = int(units_per_box)
if upb <= 0:
raise ValueError("상자당 입수량은 1 이상이어야 합니다.")
with self._pool.connection() as conn:
row = conn.execute(
"""
INSERT INTO cupang_box_rules
(product_code, product_name_snapshot, box_name, units_per_box, memo)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (product_code) DO UPDATE
SET product_name_snapshot = EXCLUDED.product_name_snapshot,
box_name = EXCLUDED.box_name,
units_per_box = EXCLUDED.units_per_box,
memo = EXCLUDED.memo,
active = TRUE
RETURNING *
""",
(code, product_name_snapshot.strip(), (box_name or "쿠팡상자").strip(), upb, memo.strip()),
).fetchone()
return self._rule_serialize(row)
def delete_box_rule(self, *, rule_id: int) -> None:
"""완전 삭제(hard). 라인의 box_rule_id 는 ON DELETE 미설정이므로
참조 중이면 FK 위반 가능 → 참조 라인의 box_rule_id 를 먼저 NULL 처리."""
with self._pool.connection() as conn:
with conn.transaction():
conn.execute(
"UPDATE cupang_shipment_lines SET box_rule_id = NULL WHERE box_rule_id = %s",
(rule_id,),
)
cur = conn.execute(
"DELETE FROM cupang_box_rules WHERE id = %s", (rule_id,)
)
if cur.rowcount == 0:
raise KeyError(rule_id)
# ════════════════════════════════════════════════════════════
# 상자 계산 임시 저장 (cupang_box_calc_drafts)
# payload 는 화면 상태 스냅샷(JSONB). 상자 수 계산은 불러온 뒤
# 서버에서 다시 하므로 여기 값은 신뢰 대상이 아니다.
# ════════════════════════════════════════════════════════════
def list_box_calc_drafts(self, *, limit: int = 50) -> list[dict[str, Any]]:
"""목록용 — payload 는 크므로 제외한다."""
with self._pool.connection() as conn:
rows = conn.execute(
"""
SELECT id, title, created_by, created_at, updated_at
FROM cupang_box_calc_drafts
ORDER BY updated_at DESC
LIMIT %s
""",
(int(limit),),
).fetchall()
return [self._draft_serialize(r) for r in rows]
def get_box_calc_draft(self, *, draft_id: int) -> dict[str, Any] | None:
with self._pool.connection() as conn:
row = conn.execute(
"SELECT * FROM cupang_box_calc_drafts WHERE id = %s", (draft_id,)
).fetchone()
return self._draft_serialize(row) if row else None
def save_box_calc_draft(
self,
*,
title: str,
payload: dict[str, Any],
created_by: str = "",
draft_id: int | None = None,
) -> dict[str, Any]:
"""draft_id 가 있으면 덮어쓰고, 없으면 새로 만든다.
같은 제목이 이미 있으면 그 건을 덮어쓴다(임시 저장이라 목록이 무한히
늘어나지 않게).
"""
name = (title or "").strip()
if not name:
raise ValueError("제목 필수")
data = Jsonb(payload if isinstance(payload, dict) else {})
with self._pool.connection() as conn:
if draft_id is None:
hit = conn.execute(
"SELECT id FROM cupang_box_calc_drafts WHERE title = %s",
(name,),
).fetchone()
if hit:
draft_id = int(hit["id"])
if draft_id is not None:
row = conn.execute(
"""
UPDATE cupang_box_calc_drafts
SET title = %s, payload = %s, created_by = %s
WHERE id = %s
RETURNING *
""",
(name, data, (created_by or "").strip(), draft_id),
).fetchone()
if not row:
raise KeyError(draft_id)
else:
row = conn.execute(
"""
INSERT INTO cupang_box_calc_drafts (title, payload, created_by)
VALUES (%s, %s, %s)
RETURNING *
""",
(name, data, (created_by or "").strip()),
).fetchone()
return self._draft_serialize(row)
def delete_box_calc_draft(self, *, draft_id: int) -> None:
with self._pool.connection() as conn:
cur = conn.execute(
"DELETE FROM cupang_box_calc_drafts WHERE id = %s", (draft_id,)
)
if cur.rowcount == 0:
raise KeyError(draft_id)
# ════════════════════════════════════════════════════════════
# 제품명 카탈로그 (cupang_products)
# itemcode_db 에서 가져와 등록한 제품 목록. 폼의 제품명 드롭다운 소스.
# 제품명 선택 → product_code 자동 채움.
# ════════════════════════════════════════════════════════════
def list_products(self, *, include_inactive: bool = False) -> list[dict[str, Any]]:
where = "" if include_inactive else "WHERE active = TRUE"
with self._pool.connection() as conn:
rows = conn.execute(
f"SELECT * FROM cupang_products {where} "
"ORDER BY product_code ASC"
).fetchall()
return [self._product_serialize(r) for r in rows]
def upsert_product(
self,
*,
product_code: str,
product_name: str,
coupang_item_code: str | None = None,
sort_order: int = 0,
) -> dict[str, Any]:
"""제품 등록/수정.
coupang_item_code 가 None 이면 기존 값을 유지한다(빈 문자열은 지우기).
"""
code = (product_code or "").strip()
name = (product_name or "").strip()
cic = None if coupang_item_code is None else coupang_item_code.strip()
if not code or not name:
raise ValueError("제품코드와 제품명 모두 필요합니다.")
with self._pool.connection() as conn:
row = conn.execute(
"""
INSERT INTO cupang_products
(product_code, product_name, coupang_item_code, sort_order)
VALUES (%(code)s, %(name)s, COALESCE(%(cic)s, ''), %(sort)s)
ON CONFLICT (product_code) DO UPDATE
SET product_name = EXCLUDED.product_name,
coupang_item_code = COALESCE(
%(cic)s, cupang_products.coupang_item_code
),
sort_order = EXCLUDED.sort_order,
active = TRUE
RETURNING *
""",
{"code": code, "name": name, "cic": cic, "sort": sort_order},
).fetchone()
return self._product_serialize(row)
def set_product_active(self, *, product_id: int, active: bool) -> None:
with self._pool.connection() as conn:
cur = conn.execute(
"UPDATE cupang_products SET active = %s WHERE id = %s",
(bool(active), product_id),
)
if cur.rowcount == 0:
raise KeyError(product_id)
def update_product(
self,
*,
product_id: int,
product_name: str,
product_code: str,
coupang_item_code: str = "",
) -> dict[str, Any]:
"""등록된 제품 수정(제품명/제품코드/쿠팡상품코드).
제품코드가 바뀌면 같은 코드로 연결돼 있던 상자 규칙(cupang_box_rules)의
product_code 도 함께 옮긴다(출고 라인은 과거 기록이라 스냅샷 유지).
"""
name = (product_name or "").strip()
code = (product_code or "").strip()
cic = (coupang_item_code or "").strip()
if not name or not code:
raise ValueError("제품코드와 제품명 모두 필요합니다.")
with self._pool.connection() as conn:
with conn.transaction():
old = conn.execute(
"SELECT * FROM cupang_products WHERE id = %s FOR UPDATE",
(product_id,),
).fetchone()
if old is None:
raise KeyError(product_id)
old_code = (old["product_code"] or "").strip()
if code != old_code:
dup = conn.execute(
"SELECT 1 FROM cupang_products "
"WHERE product_code = %s AND id <> %s",
(code, product_id),
).fetchone()
if dup:
raise ValueError(f"제품코드 {code} 는 이미 등록돼 있습니다.")
row = conn.execute(
"""
UPDATE cupang_products
SET product_name = %s,
product_code = %s,
coupang_item_code = %s
WHERE id = %s
RETURNING *
""",
(name, code, cic, product_id),
).fetchone()
if code != old_code:
# 새 코드로 된 규칙이 이미 있으면 옮기지 않는다(중복 방지).
exists_new = conn.execute(
"SELECT 1 FROM cupang_box_rules WHERE product_code = %s",
(code,),
).fetchone()
if not exists_new:
conn.execute(
"UPDATE cupang_box_rules "
"SET product_code = %s, product_name_snapshot = %s "
"WHERE product_code = %s",
(code, name, old_code),
)
else:
conn.execute(
"UPDATE cupang_box_rules SET product_name_snapshot = %s "
"WHERE product_code = %s",
(name, code),
)
return self._product_serialize(row)
def toggle_product_active(self, *, product_id: int) -> dict[str, Any]:
"""활성 ↔ 비활성 뒤집기. 갱신된 행을 반환."""
with self._pool.connection() as conn:
row = conn.execute(
"UPDATE cupang_products SET active = NOT active "
"WHERE id = %s RETURNING *",
(product_id,),
).fetchone()
if row is None:
raise KeyError(product_id)
return self._product_serialize(row)
def delete_product(self, *, product_id: int) -> None:
with self._pool.connection() as conn:
cur = conn.execute(
"DELETE FROM cupang_products WHERE id = %s", (product_id,)
)
if cur.rowcount == 0:
raise KeyError(product_id)
# ════════════════════════════════════════════════════════════
# 출고 묶음 (cupang_shipments + cupang_shipment_lines)
# ════════════════════════════════════════════════════════════
def list_shipments(
self,
*,
year: int | None = None,
month: int | None = None,
date_from: str | None = None,
date_to: str | None = None,
) -> list[dict[str, Any]]:
"""헤더 목록(라인 제외). 달력/리스트 표시에 사용.
year+month 가 주어지면 작성일/출고일/센터입고일 중 하나라도 해당 월에
걸치는 묶음을 모두 포함한다(달력 표시용).
"""
clauses: list[str] = []
params: list[Any] = []
if year and month:
clauses.append(
"(date_trunc('month', document_date) = make_date(%s, %s, 1)"
" OR date_trunc('month', ship_date) = make_date(%s, %s, 1)"
" OR date_trunc('month', center_arrival_date) = make_date(%s, %s, 1))"
)
params.extend([year, month, year, month, year, month])
if date_from:
clauses.append("ship_date >= %s")
params.append(date_from)
if date_to:
clauses.append("ship_date <= %s")
params.append(date_to)
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
with self._pool.connection() as conn:
rows = conn.execute(
f"SELECT * FROM cupang_shipments {where} "
"ORDER BY ship_date ASC, id ASC",
params,
).fetchall()
return [self._shipment_serialize(r) for r in rows]
def get_shipment(self, *, shipment_id: int, with_lines: bool = True) -> dict[str, Any] | None:
with self._pool.connection() as conn:
row = conn.execute(
"SELECT * FROM cupang_shipments WHERE id = %s", (shipment_id,)
).fetchone()
if not row:
return None
head = self._shipment_serialize(row)
if with_lines:
line_rows = conn.execute(
"SELECT * FROM cupang_shipment_lines WHERE shipment_id = %s "
"ORDER BY line_no ASC",
(shipment_id,),
).fetchall()
head["lines"] = [self._line_serialize(r) for r in line_rows]
return head
def create_shipment(
self, *, created_by: str, header: dict[str, Any], lines: list[dict[str, Any]]
) -> dict[str, Any]:
h = self._normalize_header(header)
norm_lines = self._normalize_lines(lines)
if not norm_lines:
raise ValueError("품목 라인이 최소 1개 필요합니다.")
with self._pool.connection() as conn:
with conn.transaction():
row = conn.execute(
"""
INSERT INTO cupang_shipments
(created_by, document_date, ship_date,
center_arrival_date, center_id, center_name_snapshot,
ship_method, outbound_summary, worker, status, memo,
box_plan)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
RETURNING *
""",
(
created_by.lower().strip(),
h["document_date"],
h["ship_date"],
h["center_arrival_date"],
h["center_id"],
h["center_name_snapshot"],
h["ship_method"],
h["outbound_summary"],
h["worker"],
h["status"],
h["memo"],
Jsonb(h["box_plan"]),
),
).fetchone()
shipment_id = row["id"]
self._insert_lines(conn, shipment_id, norm_lines)
return self.get_shipment(shipment_id=shipment_id)
def update_shipment(
self, *, shipment_id: int, header: dict[str, Any], lines: list[dict[str, Any]]
) -> dict[str, Any]:
h = self._normalize_header(header)
norm_lines = self._normalize_lines(lines)
if not norm_lines:
raise ValueError("품목 라인이 최소 1개 필요합니다.")
with self._pool.connection() as conn:
with conn.transaction():
row = conn.execute(
"""
UPDATE cupang_shipments
SET document_date = %s, ship_date = %s,
center_arrival_date = %s, center_id = %s,
center_name_snapshot = %s, ship_method = %s,
outbound_summary = %s, worker = %s, memo = %s,
box_plan = %s
WHERE id = %s
RETURNING *
""",
(
h["document_date"],
h["ship_date"],
h["center_arrival_date"],
h["center_id"],
h["center_name_snapshot"],
h["ship_method"],
h["outbound_summary"],
h["worker"],
h["memo"],
Jsonb(h["box_plan"]),
shipment_id,
),
).fetchone()
if not row:
raise KeyError(shipment_id)
# 라인 전체 교체 (CASCADE 아님 — 명시적 삭제 후 재삽입)
conn.execute(
"DELETE FROM cupang_shipment_lines WHERE shipment_id = %s",
(shipment_id,),
)
self._insert_lines(conn, shipment_id, norm_lines)
return self.get_shipment(shipment_id=shipment_id)
def set_status(self, *, shipment_id: int, status: str) -> dict[str, Any]:
if status not in STATUSES:
raise ValueError(f"허용되지 않는 상태: {status}")
with self._pool.connection() as conn:
row = conn.execute(
"UPDATE cupang_shipments SET status = %s WHERE id = %s RETURNING *",
(status, shipment_id),
).fetchone()
if not row:
raise KeyError(shipment_id)
return self._shipment_serialize(row)
def soft_delete(self, *, shipment_id: int) -> dict[str, Any]:
"""운영 안전을 위한 기본 삭제 — status='취소'."""
return self.set_status(shipment_id=shipment_id, status="취소")
def hard_delete(self, *, shipment_id: int) -> None:
"""완전 삭제(라인은 ON DELETE CASCADE). 취소 처리로 충분하므로 기본 미사용."""
with self._pool.connection() as conn:
cur = conn.execute(
"DELETE FROM cupang_shipments WHERE id = %s", (shipment_id,)
)
if cur.rowcount == 0:
raise KeyError(shipment_id)
def _insert_lines(self, conn: Any, shipment_id: int, lines: list[dict[str, Any]]) -> None:
for idx, ln in enumerate(lines, start=1):
calc = compute_boxes(ln["quantity"], ln.get("units_per_box"))
conn.execute(
"""
INSERT INTO cupang_shipment_lines
(shipment_id, line_no, product_code, product_name_snapshot,
quantity, box_rule_id, units_per_box, calculated_boxes,
remainder_units, manual_box_text, memo)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
""",
(
shipment_id,
idx,
ln["product_code"],
ln["product_name_snapshot"],
ln["quantity"],
ln.get("box_rule_id"),
calc["units_per_box"],
calc["required_boxes"],
calc["remainder_units"],
ln.get("manual_box_text", ""),
ln.get("memo", ""),
),
)
# ════════════════════════════════════════════════════════════
# 달력 집계
# ════════════════════════════════════════════════════════════
def calendar_counts(self, *, year: int, month: int) -> dict[str, dict[str, int]]:
"""해당 월의 날짜별 작성/출고/입고 건수.
반환: { "YYYY-MM-DD": {"document": n, "ship": n, "arrival": n} }
취소 상태는 제외한다.
"""
first = (year, month)
out: dict[str, dict[str, int]] = {}
def _accumulate(rows: list[dict[str, Any]], key: str) -> None:
for r in rows:
d = r["d"]
ds = d.isoformat() if isinstance(d, date) else str(d)
out.setdefault(ds, {"document": 0, "ship": 0, "arrival": 0})
out[ds][key] = int(r["c"])
with self._pool.connection() as conn:
doc = conn.execute(
"SELECT document_date AS d, COUNT(*) AS c FROM cupang_shipments "
"WHERE status <> '취소' AND date_trunc('month', document_date) = make_date(%s,%s,1) "
"GROUP BY 1",
first,
).fetchall()
ship = conn.execute(
"SELECT ship_date AS d, COUNT(*) AS c FROM cupang_shipments "
"WHERE status <> '취소' AND date_trunc('month', ship_date) = make_date(%s,%s,1) "
"GROUP BY 1",
first,
).fetchall()
arr = conn.execute(
"SELECT center_arrival_date AS d, COUNT(*) AS c FROM cupang_shipments "
"WHERE status <> '취소' AND date_trunc('month', center_arrival_date) = make_date(%s,%s,1) "
"GROUP BY 1",
first,
).fetchall()
_accumulate(doc, "document")
_accumulate(ship, "ship")
_accumulate(arr, "arrival")
return out
# ════════════════════════════════════════════════════════════
# 쿠팡 로켓 매출 (cupang_sales / cupang_sales_weekly)
# 구글 시트 양식을 그대로 담는다. 주차·월 소계는 저장하지 않고 화면에서 합산.
# ════════════════════════════════════════════════════════════
SALES_FIELDS = (
"week_label", "seq", "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",
"stock_deduct_memo", "memo",
)
def list_sales(
self,
*,
date_from: str = "",
date_to: str = "",
center_name: str = "",
keyword: str = "",
po_type: str = "",
week_label: str = "",
limit: int = 5000,
) -> list[dict[str, Any]]:
"""출고일 기준 조회. 기간·센터·발주유형·주차·검색어(품목/SKU/발주번호/바코드)."""
where: list[str] = []
params: list[Any] = []
if date_from:
where.append("ship_date >= %s")
params.append(date_from)
if date_to:
where.append("ship_date <= %s")
params.append(date_to)
if center_name:
where.append("center_name = %s")
params.append(center_name)
if po_type:
where.append("po_type = %s")
params.append(po_type)
if week_label:
where.append("week_label = %s")
params.append(week_label)
if keyword:
where.append(
"(item_name ILIKE %s OR sku_id ILIKE %s OR po_no ILIKE %s OR barcode ILIKE %s)"
)
like = f"%{keyword}%"
params.extend([like, like, like, like])
clause = ("WHERE " + " AND ".join(where)) if where else ""
params.append(max(1, min(int(limit or 5000), 20000)))
with self._pool.connection() as conn:
rows = conn.execute(
f"SELECT * FROM cupang_sales {clause} "
"ORDER BY ship_date DESC NULLS LAST, seq ASC, id ASC LIMIT %s",
tuple(params),
).fetchall()
return [self._sales_serialize(r) for r in rows]
def get_sale(self, *, sale_id: int) -> dict[str, Any] | None:
with self._pool.connection() as conn:
row = conn.execute(
"SELECT * FROM cupang_sales WHERE id = %s", (sale_id,)
).fetchone()
return self._sales_serialize(row) if row else None
def sales_centers(self) -> list[str]:
"""필터용 — 매출 자료에 실제로 등장하는 입고센터 목록."""
with self._pool.connection() as conn:
rows = conn.execute(
"SELECT DISTINCT center_name FROM cupang_sales "
"WHERE center_name <> '' ORDER BY center_name"
).fetchall()
return [r["center_name"] for r in rows]
def sales_weeks(self) -> list[str]:
"""필터용 — 매출 자료의 주차 목록(최근 주차부터)."""
with self._pool.connection() as conn:
rows = conn.execute(
"SELECT week_label, MIN(ship_date) AS d FROM cupang_sales "
"WHERE week_label <> '' GROUP BY week_label ORDER BY d DESC NULLS LAST"
).fetchall()
return [r["week_label"] for r in rows]
def create_sale(self, *, data: dict[str, Any]) -> dict[str, Any]:
row = self._normalize_sale(data)
cols = ", ".join(self.SALES_FIELDS)
marks = ", ".join(f"%({f})s" for f in self.SALES_FIELDS)
with self._pool.connection() as conn:
created = conn.execute(
f"INSERT INTO cupang_sales ({cols}) VALUES ({marks}) RETURNING *", row
).fetchone()
return self._sales_serialize(created)
def update_sale(self, *, sale_id: int, data: dict[str, Any]) -> dict[str, Any]:
row = self._normalize_sale(data)
row["id"] = sale_id
sets = ", ".join(f"{f} = %({f})s" for f in self.SALES_FIELDS)
with self._pool.connection() as conn:
updated = conn.execute(
f"UPDATE cupang_sales SET {sets} WHERE id = %(id)s RETURNING *", row
).fetchone()
if not updated:
raise KeyError(sale_id)
return self._sales_serialize(updated)
def delete_sale(self, *, sale_id: int) -> None:
with self._pool.connection() as conn:
cur = conn.execute("DELETE FROM cupang_sales WHERE id = %s", (sale_id,))
if cur.rowcount == 0:
raise KeyError(sale_id)
def upsert_sales_bulk(self, rows: list[dict[str, Any]]) -> dict[str, int]:
"""엑셀 업로드용 — 같은 라인(발주번호+SKU+출고일+센터+수량)은 덮어쓴다."""
cols = ", ".join(self.SALES_FIELDS)
marks = ", ".join(f"%({f})s" for f in self.SALES_FIELDS)
updates = ", ".join(
f"{f} = EXCLUDED.{f}"
for f in self.SALES_FIELDS
if f not in ("po_no", "sku_id", "ship_date", "center_name", "quantity")
)
saved = 0
with self._pool.connection() as conn:
for raw in rows:
data = self._normalize_sale(raw)
conn.execute(
f"INSERT INTO cupang_sales ({cols}) VALUES ({marks}) "
"ON CONFLICT (po_no, sku_id, ship_date, center_name, quantity) "
f"DO UPDATE SET {updates}",
data,
)
saved += 1
return {"saved": saved}
# ── 주차 단위 비용 ──────────────────────────────────────
def list_sales_weekly(self) -> list[dict[str, Any]]:
with self._pool.connection() as conn:
rows = conn.execute(
"SELECT * FROM cupang_sales_weekly ORDER BY week_from ASC NULLS LAST, id ASC"
).fetchall()
return [self._weekly_serialize(r) for r in rows]
def upsert_sales_weekly(self, *, data: dict[str, Any]) -> dict[str, Any]:
row = {
"week_label": str(data.get("week_label") or "").strip(),
"week_from": self._date_or_none(data.get("week_from")),
"week_to": self._date_or_none(data.get("week_to")),
"ad_cost": self._num(data.get("ad_cost")),
"promo_discount": self._num(data.get("promo_discount")),
"incentive": self._num(data.get("incentive")),
"memo": str(data.get("memo") or "").strip(),
}
if not row["week_label"]:
raise ValueError("주차 이름이 필요합니다.")
with self._pool.connection() as conn:
saved = conn.execute(
"""
INSERT INTO cupang_sales_weekly
(week_label, week_from, week_to, ad_cost, promo_discount, incentive, memo)
VALUES
(%(week_label)s, %(week_from)s, %(week_to)s, %(ad_cost)s,
%(promo_discount)s, %(incentive)s, %(memo)s)
ON CONFLICT (week_label) DO UPDATE
SET week_from = EXCLUDED.week_from,
week_to = EXCLUDED.week_to,
ad_cost = EXCLUDED.ad_cost,
promo_discount = EXCLUDED.promo_discount,
incentive = EXCLUDED.incentive,
memo = EXCLUDED.memo
RETURNING *
""",
row,
).fetchone()
return self._weekly_serialize(saved)
def delete_sales_weekly(self, *, weekly_id: int) -> None:
with self._pool.connection() as conn:
cur = conn.execute("DELETE FROM cupang_sales_weekly WHERE id = %s", (weekly_id,))
if cur.rowcount == 0:
raise KeyError(weekly_id)
# ── 매출 정규화 / 직렬화 ────────────────────────────────
@staticmethod
def _num(value: Any) -> float:
"""숫자 칸 파싱 — 콤마/%/빈칸/"-"/#DIV/0! 을 모두 0 또는 숫자로."""
if value is None:
return 0.0
if isinstance(value, bool):
return 0.0
if isinstance(value, (int, float)):
return float(value)
text = str(value).strip().replace(",", "").replace("%", "").replace("\u00a0", "")
if text in ("", "-", "", "#DIV/0!", "#N/A", "#VALUE!", "#REF!"):
return 0.0
try:
return float(text)
except ValueError:
return 0.0
@staticmethod
def _date_or_none(value: Any) -> str | None:
""""2024. 8. 12" / "2024-08-12" / date → "YYYY-MM-DD". 못 읽으면 None."""
if value in (None, "", "-"):
return None
if isinstance(value, datetime):
return value.date().isoformat()
if isinstance(value, date):
return value.isoformat()
text = str(value).strip().replace(".", "-").replace("/", "-").replace(" ", "")
text = text.strip("-")
parts = [p for p in text.split("-") if p]
if len(parts) != 3:
return None
try:
return date(int(parts[0]), int(parts[1]), int(parts[2])).isoformat()
except (TypeError, ValueError):
return None
def _normalize_sale(self, data: dict[str, Any]) -> dict[str, Any]:
def _t(key: str, limit: int = 200) -> str:
return str(data.get(key) or "").strip()[:limit]
try:
seq = int(data.get("seq")) if str(data.get("seq") or "").strip() else None
except (TypeError, ValueError):
seq = None
return {
"week_label": _t("week_label", 60),
"seq": seq,
"po_no": _t("po_no", 40),
"po_type": _t("po_type", 30) or "일반",
"order_date": self._date_or_none(data.get("order_date")),
"ship_date": self._date_or_none(data.get("ship_date")),
"center_arrival_date": self._date_or_none(data.get("center_arrival_date")),
"sku_id": _t("sku_id", 40),
"barcode": _t("barcode", 40),
"item_name": _t("item_name", 120),
"quantity": int(self._num(data.get("quantity"))),
"center_name": _t("center_name", 40),
"supply_unit_price": self._num(data.get("supply_unit_price")),
"supply_amount": self._num(data.get("supply_amount")),
"cost_unit_price": self._num(data.get("cost_unit_price")),
"cost_amount": self._num(data.get("cost_amount")),
"picking_unit_price": self._num(data.get("picking_unit_price")),
"picking_amount": self._num(data.get("picking_amount")),
"milkrun_amount": self._num(data.get("milkrun_amount")),
"logistics_total": self._num(data.get("logistics_total")),
"logistics_ratio": self._num(data.get("logistics_ratio")),
"margin": self._num(data.get("margin")),
"margin_rate": self._num(data.get("margin_rate")),
"stock_deduct_memo": _t("stock_deduct_memo", 200),
"memo": _t("memo", 500),
}
@staticmethod
def _sales_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
for key in ("order_date", "ship_date", "center_arrival_date"):
v = out.get(key)
if isinstance(v, date):
out[key] = v.isoformat()
elif v is None:
out[key] = ""
for key in (
"supply_unit_price", "supply_amount", "cost_unit_price", "cost_amount",
"picking_unit_price", "picking_amount", "milkrun_amount",
"logistics_total", "logistics_ratio", "margin", "margin_rate",
):
if out.get(key) is not None:
out[key] = float(out[key])
for key in ("created_at", "updated_at"):
v = out.get(key)
if isinstance(v, datetime):
out[key] = v.astimezone(KST).isoformat()
return out
@staticmethod
def _weekly_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
for key in ("week_from", "week_to"):
v = out.get(key)
out[key] = v.isoformat() if isinstance(v, date) else ""
for key in ("ad_cost", "promo_discount", "incentive"):
if out.get(key) is not None:
out[key] = float(out[key])
for key in ("created_at", "updated_at"):
v = out.get(key)
if isinstance(v, datetime):
out[key] = v.astimezone(KST).isoformat()
return out
# ════════════════════════════════════════════════════════════
# 정규화 / 직렬화 helpers
# ════════════════════════════════════════════════════════════
@staticmethod
def _normalize_header(header: dict[str, Any]) -> dict[str, Any]:
def _d(key: str) -> str:
v = str(header.get(key) or "").strip()
return v
document_date = _d("document_date")
ship_date = _d("ship_date")
center_arrival_date = _d("center_arrival_date")
if not document_date or not ship_date or not center_arrival_date:
raise ValueError("작성일/출고일/센터입고일은 필수입니다.")
status = str(header.get("status") or "작성중").strip()
if status not in STATUSES:
status = "작성중"
center_id_raw = header.get("center_id")
try:
center_id = int(center_id_raw) if center_id_raw not in (None, "", "0") else None
except (TypeError, ValueError):
center_id = None
return {
"document_date": document_date,
"ship_date": ship_date,
"center_arrival_date": center_arrival_date,
"center_id": center_id,
"center_name_snapshot": str(header.get("center_name_snapshot") or "").strip(),
"ship_method": str(header.get("ship_method") or "택배").strip() or "택배",
"outbound_summary": str(header.get("outbound_summary") or "").strip(),
"worker": str(header.get("worker") or "").strip(),
"status": status,
"memo": str(header.get("memo") or "").strip(),
# 확정 당시 상자 구성(제품별/혼합). 화면 표시 전용이라 형태만 검사한다.
"box_plan": header.get("box_plan") if isinstance(header.get("box_plan"), list) else [],
}
@staticmethod
def _normalize_lines(lines: list[dict[str, Any]]) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for raw in lines or []:
code = str(raw.get("product_code") or "").strip()
name = str(raw.get("product_name_snapshot") or raw.get("product_name") or "").strip()
try:
qty = int(raw.get("quantity") or 0)
except (TypeError, ValueError):
qty = 0
if not code or qty <= 0:
continue # 빈 라인 스킵
upb_raw = raw.get("units_per_box")
try:
upb = int(upb_raw) if upb_raw not in (None, "", "0") else None
except (TypeError, ValueError):
upb = None
rule_id_raw = raw.get("box_rule_id")
try:
rule_id = int(rule_id_raw) if rule_id_raw not in (None, "") else None
except (TypeError, ValueError):
rule_id = None
out.append(
{
"product_code": code,
"product_name_snapshot": name or code,
"quantity": qty,
"units_per_box": upb,
"box_rule_id": rule_id,
"manual_box_text": str(raw.get("manual_box_text") or "").strip(),
"memo": str(raw.get("memo") or "").strip(),
}
)
return out
@staticmethod
def _center_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["active"] = bool(out.get("active", True))
out["sort_order"] = int(out.get("sort_order", 0))
out["coupang_item_code"] = (out.get("coupang_item_code") or "").strip()
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out
@staticmethod
def _rule_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["units_per_box"] = int(out.get("units_per_box", 0))
out["active"] = bool(out.get("active", True))
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out
@staticmethod
def _draft_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["title"] = str(out.get("title") or "")
if "payload" in out and not isinstance(out.get("payload"), dict):
out["payload"] = {}
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out
@staticmethod
def _product_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["active"] = bool(out.get("active", True))
out["sort_order"] = int(out.get("sort_order", 0))
out["coupang_item_code"] = (out.get("coupang_item_code") or "").strip()
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out
@staticmethod
def _shipment_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["center_id"] = int(out["center_id"]) if out.get("center_id") is not None else None
if not isinstance(out.get("box_plan"), list):
out["box_plan"] = []
for k in ("document_date", "ship_date", "center_arrival_date"):
v = out.get(k)
if isinstance(v, date):
out[k] = v.isoformat()
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out
@staticmethod
def _line_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
if row is None:
return None
out = dict(row)
out["id"] = int(out["id"])
out["line_no"] = int(out.get("line_no", 0))
out["quantity"] = int(out.get("quantity", 0))
for k in ("box_rule_id", "units_per_box", "calculated_boxes", "remainder_units"):
v = out.get(k)
out[k] = int(v) if v is not None else None
for k in ("created_at", "updated_at"):
v = out.get(k)
if isinstance(v, datetime):
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
return out