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

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

    # ════════════════════════════════════════════════════════════
    # 세트 구성표 (BOM)
    #   별도 테이블/메뉴를 두지 않고 itemcode_db(set_components)에서 읽는다.
    #   라우터가 MalaysiaItemcodeReader.set_bom_map() 결과를 아래 메서드들에
    #   bom_map 인자로 주입한다(cross-DB 조인 회피, store 는 DB I/O 만 담당).
    # ════════════════════════════════════════════════════════════

    # ════════════════════════════════════════════════════════════
    # 재고 이동 (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,
        bom_map: dict[str, list[dict[str, Any]]],
        ref_type: str = "",
        ref_no: str = "",
        memo: str = "",
        created_by: str = "",
    ) -> list[dict[str, Any]]:
        """세트 주문 출고 → BOM(itemcode_db)으로 분해해 구성품별 OUT movement 생성.

        예: MY-0002 1세트 출고 → MT-0320, MT-0550 ... 각각 OUT.
        콤보는 입고 불가(낱개로만 입고). 출고만 BOM 분해한다.
        bom_map 은 라우터가 itemcode 리더에서 받아 주입. 구성 없으면 ValueError.
        """
        sc = store.validate_bom_set_code(set_code)
        q = store.validate_qty(set_qty, allow_zero=False)
        exploded = store.explode_set(sc, q, 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 daily_movement_totals(
        self, *, warehouse_code: str, movement_date: str, movement_type: str
    ) -> dict[str, int]:
        """특정 일자/종류의 낱개 아이템별 합계 수량.

        세트 OUT 은 생성 시 이미 구성품 OUT 으로 분해 저장되므로 그대로 합산된다.
        반환: { item_code: 합계수량 }
        """
        if movement_type not in store.MOVEMENT_TYPES:
            raise ValueError(f"허용되지 않는 이동 종류: {movement_type}")
        with self._pool.connection() as conn:
            rows = conn.execute(
                """
                SELECT item_code, SUM(qty) AS qty
                FROM stock_movement
                WHERE warehouse_code = %s
                  AND movement_date = %s
                  AND movement_type = %s
                GROUP BY item_code
                ORDER BY item_code
                """,
                (warehouse_code, movement_date, movement_type),
            ).fetchall()
        return {r["item_code"]: int(r["qty"] or 0) 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 latest_stocktake(self, *, warehouse_code: str) -> dict[str, Any] | None:
        """가장 최근(취소 제외) 재고조사 헤더. 없으면 None."""
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT * FROM daily_stocktake "
                "WHERE warehouse_code = %s AND status <> 'cancelled' "
                "ORDER BY stocktake_date DESC, id DESC LIMIT 1",
                (warehouse_code,),
            ).fetchone()
        return self._serialize(row) if row else None

    def latest_rack_stocktake(self, *, warehouse_code: str) -> dict[str, Any] | None:
        """랙 입력을 가장 최근에 저장한(취소 제외) 재고조사 헤더. 없으면 None.

        랙보기 전용. '최근'을 조사 날짜가 아니라 랙 저장 시각으로 판단한다.
        replace_rack 은 매 저장마다 daily_stocktake_rack 을 delete+insert 하므로
        그 created_at MAX 가 곧 마지막 랙 저장 시각이 된다. 랙 항목이 없는
        조사는 보여줄 게 없으므로 자연히 제외된다.
        """
        with self._pool.connection() as conn:
            row = conn.execute(
                "SELECT s.* FROM daily_stocktake s "
                "JOIN daily_stocktake_rack r ON r.stocktake_id = s.id "
                "WHERE s.warehouse_code = %s AND s.status <> 'cancelled' "
                "GROUP BY s.id "
                "ORDER BY MAX(r.created_at) DESC, s.id DESC LIMIT 1",
                (warehouse_code,),
            ).fetchone()
        return self._serialize(row) if row else None

    def previous_rack_entries(
        self, *, warehouse_code: str, exclude_stocktake_id: int
    ) -> list[dict[str, Any]]:
        """직전(현재 조사 제외, 취소 제외, 랙 입력 있는) 재고조사의 랙 항목.

        '이전값 불러오기' 버튼용. 조사 날짜 기준 가장 최근 다른 조사를 고른다.
        replace_rack 의 entries 형식(cell_code/sku_code/units_per_box/box_count)과
        호환되는 list_rack_entries 결과를 그대로 반환. 없으면 [].
        """
        with self._pool.connection() as conn:
            head = conn.execute(
                "SELECT s.id FROM daily_stocktake s "
                "JOIN daily_stocktake_rack r ON r.stocktake_id = s.id "
                "WHERE s.warehouse_code = %s AND s.id <> %s AND s.status <> 'cancelled' "
                "GROUP BY s.id "
                "ORDER BY s.stocktake_date DESC, s.id DESC LIMIT 1",
                (warehouse_code, exclude_stocktake_id),
            ).fetchone()
        if not head:
            return []
        return self.list_rack_entries(stocktake_id=head["id"])

    def delete_stocktake(self, *, stocktake_id: int) -> None:
        """재고조사 완전 삭제(헤더+라인 CASCADE). 슈퍼관리자 전용 동작.

        주의: finalize 시 생성된 STOCKTAKE movement 는 별도이므로 같이 지우지 않는다
        (재고 이력 보존). 헤더/라인만 제거.
        """
        with self._pool.connection() as conn:
            cur = conn.execute("DELETE FROM daily_stocktake WHERE id = %s", (stocktake_id,))
            if cur.rowcount == 0:
                raise KeyError(stocktake_id)

    def latest_set_quantities(self, *, warehouse_code: str) -> dict[str, Any]:
        """가장 최근(취소 제외) 재고조사에 입력된 콤보(MY-) 수량.

        반환: {"stocktake_date": str|None, "rows": [{set_code,set_name,qty}]}
        재고현황 우측 패널용 — 세트 단위로 입력된 콤보 재고를 그대로 보여줌.
        """
        with self._pool.connection() as conn:
            head = conn.execute(
                "SELECT id, stocktake_date FROM daily_stocktake "
                "WHERE warehouse_code = %s AND status <> 'cancelled' "
                "ORDER BY stocktake_date DESC, id DESC LIMIT 1",
                (warehouse_code,),
            ).fetchone()
            if not head:
                return {"stocktake_date": None, "rows": []}
            lines = conn.execute(
                "SELECT sku_code, qty FROM daily_stocktake_line "
                "WHERE stocktake_id = %s AND sku_code LIKE 'MY-%%' "
                "ORDER BY sku_code ASC",
                (head["id"],),
            ).fetchall()
        names = self.item_name_map()
        d = head["stocktake_date"]
        date_str = d.isoformat() if isinstance(d, date) else str(d)
        rows = [
            {
                "set_code": r["sku_code"],
                "set_name": names.get(r["sku_code"], r["sku_code"]),
                "qty": int(r["qty"]),
            }
            for r in lines
        ]
        return {"stocktake_date": date_str, "rows": rows}

    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)

    # ════════════════════════════════════════════════════════════
    # 재고조사 랙(rack) 입력 (daily_stocktake_rack)
    #   랙 칸 단위로 SKU×입수량×박스수 를 저장하고, 그 집계로
    #   daily_stocktake_line(SKU별 qty)을 재생성한다(랙이 라인의 단일 출처).
    # ════════════════════════════════════════════════════════════
    def list_rack_entries(self, *, stocktake_id: int) -> list[dict[str, Any]]:
        """재고조사의 랙 입력 항목 전체. 칸→순서대로 정렬."""
        with self._pool.connection() as conn:
            rows = conn.execute(
                "SELECT * FROM daily_stocktake_rack WHERE stocktake_id = %s "
                "ORDER BY cell_code ASC, line_no ASC, id ASC",
                (stocktake_id,),
            ).fetchall()
        return [self._serialize(r) for r in rows]

    def replace_rack(
        self, *, stocktake_id: int, entries: list[dict[str, Any]]
    ) -> dict[str, Any]:
        """랙 입력 전체 교체 → SKU별 집계로 재고조사 라인 재생성.

        entries: [{"cell_code","sku_code","units_per_box","box_count"}, ...]
          (입력 순서가 칸 안 line_no 가 된다.)
        한 트랜잭션:
          1) 기존 랙 항목 + 라인 전체 삭제
          2) 검증한 랙 항목 삽입(칸별 line_no 부여)
          3) aggregate_rack 결과를 daily_stocktake_line 으로 삽입
        draft 가 아니면 ValueError. 알 수 없는 칸/코드도 ValueError.
        """
        valid_cells = set(store.rack_cell_codes())
        clean: list[tuple[str, str, int, int]] = []
        for e in entries:
            cell = (e.get("cell_code") or "").strip().upper()
            if cell not in valid_cells:
                raise ValueError(f"알 수 없는 랙 위치: {cell or '(빈값)'}")
            code = store.validate_stocktake_code(str(e.get("sku_code") or ""))
            upb = store.validate_qty(
                e.get("units_per_box"), allow_zero=False, allow_negative=False
            )
            box = store.validate_qty(
                e.get("box_count"), allow_zero=False, allow_negative=False
            )
            clean.append((cell, code, upb, box))

        totals = store.aggregate_rack(
            [
                {"sku_code": c, "units_per_box": u, "box_count": b}
                for (_cell, c, u, b) in clean
            ]
        )

        cell_seq: dict[str, int] = {}
        with self._pool.connection() as conn:
            with conn.transaction():
                self._assert_draft(conn, stocktake_id)
                conn.execute(
                    "DELETE FROM daily_stocktake_rack WHERE stocktake_id = %s",
                    (stocktake_id,),
                )
                conn.execute(
                    "DELETE FROM daily_stocktake_line WHERE stocktake_id = %s",
                    (stocktake_id,),
                )
                for cell, code, upb, box in clean:
                    ln = cell_seq.get(cell, 0)
                    cell_seq[cell] = ln + 1
                    conn.execute(
                        """
                        INSERT INTO daily_stocktake_rack
                            (stocktake_id, cell_code, line_no, sku_code,
                             units_per_box, box_count)
                        VALUES (%s,%s,%s,%s,%s,%s)
                        """,
                        (stocktake_id, cell, ln, code, upb, box),
                    )
                for code in sorted(totals.keys()):
                    qty = totals[code]
                    if qty <= 0:
                        continue
                    conn.execute(
                        """
                        INSERT INTO daily_stocktake_line (stocktake_id, sku_code, qty)
                        VALUES (%s,%s,%s)
                        """,
                        (stocktake_id, code, qty),
                    )
        return {"entries": len(clean), "lines": {k: v for k, v in totals.items() if v > 0}}

    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, bom_map: dict[str, list[dict[str, Any]]]
    ) -> dict[str, Any]:
        """재고조사 최종 계산 결과(낱개 아이템 기준).

        bom_map 은 라우터가 itemcode 리더에서 받아 주입.
        반환:
          {
            "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", [])
        exploded = store.explode_stocktake(lines, bom_map)
        missing = store.missing_bom_sets(lines, bom_map)
        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, bom_map: dict[str, list[dict[str, Any]]]
    ) -> 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, bom_map=bom_map)
        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}

    # ════════════════════════════════════════════════════════════
    # 데이터 초기화 (슈퍼관리자 전용 — 라우터에서 권한 검사)
    # ════════════════════════════════════════════════════════════
    def reset_all_data(self) -> dict[str, int]:
        """전체 창고의 재고 수량 데이터 초기화.

        삭제 대상: stock_movement(입고/출고/조정/조사반영) +
                   daily_stocktake/daily_stocktake_line(재고조사).
        보존 대상: warehouses, malaysia_items(마스터).
        한 트랜잭션으로 처리. 반환: 삭제 건수.
        """
        with self._pool.connection() as conn:
            with conn.transaction():
                mv = conn.execute("DELETE FROM stock_movement").rowcount or 0
                ln = conn.execute("DELETE FROM daily_stocktake_line").rowcount or 0
                hd = conn.execute("DELETE FROM daily_stocktake").rowcount or 0
        return {"movements": mv, "stocktake_lines": ln, "stocktakes": hd}

    # ════════════════════════════════════════════════════════════
    # 직렬화
    # ════════════════════════════════════════════════════════════
    @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