"""cupang_db PostgreSQL 저장소. - 드라이버: psycopg 3 (`psycopg[binary,pool]`) — expense 모듈과 동일 패턴. - 연결 정보: 환경변수 `CUPANG_DB_URL` (예: postgresql://cupang_app:@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