king
6b43d9926d
- 세트구성(set_bom) 수동 관리 메뉴/라우트/API/템플릿 제거 - 세트 BOM은 itemcode_db set_components(set_code/single_code/quantity)에서 읽음 (MalaysiaItemcodeReader, ITEMCODE_DB_URL 재사용) - db.py: create_set_out/compute_result/finalize_stocktake 에 bom_map 주입식 - 입출고: 낱개 전체 그리드 일괄 등록 + 세트 출고 그리드(BOM 분해) - 일일 재고조사: 낱개+세트 전체 그리드에 수량만 키인 → 일괄 저장 - malaysia_stock_db.set_bom 테이블은 레거시(미사용) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
121 lines
4.5 KiB
Python
121 lines
4.5 KiB
Python
"""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 close(self) -> None:
|
|
if self._pool is not None:
|
|
self._pool.close()
|