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

"""itemcode_db 읽기 전용 — 세트 BOM(set_components) 조회.

말레이시아 재고관리는 세트 구성(어느 세트에 어떤 낱개가 몇 개)을 별도로
관리하지 않고 itemcode_db 에서 직접 읽는다.

itemcode_db 실제 스키마(운영 확인됨):
  set_items(item_code, sabangnet_code, name)            -- 세트 마스터
  single_items(item_code, sabangnet_code, name)         -- 낱개 마스터
  set_components(set_code, single_code, quantity)        -- 세트 ↔ 낱개 구성(BOM)
      FK: set_code → set_items, single_code → single_items

환경변수 `ITEMCODE_DB_URL`(읽기 전용 역할 itemcode_ro). cupang 모듈과 공유.
미설정/실패 시 enabled=False, 빈 결과(앱/모듈을 죽이지 않음).
"""

from __future__ import annotations

import logging
import os
from typing import Any

logger = logging.getLogger("malaysia.itemcode")

# BOM 구성품으로 인정할 prefix (낱개). MD- 뚜껑은 제외.
_COMPONENT_PREFIXES = ("MT-", "MX-", "MZ-")


class MalaysiaItemcodeReader:
    """itemcode_db 읽기 전용 풀 + 세트 BOM 조회.

    설정이 없으면 enabled=False, 메서드는 빈 결과를 돌려준다.
    """

    def __init__(self) -> None:
        self._pool: Any = None
        self.enabled = False
        self.reason = ""
        self.last_error = ""
        self._configure()

    def _configure(self) -> None:
        dsn = os.getenv("ITEMCODE_DB_URL", "").strip()
        if not dsn:
            self.reason = "ITEMCODE_DB_URL 미설정 — 세트 BOM 조회 비활성."
            return
        try:
            from psycopg.rows import dict_row
            from psycopg_pool import ConnectionPool

            self._pool = ConnectionPool(
                conninfo=dsn,
                min_size=1,
                max_size=3,
                kwargs={"row_factory": dict_row, "autocommit": True},
                open=False,
            )
            self._pool.open(wait=False)
            self.enabled = True
            self.reason = ""
        except Exception as exc:  # noqa: BLE001 — 설정/드라이버 문제로 모듈을 죽이지 않음
            self.reason = f"itemcode_db 연결 풀 생성 실패: {type(exc).__name__}"

    def set_bom_map(self) -> dict[str, list[dict[str, Any]]]:
        """MY- 세트의 BOM 전체.

        반환: { set_code: [{"component_code","component_qty"}, ...] }
        store.explode_stocktake / explode_set 에 그대로 전달 가능.
        구성품은 낱개(MT/MX/MZ)만 포함(혹시 모를 비낱개/뚜껑은 제외).
        """
        if not self.enabled or not self._pool:
            return {}
        try:
            with self._pool.connection() as conn:
                rows = conn.execute(
                    "SELECT set_code, single_code, quantity "
                    "FROM set_components WHERE set_code LIKE 'MY-%'"
                ).fetchall()
            self.last_error = ""
        except Exception as exc:  # noqa: BLE001 — 모듈을 죽이지 않음
            self.last_error = f"{type(exc).__name__}: {exc}"
            logger.exception("itemcode set_components 조회 실패")
            return {}
        out: dict[str, list[dict[str, Any]]] = {}
        for r in rows:
            comp = str(r.get("single_code") or "").strip().upper()
            if not comp.startswith(_COMPONENT_PREFIXES):
                continue
            try:
                qty = int(r.get("quantity") or 0)
            except (TypeError, ValueError):
                qty = 0
            if qty <= 0:
                continue
            out.setdefault(str(r["set_code"]).strip().upper(), []).append(
                {"component_code": comp, "component_qty": qty}
            )
        return out

    def name_map(self) -> dict[str, str]:
        """낱개+세트 코드 → 이름(itemcode_db 기준). 표시 보강용(선택)."""
        if not self.enabled or not self._pool:
            return {}
        try:
            with self._pool.connection() as conn:
                rows = conn.execute(
                    "SELECT item_code, name FROM single_items "
                    "UNION ALL SELECT item_code, name FROM set_items"
                ).fetchall()
            self.last_error = ""
        except Exception as exc:  # noqa: BLE001
            self.last_error = f"{type(exc).__name__}: {exc}"
            return {}
        return {
            str(r["item_code"]).strip().upper(): str(r.get("name") or "").strip()
            for r in rows
        }

    def sabangnet_code_map(self) -> dict[str, str]:
        """낱개+세트 item_code → 사방넷 코드(itemcode_db). 엑셀 내보내기용."""
        if not self.enabled or not self._pool:
            return {}
        try:
            with self._pool.connection() as conn:
                rows = conn.execute(
                    "SELECT item_code, sabangnet_code FROM single_items "
                    "UNION ALL SELECT item_code, sabangnet_code FROM set_items"
                ).fetchall()
            self.last_error = ""
        except Exception as exc:  # noqa: BLE001
            self.last_error = f"{type(exc).__name__}: {exc}"
            return {}
        return {
            str(r["item_code"]).strip().upper(): str(r.get("sabangnet_code") or "").strip()
            for r in rows
        }

    def close(self) -> None:
        if self._pool is not None:
            self._pool.close()