화학물질 반입·반출대장 자동화: 파이썬으로 구축하는 현장형 기록·재고 시스템

이 글의 목적은 실험실·공장·창고에서 사용하는 화학물질 반입·반출대장을 파이썬과 SQLite로 자동화하여 누구나 즉시 도입 가능한 표준 절차와 예시 코드를 제공하는 것이다.

1. 왜 자동화해야 하는가

수기 대장은 누락과 오기 가능성이 높고 실시간 재고 파악이 어렵다. 자동화하면 입력 즉시 재고가 갱신되고 월별 보고서가 반복 없이 생성된다. 변경 이력과 무결성을 관리하면 내부감사 대응력이 향상된다. 표준화된 CSV 템플릿과 명령형 인터페이스를 사용하면 교육 부담도 낮아진다.

2. 설계 원칙

표준 라이브러리만 사용하여 배포와 유지보수를 단순화한다.
로컬 파일 기반 DB(SQLite)로 네트워크 없이도 동작하게 한다.
트랜잭션 해시체인으로 위변조를 탐지한다.
CLI 명령을 반입(IN)·반출(OUT)·재고·보고·무결성 점검으로 최소화한다.
단위와 임계치를 화학물질 마스터에서 관리하여 입력 오류를 줄인다.

3. 데이터 모델 개요

두 개의 핵심 테이블을 사용한다. 하나는 화학물질 마스터이고 다른 하나는 입출고 트랜잭션이다. 마스터에는 CAS, 물질명, 기본단위, 보관장소, 임계치, 규제대상 여부를 둔다. 트랜잭션에는 시간, 방향(IN·OUT), 수량, 단위, 위치, 담당자, 사용목적, LOT, 공급처, 관련문서번호, 비고, 해시체인 필드를 둔다.

테이블	주요 필드	설명
chemicals	cas, name, unit, storage, threshold, is_controlled	물질 속성과 운영 임계치를 보관한다.
transactions	ts, cas, direction, qty, unit, location, person, purpose, lot, supplier, ref_doc, remarks, prev_hash, hash	반입·반출 기록과 해시체인으로 무결성을 보장한다.

4. 해시체인 무결성

각 트랜잭션은 이전 레코드의 해시와 현재 페이로드를 결합해 새 해시를 만든다. 중간 변경이 발생하면 이후 모든 해시 검증이 실패한다. 무결성 점검은 입력 순서대로 재계산하여 기대 해시와 저장 해시를 비교한다.

5. 설치와 초기화 절차

파이썬 3.8 이상을 준비한다.
폴더에 스크립트를 저장한다.
초기 실행에서 스키마를 생성하고 화학물질 목록을 일괄 등록한다.

# DB 초기화와 마스터 등록
python chem_ledger.py init --db chem_ledger.db --import-chemicals chemicals.csv

6. 표준 운영 절차(SOP)

단계	행동	성공 기준	산출물
1	반입 수령 시 즉시 IN 기록	필수 필드 누락 없음	transactions 레코드
2	사용 또는 출고 시 OUT 기록	수량 단위 일치	transactions 레코드
3	교대 종료 전 balance 점검	임계치 미만 경고 확인	재고 리포트
4	월말 report 실행	CSV 생성 확인	월별 CSV
5	audit 실행	무결성 정상	점검 로그

7. CLI 명령 요약

명령	필수 옵션	주요 옵션	용도
init	-	--import-chemicals	DB 스키마 생성 및 마스터 일괄 등록
add-chem	--cas, --name	--unit, --storage, --threshold, --controlled	신규 물질 단건 등록
in	--cas, --qty	--unit, --location, --person, --purpose, --lot, --supplier, --ref, --remarks	반입 기록
out	--cas, --qty	--unit, --location, --person, --purpose, --ref, --remarks	반출 기록
balance	-	--cas, --by-location	재고 집계
report	-	--month, --out	CSV 보고서 생성
audit	-	-	해시체인 무결성 점검

8. 예시 입력과 실행 예

# 화학물질 단건 등록
python chem_ledger.py add-chem --cas 64-17-5 --name 에탄올 --unit L --storage 창고A --threshold 2.0 --controlled 0

# 반입

python chem\_ledger.py in --cas 64-17-5 --qty 10 --unit L --person 홍길동 --location 창고A --purpose 재고보충 --lot LOT-2025-09 --supplier ABC --ref PO-2025-0919

# 반출

python chem\_ledger.py out --cas 64-17-5 --qty 1.5 --unit L --person 김철수 --location 실험실1 --purpose 실험 --ref JOB-99

# 재고

python chem\_ledger.py balance
python chem\_ledger.py balance --by-location

# 월별 보고서

python chem\_ledger.py report --month 2025-09 --out ledger\_2025-09.csv

# 무결성 점검

python chem\_ledger.py audit

9. CSV 템플릿 예시

cas,name,unit,storage,threshold,is_controlled
67-56-1,메탄올,L,창고A,1.0,0
64-17-5,에탄올,L,창고A,2.0,0
7732-18-5,물,L,창고B,,0

CSV 헤더는 소문자를 권장하며 필수 필드는 cas와 name이다. 임계치가 없으면 빈값을 두고 is_controlled는 0 또는 1로 관리한다.

10. 임계치 경고 운용

재고 집계 시 마스터의 threshold와 비교하여 미만인 물질을 경고 목록으로 출력한다. 경고는 물질별 재고와 임계치를 함께 표시하여 보충 우선순위를 정하는 근거가 된다. 경고 발생 시 책임자는 반입 계획을 수립하고 필요한 경우 대체 용제를 지정한다.

11. 보고서 구성과 활용

월별 보고서는 트랜잭션의 시각, CAS, 방향, 수량, 단위, 위치, 담당자, 목적, LOT, 공급처, 관련문서, 비고, 레코드 해시를 포함한다. CSV는 엑셀 피벗테이블과 호환되어 공정·부서·위치별 사용량 분석에 바로 활용 가능하다. 보고서 파일명은 ledger_YYYY-MM.csv 규칙을 사용한다.

12. 시간대와 단위 처리

기본은 로컬 시스템 시간의 ISO 형식을 사용한다. 조직이 한국 표준시 고정 저장을 원하면 Asia/Seoul 기준으로 고정 변환하여 저장한다. 단위는 마스터의 unit을 기본값으로 사용하며 입력 시 별도 지정이 없으면 마스터 단위가 자동 적용된다. 서로 다른 단위를 혼용하지 않도록 운영 기준을 사전에 정한다.

13. 백업과 복구

운영 DB 파일과 월별 보고서를 매일 사본으로 백업한다.
무결성 점검 결과 로그를 함께 보관한다.
복구 시 가장 최근의 정상 무결성 지점까지 되돌리고 이후 기록을 재입력한다.

14. 권한과 책임

입력 권한자는 반입·반출 기록을 수행하고 승인 권한자는 일일 재고와 경고를 확인한다. 승인 없이 스키마 변경을 금지한다. 관리자만 마스터 등록과 임계치 변경을 수행한다.

15. 오류 예방과 검증

리스크	징후	예방 조치	검증 방법
단위 혼용	재고 수치 급변	마스터 단위 고정	월말 사용량 분산 확인
위치 누락	위치별 재고 불일치	--by-location 정례 점검	교대 종료 체크리스트
레코드 변조	audit 오류	쓰기 권한 제한	주간 무결성 리포트

16. 확장 아이디어

바코드 또는 QR 스캔으로 CAS와 LOT 자동 입력을 구현한다.
엑셀 양식과 양방향 연동을 위해 CSV 입출력 스크립트를 분리한다.
폐기 트랜잭션 타입을 추가하여 재고와 회수량을 구분한다.
다중 사용자 환경에서는 파일 잠금 정책을 정의한다.

17. 예시 코드 전체

#!/usr/bin/env python3
# chem_ledger.py
# 화학물질 반입·반출대장 자동화 예시 (SQLite, 표준라이브러리만 사용)

import argparse
import csv
import hashlib
import sqlite3
from datetime import datetime, timezone
from typing import Optional, Tuple

DEFAULT\_DB = "chem\_ledger.db"

SCHEMA\_SQL = """
PRAGMA journal\_mode=WAL;
CREATE TABLE IF NOT EXISTS chemicals (
id INTEGER PRIMARY KEY,
cas TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
unit TEXT,
storage TEXT,
threshold REAL,
is\_controlled INTEGER DEFAULT 0,
created\_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY,
ts TEXT NOT NULL,
cas TEXT NOT NULL,
direction TEXT NOT NULL CHECK (direction IN ('IN','OUT')),
qty REAL NOT NULL CHECK (qty >= 0),
unit TEXT,
location TEXT,
person TEXT,
purpose TEXT,
lot TEXT,
supplier TEXT,
ref\_doc TEXT,
remarks TEXT,
prev\_hash TEXT,
hash TEXT,
FOREIGN KEY (cas) REFERENCES chemicals(cas)
);
CREATE INDEX IF NOT EXISTS idx\_trx\_cas\_ts ON transactions(cas, ts);
CREATE INDEX IF NOT EXISTS idx\_trx\_ts ON transactions(ts);
"""

def now\_iso() -> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")

def connect(db\_path: str):
con = sqlite3.connect(db\_path, isolation\_level=None)
con.execute("PRAGMA foreign\_keys=ON;")
return con

def ensure\_schema(con: sqlite3.Connection):
con.executescript(SCHEMA\_SQL)

def import\_chemicals(con: sqlite3.Connection, csv\_path: str):
with open(csv\_path, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
cols = \[c.lower() for c in reader.fieldnames or \[]]
required = {"cas", "name"}
if not required.issubset(set(cols)):
raise ValueError(f"CSV 헤더에 cas,name 필드가 필요함")
for row in reader:
cas = row\.get("cas","").strip()
name = row\.get("name","").strip()
unit = (row\.get("unit") or "").strip() or None
storage = (row\.get("storage") or "").strip() or None
threshold = row\.get("threshold")
threshold\_v = float(threshold) if threshold not in (None,"") else None
controlled = row\.get("is\_controlled") or row\.get("controlled") or "0"
is\_ctrl = 1 if str(controlled).strip() in ("1","true","True","Y","y") else 0
if not cas or not name:
continue
con.execute(
"INSERT OR IGNORE INTO chemicals(cas,name,unit,storage,threshold,is\_controlled,created\_at) VALUES (?,?,?,?,?,?,?)",
(cas, name, unit, storage, threshold\_v, is\_ctrl, now\_iso()),
)

def add\_chemical(con: sqlite3.Connection, cas: str, name: str, unit: Optional\[str],
storage: Optional\[str], threshold: Optional\[float], is\_ctrl: int):
con.execute(
"INSERT OR IGNORE INTO chemicals(cas,name,unit,storage,threshold,is\_controlled,created\_at) VALUES (?,?,?,?,?,?,?)",
(cas, name, unit, storage, threshold, is\_ctrl, now\_iso()),
)

def get\_prev\_hash(con: sqlite3.Connection):
cur = con.execute("SELECT hash FROM transactions ORDER BY id DESC LIMIT 1")
row = cur.fetchone()
return row\[0] if row and row\[0] else None

def compute\_hash(prev: Optional\[str], payload: Tuple) -> str:
m = hashlib.sha256()
if prev:
m.update(prev.encode("utf-8"))
m.update("|".join("" if v is None else str(v) for v in payload).encode("utf-8"))
return m.hexdigest()

def default\_unit\_for(con: sqlite3.Connection, cas: str) -> Optional\[str]:
cur = con.execute("SELECT unit FROM chemicals WHERE cas=?", (cas,))
row = cur.fetchone()
return row\[0] if row else None

def record(con: sqlite3.Connection, direction: str, cas: str, qty: float, unit: Optional\[str],
location: Optional\[str], person: Optional\[str], purpose: Optional\[str],
lot: Optional\[str], supplier: Optional\[str], ref\_doc: Optional\[str], remarks: Optional\[str]):
cur = con.execute("SELECT 1 FROM chemicals WHERE cas=?", (cas,))
if not cur.fetchone():
raise ValueError(f"chemicals 테이블에 CAS {cas} 없음")
ts = now\_iso()
unit\_effective = unit or default\_unit\_for(con, cas)
prev = get\_prev\_hash(con)
payload = (ts, cas, direction, qty, unit\_effective, location, person, purpose, lot, supplier, ref\_doc, remarks)
trx\_hash = compute\_hash(prev, payload)
con.execute("""
INSERT INTO transactions(ts,cas,direction,qty,unit,location,person,purpose,lot,supplier,ref\_doc,remarks,prev\_hash,hash)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
""", (\*payload, prev, trx\_hash))
return trx\_hash

def current\_balance(con: sqlite3.Connection, cas: Optional\[str] = None, by\_location: bool = False):
if by\_location:
if cas:
sql = """SELECT cas, COALESCE(location,'(미지정)') AS loc,
SUM(CASE direction WHEN 'IN' THEN qty ELSE -qty END) AS balance,
MAX(unit) AS unit
FROM transactions
WHERE cas=?
GROUP BY cas, loc
ORDER BY cas, loc"""
args = (cas,)
else:
sql = """SELECT cas, COALESCE(location,'(미지정)') AS loc,
SUM(CASE direction WHEN 'IN' THEN qty ELSE -qty END) AS balance,
MAX(unit) AS unit
FROM transactions
GROUP BY cas, loc
ORDER BY cas, loc"""
args = ()
else:
if cas:
sql = """SELECT cas,
SUM(CASE direction WHEN 'IN' THEN qty ELSE -qty END) AS balance,
MAX(unit) AS unit
FROM transactions
WHERE cas=?
GROUP BY cas
ORDER BY cas"""
args = (cas,)
else:
sql = """SELECT cas,
SUM(CASE direction WHEN 'IN' THEN qty ELSE -qty END) AS balance,
MAX(unit) AS unit
FROM transactions
GROUP BY cas
ORDER BY cas"""
args = ()
return con.execute(sql, args).fetchall()

def print\_balance(con: sqlite3.Connection, cas: Optional\[str], by\_location: bool):
rows = current\_balance(con, cas, by\_location)
hdr = \["CAS", "위치", "재고", "단위"] if by\_location else \["CAS", "재고", "단위"]
print(",".join(hdr))
if by\_location:
for cas, loc, bal, unit in rows:
print(f"{cas},{loc},{bal or 0},{unit or ''}")
else:
for cas, bal, unit in rows:
print(f"{cas},{bal or 0},{unit or ''}")
warn = con.execute("""
SELECT c.cas, c.name, c.threshold, b.balance, COALESCE(b.unit, c.unit) AS unit
FROM chemicals c
LEFT JOIN (
SELECT cas, SUM(CASE direction WHEN 'IN' THEN qty ELSE -qty END) AS balance, MAX(unit) AS unit
FROM transactions GROUP BY cas
) b ON c.cas=b.cas
WHERE c.threshold IS NOT NULL AND COALESCE(b.balance,0) < c.threshold
ORDER BY c.cas
""").fetchall()
if warn:
print("\n\[경고] 임계치 미만 물질:")
for casx, name, thr, bal, unit in warn:
print(f" - {casx} {name}: 재고 {bal or 0}{unit or ''} < 임계치 {thr}{unit or ''}")

def write\_csv(path: str, rows, header):
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.writer(f)
w\.writerow(header)
for r in rows:
w\.writerow(r)

def month\_range(month: str) -> Tuple\[str,str]:
dt = datetime.strptime(month, "%Y-%m")
start = dt.replace(day=1)
if start.month == 12:
end = start.replace(year=start.year+1, month=1, day=1)
else:
end = start.replace(month=start.month+1, day=1)
return (start.date().isoformat(), end.date().isoformat())

def report(con: sqlite3.Connection, month: Optional\[str], out\_path: str):
if month:
start, end = month\_range(month)
rows = con.execute("""
SELECT ts, cas, direction, qty, unit, COALESCE(location,''), COALESCE(person,''), COALESCE(purpose,''),
COALESCE(lot,''), COALESCE(supplier,''), COALESCE(ref\_doc,''), COALESCE(remarks,''), hash
FROM transactions
WHERE date(ts) >= ? AND date(ts) < ?
ORDER BY ts, id
""", (start, end)).fetchall()
fname = out\_path or f"ledger\_{month}.csv"
else:
rows = con.execute("""
SELECT ts, cas, direction, qty, unit, COALESCE(location,''), COALESCE(person,''), COALESCE(purpose,''),
COALESCE(lot,''), COALESCE(supplier,''), COALESCE(ref\_doc,''), COALESCE(remarks,''), hash
FROM transactions
ORDER BY ts, id
""").fetchall()
fname = out\_path or "ledger\_all.csv"
header = \["ts","cas","direction","qty","unit","location","person","purpose","lot","supplier","ref\_doc","remarks","hash"]
write\_csv(fname, rows, header)
print(f"보고서 저장: {fname} (행 {len(rows)})")

def audit(con: sqlite3.Connection):
cur = con.execute("""
SELECT id, ts, cas, direction, qty, unit, location, person, purpose, lot, supplier, ref\_doc, remarks, prev\_hash, hash
FROM transactions ORDER BY id
""")
prev = None
ok = True
for row in cur:
(rid, ts, cas, direction, qty, unit, location, person, purpose, lot, supplier, ref\_doc, remarks, prev\_hash, h) = row
payload = (ts, cas, direction, qty, unit, location, person, purpose, lot, supplier, ref\_doc, remarks)
expect = compute\_hash(prev, payload)
if prev\_hash != prev or h != expect:
ok = False
print(f"\[무결성오류] id={rid}")
prev = h
print("무결성: 정상" if ok else "무결성: 오류 존재")

def positive\_float(x: str) -> float:
v = float(x)
if v < 0:
raise argparse.ArgumentTypeError("qty는 0 이상이어야 함")
return v

def cli():
p = argparse.ArgumentParser(description="화학물질 반입·반출대장 자동화 (SQLite)")
p.add\_argument("--db", default=DEFAULT\_DB, help=f"DB 파일 경로 (기본: {DEFAULT\_DB})")
sub = p.add\_subparsers(dest="cmd", required=True)

```
sp_init = sub.add_parser("init", help="DB 초기화 및 화학물질 일괄등록")
sp_init.add_argument("--import-chemicals", help="chemicals.csv 경로")

sp_add = sub.add_parser("add-chem", help="화학물질 단건 등록")
sp_add.add_argument("--cas", required=True)
sp_add.add_argument("--name", required=True)
sp_add.add_argument("--unit")
sp_add.add_argument("--storage")
sp_add.add_argument("--threshold", type=float)
sp_add.add_argument("--controlled", type=int, default=0, choices=[0,1])

for cmd in ("in","out"):
    sp_io = sub.add_parser(cmd, help=f"{'반입' if cmd=='in' else '반출'} 기록")
    sp_io.add_argument("--cas", required=True)
    sp_io.add_argument("--qty", required=True, type=positive_float)
    sp_io.add_argument("--unit")
    sp_io.add_argument("--location")
    sp_io.add_argument("--person")
    sp_io.add_argument("--purpose")
    sp_io.add_argument("--lot")
    sp_io.add_argument("--supplier")
    sp_io.add_argument("--ref")
    sp_io.add_argument("--remarks")

sp_bal = sub.add_parser("balance", help="재고 현황")
sp_bal.add_argument("--cas")
sp_bal.add_argument("--by-location", action="store_true")

sp_rep = sub.add_parser("report", help="CSV 보고서 생성")
sp_rep.add_argument("--month", help="YYYY-MM 형식. 미지정 시 전체")
sp_rep.add_argument("--out", help="CSV 출력 경로")

sp_audit = sub.add_parser("audit", help="해시체인 무결성 검사")

args = p.parse_args()
con = connect(args.db)
ensure_schema(con)

if args.cmd == "init":
    if args.import_chemicals:
        import_chemicals(con, args.import_chemicals)
        print(f"화학물질 일괄등록 완료")
    else:
        print("스키마만 초기화 완료")
elif args.cmd == "add-chem":
    add_chemical(con, args.cas, args.name, args.unit, args.storage, args.threshold, args.controlled)
    print(f"등록 완료: {args.cas} {args.name}")
elif args.cmd in ("in","out"):
    h = record(
        con,
        "IN" if args.cmd=="in" else "OUT",
        args.cas, args.qty, args.unit, args.location, args.person,
        args.purpose, args.lot, args.supplier, args.ref, args.remarks
    )
    print(f"기록 완료: {args.cmd.upper()} {args.cas} {args.qty}{args.unit or ''} hash={h[:12]}")
elif args.cmd == "balance":
    print_balance(con, args.cas, args.by_location)
elif args.cmd == "report":
    report(con, args.month, args.out)
elif args.cmd == "audit":
    audit(con)
else:
    p.print_help()
```

if **name** == "**main**":
cli()

18. 운영 체크리스트

교대 종료 전 balance 실행과 경고 검토를 완료한다.
주간 audit 결과를 기록한다.
월말 report 파일을 표준 경로에 보관한다.
DB 파일과 보고서를 이중화 스토리지에 백업한다.

FAQ

엑셀만 사용하는 팀과 어떻게 연계하나

CSV 보고서를 생성하여 공용 폴더에 저장하고 엑셀에서 데이터 가져오기를 사용하여 피벗테이블을 구성하면 된다.

여러 위치의 재고를 따로 관리하려면 어떻게 하나

반입·반출 시 location 필드를 반드시 채우고 재고 조회에서 --by-location 옵션을 사용하면 된다.

단위가 다른 물질을 혼용 입력하면 어떻게 되나

마스터의 unit을 기본으로 사용하므로 입력 시 단위를 생략하면 일관성이 유지된다. 단위를 변경해야 하면 마스터 단위부터 수정한 다음 사용한다.

기록을 잘못 입력했을 때 수정은 가능한가

무결성 보전을 위해 직접 수정 대신 반대 방향의 보정 트랜잭션을 추가한다. 예를 들어 OUT 1.0을 잘못 입력했으면 IN 1.0 보정 레코드를 추가한다.

시간대는 어떻게 표준화하나

시스템 시간을 그대로 저장하거나 한국 표준시로 변환하여 저장한다. 조직 표준을 하나로 정하고 전 팀에 동일하게 적용한다.

규제대상 물질은 따로 표시할 수 있나

is_controlled 플래그를 사용하여 목록과 보고서에서 구분한다. 필요하면 보고서 후처리에서 필터링한다.

SafeChem EHS Platform

이 블로그 검색

토사 붕락 사고 사례 분석과 예방대책 총정리