"""malaysia_stock_db PostgreSQL 저장소.

- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — cupang/expense 모듈과 동일 패턴.
- 연결 정보: 환경변수 `MALAYSIA_STOCK_DB_URL`
  (예: postgresql://malaysia_app:<pwd>@postgres-db:5432/malaysia_stock_db)
- 스키마(테이블/인덱스/트리거/seed)는 앱이 만들지 않는다.
  `scripts/sql/malaysia_stock_db_init.sql` 을 superuser 가 사전 적용한다.
  앱 계정(malaysia_app)은 SELECT/INSERT/UPDATE/DELETE 권한만 받는다.
- 연결 풀은 lazy open — 부팅 시 DB 가 잠시 끊겨도 컨테이너가 죽지 않게.

재고 계산 / BOM 분해의 순수 로직은 `store.py` 에 있고, 여기서는 DB I/O 와
조립만 담당한다(클라이언트 계산을 신뢰하지 않고 서버에서 재계산).
"""

from __future__ import annotations

from datetime import date, datetime
from typing import Any

from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

from app.timezone import KST, now_kst

from . import store


class MalaysiaStockStore:
    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()

    # ════════════════════════════════════════════════════════════
    # 창고 (warehouses)
    # ════════════════════════════════════════════════════════════
    def list_warehouses(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 warehouses {where} "
                "ORDER BY active DESC, warehouse_code ASC"
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def get_warehouse(self, *, warehouse_code: str) -> dict[str, Any] | None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT * FROM warehouses WHERE warehouse_code = %s",
                (warehouse_code,),
            ).fetchone()
        return self._serialize(row) if row else None

    def create_warehouse(
        self, *, warehouse_code: str, warehouse_name: str, country: str = "MY"
    ) -> dict[str, Any]:
        code = (warehouse_code or "").strip()
        name = (warehouse_name or "").strip()
        if not code or not name:
            raise ValueError("창고코드와 창고명은 필수입니다.")
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO warehouses (warehouse_code, warehouse_name, country)
                VALUES (%s, %s, %s)
                ON CONFLICT (warehouse_code) DO UPDATE
                   SET warehouse_name = EXCLUDED.warehouse_name,
                       country        = EXCLUDED.country,
                       active         = TRUE
                RETURNING *
                """,
                (code, name, (country or "MY").strip()),
            ).fetchone()
        return self._serialize(row)

    # ════════════════════════════════════════════════════════════
    # 재고관리 대상 아이템 (malaysia_items)
    # ════════════════════════════════════════════════════════════
    def list_items(
        self, *, kind: str | None = None, include_inactive: bool = False
    ) -> list[dict[str, Any]]:
        """kind: 'individual' | 'set' | None(전체)."""
        clauses: list[str] = []
        params: list[Any] = []
        if not include_inactive:
            clauses.append("active = TRUE")
        if kind in ("individual", "set"):
            clauses.append("kind = %s")
            params.append(kind)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT * FROM malaysia_items {where} "
                "ORDER BY sort_order ASC, item_code ASC",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def item_name_map(self) -> dict[str, str]:
        """item_code → item_name (이름 붙이기용)."""
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT item_code, item_name FROM malaysia_items"
            ).fetchall()
        return {r["item_code"]: (r["item_name"] or r["item_code"]) for r in rows}

    def upsert_item(
        self, *, item_code: str, item_name: str, kind: str, sort_order: int = 0
    ) -> dict[str, Any]:
        code = store._norm(item_code)
        name = (item_name or "").strip()
        if kind not in ("individual", "set"):
            raise ValueError("kind 는 'individual' 또는 'set' 이어야 합니다.")
        # prefix 와 kind 정합성 검증
        if kind == "individual" and not store.is_individual_code(code):
            raise ValueError(f"낱개 아이템은 MT-/MX-/MZ- 만 가능합니다: {code}")
        if kind == "set" and not store.is_set_code(code):
            raise ValueError(f"세트 아이템은 MY- 만 가능합니다: {code}")
        if store.is_blocked_code(code):
            raise ValueError(f"뚜껑 코드(MD-)는 등록할 수 없습니다: {code}")
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO malaysia_items (item_code, item_name, kind, sort_order)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (item_code) DO UPDATE
                   SET item_name  = EXCLUDED.item_name,
                       kind       = EXCLUDED.kind,
                       sort_order = EXCLUDED.sort_order,
                       active     = TRUE
                RETURNING *
                """,
                (code, name or code, kind, sort_order),
            ).fetchone()
        return self._serialize(row)

    def sync_item_names(self, name_lookup: dict[str, str]) -> int:
        """itemcode_db 등 외부 소스에서 받은 {code: name} 으로 이름 갱신.

        등록된 코드만 갱신(신규 추가 안 함). 반환: 갱신 건수.
        """
        updated = 0
        with self._pool.connection() as conn:
            for code, name in name_lookup.items():
                if not code or not name:
                    continue
                cur = conn.execute(
                    "UPDATE malaysia_items SET item_name = %s "
                    "WHERE item_code = %s AND item_name <> %s",
                    (name.strip(), store._norm(code), name.strip()),
                )
                updated += cur.rowcount or 0
        return updated

    # ════════════════════════════════════════════════════════════
    # 세트 구성표 (set_bom)
    # ════════════════════════════════════════════════════════════
    def list_bom(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 set_bom {where} "
                "ORDER BY set_code ASC, component_code ASC"
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def list_bom_for_set(self, *, set_code: str) -> list[dict[str, Any]]:
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM set_bom WHERE set_code = %s AND active = TRUE "
                "ORDER BY component_code ASC",
                (store._norm(set_code),),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def bom_map(self) -> dict[str, list[dict[str, Any]]]:
        """{ set_code: [{component_code, component_qty}, ...] } — active 만.

        store.explode_stocktake / explode_set 에 그대로 전달.
        """
        out: dict[str, list[dict[str, Any]]] = {}
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT set_code, component_code, component_qty "
                "FROM set_bom WHERE active = TRUE"
            ).fetchall()
        for r in rows:
            out.setdefault(r["set_code"], []).append(
                {
                    "component_code": r["component_code"],
                    "component_qty": int(r["component_qty"]),
                }
            )
        return out

    def upsert_bom_line(
        self, *, set_code: str, component_code: str, component_qty: int, active: bool = True
    ) -> dict[str, Any]:
        sc = store.validate_bom_set_code(set_code)
        cc = store.validate_bom_component_code(component_code)
        qty = store.validate_qty(component_qty, allow_zero=False)
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO set_bom (set_code, component_code, component_qty, active)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (set_code, component_code) DO UPDATE
                   SET component_qty = EXCLUDED.component_qty,
                       active        = EXCLUDED.active
                RETURNING *
                """,
                (sc, cc, qty, bool(active)),
            ).fetchone()
        return self._serialize(row)

    def delete_bom_line(self, *, bom_id: int) -> None:
        with self._pool.connection() as conn:
            cur = conn.execute("DELETE FROM set_bom WHERE id = %s", (bom_id,))
            if cur.rowcount == 0:
                raise KeyError(bom_id)

    # ════════════════════════════════════════════════════════════
    # 재고 이동 (stock_movement)
    # ════════════════════════════════════════════════════════════
    def create_movement(
        self,
        *,
        movement_date: str,
        warehouse_code: str,
        item_code: str,
        movement_type: str,
        qty: int,
        ref_type: str = "",
        ref_no: str = "",
        memo: str = "",
        created_by: str = "",
    ) -> dict[str, Any]:
        """단일 낱개 아이템 movement 1건 생성. 세트 코드는 차단."""
        if movement_type not in store.MOVEMENT_TYPES:
            raise ValueError(f"허용되지 않는 이동 종류: {movement_type}")
        code = store.validate_movement_code(item_code)  # MY-/MD- 차단
        # IN/OUT 은 양수, ADJUST/STOCKTAKE 는 ± 허용
        if movement_type in ("IN", "OUT"):
            n = store.validate_qty(qty, allow_zero=False, allow_negative=False)
        else:
            n = store.validate_qty(qty, allow_zero=True, allow_negative=True)
        if not (movement_date or "").strip():
            raise ValueError("이동일자는 필수입니다.")
        with self._pool.connection() as conn:
            row = conn.execute(
                """
                INSERT INTO stock_movement
                    (movement_date, warehouse_code, item_code, movement_type,
                     qty, ref_type, ref_no, memo, created_by)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
                RETURNING *
                """,
                (
                    movement_date.strip(),
                    warehouse_code.strip(),
                    code,
                    movement_type,
                    n,
                    (ref_type or "").strip(),
                    (ref_no or "").strip(),
                    (memo or "").strip(),
                    (created_by or "").lower().strip(),
                ),
            ).fetchone()
        return self._serialize(row)

    def create_set_out(
        self,
        *,
        movement_date: str,
        warehouse_code: str,
        set_code: str,
        set_qty: int,
        ref_type: str = "",
        ref_no: str = "",
        memo: str = "",
        created_by: str = "",
    ) -> list[dict[str, Any]]:
        """세트 주문 출고 → set_bom 으로 분해해 구성품별 OUT movement 생성.

        예: MY-0002 1세트 출고 → MT-0320, MT-0550 ... 각각 OUT.
        BOM 이 없으면 ValueError(요구사항 8).
        """
        sc = store.validate_bom_set_code(set_code)
        q = store.validate_qty(set_qty, allow_zero=False)
        exploded = store.explode_set(sc, q, self.bom_map())
        if not exploded:
            raise ValueError(f"세트 {sc} 의 BOM 구성이 없습니다. 세트구성을 먼저 등록하세요.")
        ref_t = (ref_type or "set").strip() or "set"
        results: list[dict[str, Any]] = []
        with self._pool.connection() as conn:
            with conn.transaction():
                for comp_code, comp_qty in exploded.items():
                    row = conn.execute(
                        """
                        INSERT INTO stock_movement
                            (movement_date, warehouse_code, item_code, movement_type,
                             qty, ref_type, ref_no, memo, created_by)
                        VALUES (%s,%s,%s,'OUT',%s,%s,%s,%s,%s)
                        RETURNING *
                        """,
                        (
                            movement_date.strip(),
                            warehouse_code.strip(),
                            comp_code,
                            comp_qty,
                            ref_t,
                            (ref_no or "").strip(),
                            (memo or f"{sc} {q}세트 분해").strip(),
                            (created_by or "").lower().strip(),
                        ),
                    ).fetchone()
                    results.append(self._serialize(row))
        return results

    def list_movements(
        self,
        *,
        warehouse_code: str | None = None,
        item_code: str | None = None,
        movement_type: str | None = None,
        date_from: str | None = None,
        date_to: str | None = None,
        limit: int = 500,
    ) -> list[dict[str, Any]]:
        clauses: list[str] = []
        params: list[Any] = []
        if warehouse_code:
            clauses.append("warehouse_code = %s")
            params.append(warehouse_code)
        if item_code:
            clauses.append("item_code = %s")
            params.append(store._norm(item_code))
        if movement_type in store.MOVEMENT_TYPES:
            clauses.append("movement_type = %s")
            params.append(movement_type)
        if date_from:
            clauses.append("movement_date >= %s")
            params.append(date_from)
        if date_to:
            clauses.append("movement_date <= %s")
            params.append(date_to)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        params.append(int(limit))
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT * FROM stock_movement {where} "
                "ORDER BY movement_date DESC, id DESC LIMIT %s",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def current_stock(self, *, warehouse_code: str) -> dict[str, int]:
        """창고별 현재고(낱개 아이템 기준).

        현재고 = SUM(IN) - SUM(OUT) + SUM(ADJUST) + SUM(STOCKTAKE)
        반환: { item_code: qty }
        """
        with self._pool.connection() as conn:
            rows = conn.execute(
                """
                SELECT item_code,
                       SUM(CASE
                             WHEN movement_type = 'IN'  THEN qty
                             WHEN movement_type = 'OUT' THEN -qty
                             ELSE qty                       -- ADJUST/STOCKTAKE 는 부호 포함 저장
                           END) AS qty
                FROM stock_movement
                WHERE warehouse_code = %s
                GROUP BY item_code
                """,
                (warehouse_code,),
            ).fetchall()
        return {r["item_code"]: int(r["qty"] or 0) for r in rows}

    def stock_status(self, *, warehouse_code: str) -> list[dict[str, Any]]:
        """재고 현황 — 등록된 낱개 아이템 전체 + 현재고 + 마지막 조사일.

        등록 아이템 기준으로 0 재고도 표시(좌측 조인 효과).
        """
        stock = self.current_stock(warehouse_code=warehouse_code)
        last_dates = self._last_stocktake_dates(warehouse_code=warehouse_code)
        items = self.list_items(kind="individual")
        out: list[dict[str, Any]] = []
        for it in items:
            code = it["item_code"]
            out.append(
                {
                    "item_code": code,
                    "item_name": it.get("item_name") or code,
                    "current_qty": stock.get(code, 0),
                    "last_stocktake_date": last_dates.get(code),
                }
            )
        # 등록 아이템에 없지만 movement 가 있는 코드도 노출(누락 방지)
        for code, qty in stock.items():
            if code not in {i["item_code"] for i in out}:
                out.append(
                    {
                        "item_code": code,
                        "item_name": code,
                        "current_qty": qty,
                        "last_stocktake_date": last_dates.get(code),
                    }
                )
        out.sort(key=lambda x: x["item_code"])
        return out

    def _last_stocktake_dates(self, *, warehouse_code: str) -> dict[str, str]:
        """아이템별 마지막 STOCKTAKE movement 날짜."""
        with self._pool.connection() as conn:
            rows = conn.execute(
                """
                SELECT item_code, MAX(movement_date) AS d
                FROM stock_movement
                WHERE warehouse_code = %s AND movement_type = 'STOCKTAKE'
                GROUP BY item_code
                """,
                (warehouse_code,),
            ).fetchall()
        out: dict[str, str] = {}
        for r in rows:
            d = r["d"]
            out[r["item_code"]] = d.isoformat() if isinstance(d, date) else str(d)
        return out

    # ════════════════════════════════════════════════════════════
    # 일일 재고조사 (daily_stocktake + daily_stocktake_line)
    # ════════════════════════════════════════════════════════════
    def list_stocktakes(
        self, *, warehouse_code: str | None = None, limit: int = 200
    ) -> list[dict[str, Any]]:
        clauses: list[str] = []
        params: list[Any] = []
        if warehouse_code:
            clauses.append("warehouse_code = %s")
            params.append(warehouse_code)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        params.append(int(limit))
        with self._pool.connection() as conn:
            rows = conn.execute(
                f"SELECT * FROM daily_stocktake {where} "
                "ORDER BY stocktake_date DESC, id DESC LIMIT %s",
                params,
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def get_stocktake(
        self, *, stocktake_id: int, with_lines: bool = True
    ) -> dict[str, Any] | None:
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT * FROM daily_stocktake WHERE id = %s", (stocktake_id,)
            ).fetchone()
            if not row:
                return None
            head = self._serialize(row)
            if with_lines:
                line_rows = conn.execute(
                    "SELECT * FROM daily_stocktake_line WHERE stocktake_id = %s "
                    "ORDER BY sku_code ASC",
                    (stocktake_id,),
                ).fetchall()
                head["lines"] = [self._serialize(r) for r in line_rows]
        return head

    def create_stocktake(
        self,
        *,
        stocktake_date: str,
        warehouse_code: str,
        memo: str = "",
        created_by: str = "",
    ) -> dict[str, Any]:
        if not (stocktake_date or "").strip():
            raise ValueError("조사 날짜는 필수입니다.")
        if not (warehouse_code or "").strip():
            raise ValueError("창고는 필수입니다.")
        # 같은 날짜+창고에 이미 finalized 가 있으면 신규 생성 차단(요구사항 8)
        with self._pool.connection() as conn:
            dup = conn.execute(
                "SELECT 1 FROM daily_stocktake "
                "WHERE stocktake_date = %s AND warehouse_code = %s AND status = 'finalized' "
                "LIMIT 1",
                (stocktake_date.strip(), warehouse_code.strip()),
            ).fetchone()
            if dup:
                raise ValueError(
                    f"{stocktake_date} / {warehouse_code} 에 이미 확정된 재고조사가 있습니다."
                )
            row = conn.execute(
                """
                INSERT INTO daily_stocktake
                    (stocktake_date, warehouse_code, status, memo, created_by)
                VALUES (%s, %s, 'draft', %s, %s)
                RETURNING *
                """,
                (
                    stocktake_date.strip(),
                    warehouse_code.strip(),
                    (memo or "").strip(),
                    (created_by or "").lower().strip(),
                ),
            ).fetchone()
        return self._serialize(row)

    def _assert_draft(self, conn: Any, stocktake_id: int) -> dict[str, Any]:
        row = conn.execute(
            "SELECT * FROM daily_stocktake WHERE id = %s", (stocktake_id,)
        ).fetchone()
        if not row:
            raise KeyError(stocktake_id)
        if row["status"] != "draft":
            raise ValueError("확정/취소된 재고조사는 라인을 수정할 수 없습니다.")
        return row

    def upsert_line(
        self, *, stocktake_id: int, sku_code: str, qty: int, memo: str = ""
    ) -> dict[str, Any]:
        code = store.validate_stocktake_code(sku_code)  # MD- 차단, MY/낱개 허용
        n = store.validate_qty(qty, allow_zero=True, allow_negative=False)
        with self._pool.connection() as conn:
            self._assert_draft(conn, stocktake_id)
            row = conn.execute(
                """
                INSERT INTO daily_stocktake_line (stocktake_id, sku_code, qty, memo)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (stocktake_id, sku_code) DO UPDATE
                   SET qty = EXCLUDED.qty, memo = EXCLUDED.memo
                RETURNING *
                """,
                (stocktake_id, code, n, (memo or "").strip()),
            ).fetchone()
        return self._serialize(row)

    def delete_line(self, *, stocktake_id: int, line_id: int) -> None:
        with self._pool.connection() as conn:
            self._assert_draft(conn, stocktake_id)
            cur = conn.execute(
                "DELETE FROM daily_stocktake_line WHERE id = %s AND stocktake_id = %s",
                (line_id, stocktake_id),
            )
            if cur.rowcount == 0:
                raise KeyError(line_id)

    def cancel_stocktake(self, *, stocktake_id: int) -> dict[str, Any]:
        with self._pool.connection() as conn:
            row = conn.execute(
                "UPDATE daily_stocktake SET status = 'cancelled' "
                "WHERE id = %s AND status <> 'finalized' RETURNING *",
                (stocktake_id,),
            ).fetchone()
        if not row:
            raise ValueError("확정된 조사는 취소할 수 없거나, 대상이 없습니다.")
        return self._serialize(row)

    def compute_result(self, *, stocktake_id: int) -> dict[str, Any]:
        """재고조사 최종 계산 결과(낱개 아이템 기준).

        반환:
          {
            "stocktake": {...헤더...},
            "rows": [{item_code,item_name,direct_qty,from_set_qty,total_qty}, ...],
            "missing_bom": [MY-... BOM 없는 세트],
          }
        """
        head = self.get_stocktake(stocktake_id=stocktake_id, with_lines=True)
        if not head:
            raise KeyError(stocktake_id)
        lines = head.get("lines", [])
        bmap = self.bom_map()
        exploded = store.explode_stocktake(lines, bmap)
        missing = store.missing_bom_sets(lines, bmap)
        names = self.item_name_map()
        rows = []
        for code in sorted(exploded.keys()):
            v = exploded[code]
            rows.append(
                {
                    "stocktake_id": stocktake_id,
                    "stocktake_date": head["stocktake_date"],
                    "warehouse_code": head["warehouse_code"],
                    "item_code": code,
                    "item_name": names.get(code, code),
                    "direct_qty": v["direct_qty"],
                    "from_set_qty": v["from_set_qty"],
                    "total_qty": v["total_qty"],
                }
            )
        return {"stocktake": head, "rows": rows, "missing_bom": missing}

    def finalize_stocktake(self, *, stocktake_id: int) -> dict[str, Any]:
        """재고조사 확정 — 최종 낱개 total_qty 와 현재 시스템 재고의 '차이'만
        STOCKTAKE movement 로 생성한다(요구사항 7).

        - 차이 = total_qty - current_system_qty → qty 부호 그대로 STOCKTAKE 저장
        - movement 이력이 남고, 현재고 계산은 단순 합산으로 유지된다.
        - BOM 없는 MY- 세트가 있으면 확정 거부(요구사항 8).
        반환: {"stocktake": {...}, "adjustments": [{item_code, before, after, diff}], "rows": [...]}
        """
        computed = self.compute_result(stocktake_id=stocktake_id)
        head = computed["stocktake"]
        if head["status"] != "draft":
            raise ValueError("이미 확정/취소된 재고조사입니다.")
        if computed["missing_bom"]:
            raise ValueError(
                "BOM 구성이 없는 세트가 있어 확정할 수 없습니다: "
                + ", ".join(computed["missing_bom"])
            )
        wh = head["warehouse_code"]
        st_date = head["stocktake_date"]
        current = self.current_stock(warehouse_code=wh)
        rows = computed["rows"]
        adjustments: list[dict[str, Any]] = []
        with self._pool.connection() as conn:
            with conn.transaction():
                for r in rows:
                    code = r["item_code"]
                    before = current.get(code, 0)
                    after = r["total_qty"]
                    diff = after - before
                    if diff != 0:
                        conn.execute(
                            """
                            INSERT INTO stock_movement
                                (movement_date, warehouse_code, item_code, movement_type,
                                 qty, ref_type, ref_no, memo, created_by)
                            VALUES (%s,%s,%s,'STOCKTAKE',%s,'stocktake',%s,%s,%s)
                            """,
                            (
                                st_date,
                                wh,
                                code,
                                diff,
                                str(stocktake_id),
                                f"재고조사 #{stocktake_id} 반영(차이 {diff:+d})",
                                head.get("created_by", ""),
                            ),
                        )
                    adjustments.append(
                        {"item_code": code, "before": before, "after": after, "diff": diff}
                    )
                conn.execute(
                    "UPDATE daily_stocktake SET status = 'finalized', finalized_at = %s "
                    "WHERE id = %s",
                    (now_kst(), stocktake_id),
                )
        head = self.get_stocktake(stocktake_id=stocktake_id, with_lines=True)
        return {"stocktake": head, "adjustments": adjustments, "rows": rows}

    # ════════════════════════════════════════════════════════════
    # 직렬화
    # ════════════════════════════════════════════════════════════
    @staticmethod
    def _serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
        if row is None:
            return None
        out = dict(row)
        for k, v in list(out.items()):
            if isinstance(v, datetime):
                out[k] = v.astimezone(KST).isoformat(timespec="seconds")
            elif isinstance(v, date):
                out[k] = v.isoformat()
        for k in ("id", "stocktake_id", "qty", "component_qty", "sort_order", "line_no"):
            if k in out and out[k] is not None:
                try:
                    out[k] = int(out[k])
                except (TypeError, ValueError):
                    pass
        if "active" in out:
            out["active"] = bool(out["active"])
        return out