"""malaysia_stock_db PostgreSQL 저장소. - 드라이버: psycopg 3 (`psycopg[binary,pool]`) — cupang/expense 모듈과 동일 패턴. - 연결 정보: 환경변수 `MALAYSIA_STOCK_DB_URL` (예: postgresql://malaysia_app:@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 MAX(r.created_at) 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 reposition_rack( self, *, stocktake_id: int, entries: list[dict[str, Any]] ) -> dict[str, Any]: """랙 위치 재배치 전용(랙 보기 드래그 저장). daily_stocktake_rack 의 칸(cell_code) 배치만 교체한다. 위치 이동은 SKU 총수량을 바꾸지 않으므로 daily_stocktake_line 은 건드리지 않는다 (확정본의 집계/전산재고 보존). 확정(finalized) 조사도 허용 — 위치만 변경. 취소된 조사는 불가. entries 형식은 replace_rack 과 동일. """ 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)) cell_seq: dict[str, int] = {} with self._pool.connection() as conn: with conn.transaction(): head = conn.execute( "SELECT status FROM daily_stocktake WHERE id = %s FOR UPDATE", (stocktake_id,), ).fetchone() if not head: raise KeyError(stocktake_id) if head["status"] == "cancelled": raise ValueError("취소된 조사는 수정할 수 없습니다.") conn.execute( "DELETE FROM daily_stocktake_rack 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), ) return {"entries": len(clean)} 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