"""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_pool import ConnectionPool 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 get_box_rule(self, *, product_code: str) -> dict[str, Any] | None: code = (product_code or "").strip() if not code: return None with self._pool.connection() as conn: row = conn.execute( "SELECT * FROM cupang_box_rules WHERE product_code = %s AND active = TRUE", (code,), ).fetchone() return self._rule_serialize(row) if row else None 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 deactivate_box_rule(self, *, rule_id: int) -> None: with self._pool.connection() as conn: cur = conn.execute( "UPDATE cupang_box_rules SET active = FALSE WHERE id = %s", (rule_id,), ) if cur.rowcount == 0: raise KeyError(rule_id) 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) def box_rule_map(self) -> dict[str, dict[str, Any]]: """product_code → rule. 폼/계산에서 빠르게 조회하기 위한 dict.""" return {r["product_code"]: r for r in self.list_box_rules()} # ════════════════════════════════════════════════════════════ # 제품명 카탈로그 (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, sort_order: int = 0 ) -> dict[str, Any]: code = (product_code or "").strip() name = (product_name or "").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, sort_order) VALUES (%s, %s, %s) ON CONFLICT (product_code) DO UPDATE SET product_name = EXCLUDED.product_name, sort_order = EXCLUDED.sort_order, active = TRUE RETURNING * """, (code, name, 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 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) VALUES (%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"], ), ).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 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"], 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 def shipments_for_export( self, *, year: int | None = None, month: int | None = None, shipment_id: int | None = None ) -> list[dict[str, Any]]: """엑셀 내보내기용 — 헤더+라인 평탄화 전 단계. 라인 포함 묶음 리스트 반환.""" if shipment_id is not None: s = self.get_shipment(shipment_id=shipment_id) return [s] if s else [] heads = self.list_shipments(year=year, month=month) result = [] for h in heads: full = self.get_shipment(shipment_id=h["id"]) if full: result.append(full) return result # ════════════════════════════════════════════════════════════ # 정규화 / 직렬화 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(), } @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)) for k in ("created_at", "updated_at"): v = out.get(k) if isinstance(v, datetime): out[k] = v.astimezone(timezone.utc).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(timezone.utc).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)) for k in ("created_at", "updated_at"): v = out.get(k) if isinstance(v, datetime): out[k] = v.astimezone(timezone.utc).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 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(timezone.utc).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(timezone.utc).isoformat(timespec="seconds") return out