king
c07d3301e4
- 재고현황 헤더 "전산 재고 현황", Last Stocktake 옆 조사수량+차이(±) 표기 (차이 = 마지막 재고조사 낱개 total − 전산재고) - 재고조사 확정(전산 반영)은 관리자(is_admin)만 가능, 비관리자 안내 - 슈퍼관리자만 재고조사 완전 삭제(목록/상세 버튼 + /delete 라우트) - "◀◀ 조사 목록" 버튼 검정 바탕(primary) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
687 lines
30 KiB
Python
687 lines
30 KiB
Python
"""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_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 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 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)
|
|
|
|
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}
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 직렬화
|
|
# ════════════════════════════════════════════════════════════
|
|
@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
|