시골약사
2fddc89bca
✨ 주요 기능 - 환자 관리: 환자 등록 및 조회 (이름, 전화번호, 주민번호, 성별) - 입고 관리: Excel 파일 업로드로 대량 입고 처리 - 처방 관리: 약속 처방 템플릿 등록 및 관리 - 조제 관리: 처방 기반 조제 및 약재 가감 기능 - 재고 관리: 실시간 재고 현황 및 로트별 관리 🛠️ 기술 스택 - Backend: Flask (Python 웹 프레임워크) - Database: SQLite (경량 관계형 데이터베이스) - Frontend: Bootstrap + jQuery - Excel 처리: pandas + openpyxl 🔧 핵심 개념 - 1제 = 20첩 = 30파우치 (기본값) - FIFO 방식 재고 차감 - 로트별 원산지/단가 관리 - 정확한 조제 원가 계산 📁 프로젝트 구조 - app.py: Flask 백엔드 서버 - database/: 데이터베이스 스키마 및 파일 - templates/: HTML 템플릿 - static/: JavaScript 및 CSS - sample/: 샘플 Excel 파일 🤖 Generated with Claude Code Co-Authored-By: Claude <noreply@anthropic.com>
513 lines
19 KiB
Python
513 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
한약 재고관리 시스템 - Flask Backend
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from flask import Flask, request, jsonify, render_template, send_from_directory
|
|
from flask_cors import CORS
|
|
import pandas as pd
|
|
from werkzeug.utils import secure_filename
|
|
import json
|
|
from contextlib import contextmanager
|
|
|
|
# Flask 앱 초기화
|
|
app = Flask(__name__, static_folder='static', template_folder='templates')
|
|
app.config['SECRET_KEY'] = 'your-secret-key-change-in-production'
|
|
app.config['DATABASE'] = 'database/kdrug.db'
|
|
app.config['UPLOAD_FOLDER'] = 'uploads'
|
|
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024 # 16MB max file size
|
|
|
|
CORS(app)
|
|
|
|
# 업로드 폴더 생성
|
|
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
|
|
os.makedirs('database', exist_ok=True)
|
|
|
|
# 허용된 파일 확장자
|
|
ALLOWED_EXTENSIONS = {'xlsx', 'xls'}
|
|
|
|
def allowed_file(filename):
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
|
|
|
# 데이터베이스 연결 컨텍스트 매니저
|
|
@contextmanager
|
|
def get_db():
|
|
conn = sqlite3.connect(app.config['DATABASE'])
|
|
conn.row_factory = sqlite3.Row # 딕셔너리 형태로 반환
|
|
conn.execute('PRAGMA foreign_keys = ON') # 외래키 제약 활성화
|
|
try:
|
|
yield conn
|
|
conn.commit()
|
|
except Exception as e:
|
|
conn.rollback()
|
|
raise e
|
|
finally:
|
|
conn.close()
|
|
|
|
# 데이터베이스 초기화
|
|
def init_db():
|
|
with open('database/schema.sql', 'r', encoding='utf-8') as f:
|
|
schema = f.read()
|
|
|
|
with get_db() as conn:
|
|
conn.executescript(schema)
|
|
print("Database initialized successfully")
|
|
|
|
# 라우트: 메인 페이지
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
# ==================== 환자 관리 API ====================
|
|
|
|
@app.route('/api/patients', methods=['GET'])
|
|
def get_patients():
|
|
"""환자 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT patient_id, name, phone, gender, birth_date, notes
|
|
FROM patients
|
|
WHERE is_active = 1
|
|
ORDER BY created_at DESC
|
|
""")
|
|
patients = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': patients})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/patients', methods=['POST'])
|
|
def create_patient():
|
|
"""새 환자 등록"""
|
|
try:
|
|
data = request.json
|
|
required_fields = ['name', 'phone']
|
|
|
|
# 필수 필드 검증
|
|
for field in required_fields:
|
|
if field not in data or not data[field]:
|
|
return jsonify({'success': False, 'error': f'{field}는 필수입니다'}), 400
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO patients (name, phone, jumin_no, gender, birth_date, address, notes)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
data['name'],
|
|
data['phone'],
|
|
data.get('jumin_no'),
|
|
data.get('gender'),
|
|
data.get('birth_date'),
|
|
data.get('address'),
|
|
data.get('notes')
|
|
))
|
|
patient_id = cursor.lastrowid
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '환자가 등록되었습니다',
|
|
'patient_id': patient_id
|
|
})
|
|
except sqlite3.IntegrityError:
|
|
return jsonify({'success': False, 'error': '이미 등록된 환자입니다'}), 400
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 약재 관리 API ====================
|
|
|
|
@app.route('/api/herbs', methods=['GET'])
|
|
def get_herbs():
|
|
"""약재 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT h.*, COALESCE(s.total_quantity, 0) as current_stock
|
|
FROM herb_items h
|
|
LEFT JOIN v_current_stock s ON h.herb_item_id = s.herb_item_id
|
|
WHERE h.is_active = 1
|
|
ORDER BY h.herb_name
|
|
""")
|
|
herbs = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': herbs})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 처방 관리 API ====================
|
|
|
|
@app.route('/api/formulas', methods=['GET'])
|
|
def get_formulas():
|
|
"""처방 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT formula_id, formula_code, formula_name, formula_type,
|
|
base_cheop, base_pouches, description
|
|
FROM formulas
|
|
WHERE is_active = 1
|
|
ORDER BY formula_name
|
|
""")
|
|
formulas = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': formulas})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/formulas', methods=['POST'])
|
|
def create_formula():
|
|
"""새 처방 등록"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 처방 마스터 생성
|
|
cursor.execute("""
|
|
INSERT INTO formulas (formula_code, formula_name, formula_type,
|
|
base_cheop, base_pouches, description, created_by)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
data.get('formula_code'),
|
|
data['formula_name'],
|
|
data.get('formula_type', 'CUSTOM'),
|
|
data.get('base_cheop', 20),
|
|
data.get('base_pouches', 30),
|
|
data.get('description'),
|
|
data.get('created_by', 'system')
|
|
))
|
|
formula_id = cursor.lastrowid
|
|
|
|
# 구성 약재 추가
|
|
if 'ingredients' in data:
|
|
for idx, ingredient in enumerate(data['ingredients']):
|
|
cursor.execute("""
|
|
INSERT INTO formula_ingredients (formula_id, herb_item_id,
|
|
grams_per_cheop, notes, sort_order)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (
|
|
formula_id,
|
|
ingredient['herb_item_id'],
|
|
ingredient['grams_per_cheop'],
|
|
ingredient.get('notes'),
|
|
idx
|
|
))
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '처방이 등록되었습니다',
|
|
'formula_id': formula_id
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/formulas/<int:formula_id>/ingredients', methods=['GET'])
|
|
def get_formula_ingredients(formula_id):
|
|
"""처방 구성 약재 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT fi.*, h.herb_name, h.insurance_code
|
|
FROM formula_ingredients fi
|
|
JOIN herb_items h ON fi.herb_item_id = h.herb_item_id
|
|
WHERE fi.formula_id = ?
|
|
ORDER BY fi.sort_order
|
|
""", (formula_id,))
|
|
ingredients = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': ingredients})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 입고 관리 API ====================
|
|
|
|
@app.route('/api/upload/purchase', methods=['POST'])
|
|
def upload_purchase_excel():
|
|
"""Excel 파일 업로드 및 입고 처리"""
|
|
try:
|
|
if 'file' not in request.files:
|
|
return jsonify({'success': False, 'error': '파일이 없습니다'}), 400
|
|
|
|
file = request.files['file']
|
|
if file.filename == '':
|
|
return jsonify({'success': False, 'error': '파일이 선택되지 않았습니다'}), 400
|
|
|
|
if not allowed_file(file.filename):
|
|
return jsonify({'success': False, 'error': '허용되지 않는 파일 형식입니다'}), 400
|
|
|
|
# 파일 저장
|
|
filename = secure_filename(file.filename)
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
filename = f"{timestamp}_{filename}"
|
|
filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
|
file.save(filepath)
|
|
|
|
# Excel 파일 읽기
|
|
df = pd.read_excel(filepath)
|
|
|
|
# 컬럼 매핑 (Excel 컬럼명 -> DB 필드)
|
|
column_mapping = {
|
|
'제품코드': 'insurance_code',
|
|
'업체명': 'supplier_name',
|
|
'약재명': 'herb_name',
|
|
'구입일자': 'receipt_date',
|
|
'구입량': 'quantity',
|
|
'구입액': 'total_amount',
|
|
'원산지': 'origin_country'
|
|
}
|
|
|
|
df = df.rename(columns=column_mapping)
|
|
|
|
# 데이터 처리
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 날짜별, 업체별로 그룹화
|
|
grouped = df.groupby(['receipt_date', 'supplier_name'])
|
|
|
|
for (receipt_date, supplier_name), group in grouped:
|
|
# 공급업체 확인/생성
|
|
cursor.execute("SELECT supplier_id FROM suppliers WHERE name = ?", (supplier_name,))
|
|
supplier = cursor.fetchone()
|
|
|
|
if not supplier:
|
|
cursor.execute("""
|
|
INSERT INTO suppliers (name) VALUES (?)
|
|
""", (supplier_name,))
|
|
supplier_id = cursor.lastrowid
|
|
else:
|
|
supplier_id = supplier[0]
|
|
|
|
# 입고장 헤더 생성
|
|
total_amount = group['total_amount'].sum()
|
|
cursor.execute("""
|
|
INSERT INTO purchase_receipts (supplier_id, receipt_date, total_amount, source_file)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (supplier_id, str(receipt_date), total_amount, filename))
|
|
receipt_id = cursor.lastrowid
|
|
|
|
# 입고장 라인 생성
|
|
for _, row in group.iterrows():
|
|
# 약재 확인/생성
|
|
cursor.execute("""
|
|
SELECT herb_item_id FROM herb_items
|
|
WHERE insurance_code = ? OR herb_name = ?
|
|
""", (row.get('insurance_code'), row['herb_name']))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
cursor.execute("""
|
|
INSERT INTO herb_items (insurance_code, herb_name)
|
|
VALUES (?, ?)
|
|
""", (row.get('insurance_code'), row['herb_name']))
|
|
herb_item_id = cursor.lastrowid
|
|
else:
|
|
herb_item_id = herb[0]
|
|
|
|
# 단가 계산 (총액 / 수량)
|
|
quantity = float(row['quantity'])
|
|
total = float(row['total_amount'])
|
|
unit_price = total / quantity if quantity > 0 else 0
|
|
|
|
# 입고장 라인 생성
|
|
cursor.execute("""
|
|
INSERT INTO purchase_receipt_lines
|
|
(receipt_id, herb_item_id, origin_country, quantity_g, unit_price_per_g, line_total)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (receipt_id, herb_item_id, row.get('origin_country'),
|
|
quantity, unit_price, total))
|
|
line_id = cursor.lastrowid
|
|
|
|
# 재고 로트 생성
|
|
cursor.execute("""
|
|
INSERT INTO inventory_lots
|
|
(herb_item_id, supplier_id, receipt_line_id, received_date, origin_country,
|
|
unit_price_per_g, quantity_received, quantity_onhand)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (herb_item_id, supplier_id, line_id, str(receipt_date),
|
|
row.get('origin_country'), unit_price, quantity, quantity))
|
|
lot_id = cursor.lastrowid
|
|
|
|
# 재고 원장 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger
|
|
(event_type, herb_item_id, lot_id, quantity_delta, unit_cost_per_g,
|
|
reference_table, reference_id)
|
|
VALUES ('RECEIPT', ?, ?, ?, ?, 'purchase_receipts', ?)
|
|
""", (herb_item_id, lot_id, quantity, unit_price, receipt_id))
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'입고 데이터가 성공적으로 처리되었습니다',
|
|
'filename': filename
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 조제 관리 API ====================
|
|
|
|
@app.route('/api/compounds', methods=['POST'])
|
|
def create_compound():
|
|
"""조제 실행"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 조제 마스터 생성
|
|
cursor.execute("""
|
|
INSERT INTO compounds (patient_id, formula_id, compound_date,
|
|
je_count, cheop_total, pouch_total,
|
|
prescription_no, notes, created_by)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
data.get('patient_id'),
|
|
data.get('formula_id'),
|
|
data.get('compound_date', datetime.now().strftime('%Y-%m-%d')),
|
|
data['je_count'],
|
|
data['cheop_total'],
|
|
data['pouch_total'],
|
|
data.get('prescription_no'),
|
|
data.get('notes'),
|
|
data.get('created_by', 'system')
|
|
))
|
|
compound_id = cursor.lastrowid
|
|
|
|
total_cost = 0
|
|
|
|
# 조제 약재 처리
|
|
for ingredient in data['ingredients']:
|
|
herb_item_id = ingredient['herb_item_id']
|
|
total_grams = ingredient['total_grams']
|
|
|
|
# 조제 약재 구성 기록
|
|
cursor.execute("""
|
|
INSERT INTO compound_ingredients (compound_id, herb_item_id,
|
|
grams_per_cheop, total_grams)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (compound_id, herb_item_id,
|
|
ingredient['grams_per_cheop'], total_grams))
|
|
|
|
# 재고 차감 (FIFO 방식)
|
|
remaining_qty = total_grams
|
|
cursor.execute("""
|
|
SELECT lot_id, quantity_onhand, unit_price_per_g
|
|
FROM inventory_lots
|
|
WHERE herb_item_id = ? AND is_depleted = 0 AND quantity_onhand > 0
|
|
ORDER BY received_date, lot_id
|
|
""", (herb_item_id,))
|
|
|
|
lots = cursor.fetchall()
|
|
|
|
for lot in lots:
|
|
if remaining_qty <= 0:
|
|
break
|
|
|
|
lot_id = lot[0]
|
|
available = lot[1]
|
|
unit_price = lot[2]
|
|
|
|
used = min(remaining_qty, available)
|
|
cost = used * unit_price
|
|
total_cost += cost
|
|
|
|
# 소비 내역 기록
|
|
cursor.execute("""
|
|
INSERT INTO compound_consumptions (compound_id, herb_item_id, lot_id,
|
|
quantity_used, unit_cost_per_g, cost_amount)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (compound_id, herb_item_id, lot_id, used, unit_price, cost))
|
|
|
|
# 로트 재고 감소
|
|
new_qty = available - used
|
|
cursor.execute("""
|
|
UPDATE inventory_lots
|
|
SET quantity_onhand = ?, is_depleted = ?
|
|
WHERE lot_id = ?
|
|
""", (new_qty, 1 if new_qty == 0 else 0, lot_id))
|
|
|
|
# 재고 원장 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger (event_type, herb_item_id, lot_id,
|
|
quantity_delta, unit_cost_per_g,
|
|
reference_table, reference_id)
|
|
VALUES ('CONSUME', ?, ?, ?, ?, 'compounds', ?)
|
|
""", (herb_item_id, lot_id, -used, unit_price, compound_id))
|
|
|
|
remaining_qty -= used
|
|
|
|
if remaining_qty > 0:
|
|
raise Exception(f"재고 부족: {ingredient.get('herb_name', herb_item_id)}")
|
|
|
|
# 총 원가 업데이트
|
|
cursor.execute("""
|
|
UPDATE compounds SET cost_total = ? WHERE compound_id = ?
|
|
""", (total_cost, compound_id))
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '조제가 완료되었습니다',
|
|
'compound_id': compound_id,
|
|
'total_cost': total_cost
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 재고 현황 API ====================
|
|
|
|
@app.route('/api/inventory/summary', methods=['GET'])
|
|
def get_inventory_summary():
|
|
"""재고 현황 요약"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
h.herb_item_id,
|
|
h.insurance_code,
|
|
h.herb_name,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as total_quantity,
|
|
COUNT(DISTINCT il.lot_id) as lot_count,
|
|
AVG(il.unit_price_per_g) as avg_price,
|
|
COALESCE(SUM(il.quantity_onhand * il.unit_price_per_g), 0) as total_value
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id AND il.is_depleted = 0
|
|
GROUP BY h.herb_item_id, h.insurance_code, h.herb_name
|
|
HAVING total_quantity > 0
|
|
ORDER BY h.herb_name
|
|
""")
|
|
inventory = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 전체 요약
|
|
total_value = sum(item['total_value'] for item in inventory)
|
|
total_items = len(inventory)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': inventory,
|
|
'summary': {
|
|
'total_items': total_items,
|
|
'total_value': total_value
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# 서버 실행
|
|
if __name__ == '__main__':
|
|
# 데이터베이스 초기화
|
|
if not os.path.exists(app.config['DATABASE']):
|
|
init_db()
|
|
|
|
# 개발 서버 실행
|
|
app.run(debug=True, host='0.0.0.0', port=5001) |