dbx-main/app/modules/cupang/db.py

"""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

    # ════════════════════════════════════════════════════════════
    # 정규화 / 직렬화 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