시골약사
8f2823e6df
DB 구조 개선: - 454개 주성분코드 기반 herb_masters 테이블 생성 - 53,769개 제품 데이터를 herb_products 테이블에 임포트 - 128개 업체 정보를 product_companies 테이블에 추가 - 기존 herb_items에 ingredient_code 매핑 (100% 완료) UI/API 개선: - 급여 약재 보유 현황 표시 (28/454 = 6.2%) - 재고 현황에 프로그레스 바 추가 - 주성분코드 기준 통계 API 추가 문서화: - 데이터베이스 리팩토링 제안서 작성 - 리팩토링 결과 보고서 작성 - 백업 정보 문서화 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
1078 lines
42 KiB
Python
1078 lines
42 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
|
|
from excel_processor import ExcelProcessor
|
|
|
|
# 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.herb_item_id,
|
|
h.insurance_code,
|
|
h.herb_name,
|
|
h.is_active,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as current_stock,
|
|
GROUP_CONCAT(DISTINCT et.tag_name) as efficacy_tags
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
AND il.is_depleted = 0
|
|
LEFT JOIN herb_item_tags hit ON h.herb_item_id = hit.herb_item_id
|
|
LEFT JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
|
|
WHERE h.is_active = 1
|
|
GROUP BY h.herb_item_id, h.insurance_code, h.herb_name, h.is_active
|
|
ORDER BY h.herb_name
|
|
""")
|
|
herbs = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 태그를 리스트로 변환
|
|
for herb in herbs:
|
|
if herb['efficacy_tags']:
|
|
herb['efficacy_tags'] = herb['efficacy_tags'].split(',')
|
|
else:
|
|
herb['efficacy_tags'] = []
|
|
|
|
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/suppliers', methods=['GET'])
|
|
def get_suppliers():
|
|
"""도매상 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT supplier_id, name, business_no, phone, address, is_active
|
|
FROM suppliers
|
|
WHERE is_active = 1
|
|
ORDER BY name
|
|
""")
|
|
suppliers = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': suppliers})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/suppliers', methods=['POST'])
|
|
def create_supplier():
|
|
"""도매상 등록"""
|
|
try:
|
|
data = request.json
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO suppliers (name, business_no, contact_person, phone, address)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (
|
|
data['name'],
|
|
data.get('business_no'),
|
|
data.get('contact_person'),
|
|
data.get('phone'),
|
|
data.get('address')
|
|
))
|
|
supplier_id = cursor.lastrowid
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '도매상이 등록되었습니다',
|
|
'supplier_id': supplier_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/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
|
|
|
|
# 도매상 ID 가져오기 (폼 데이터에서)
|
|
supplier_id = request.form.get('supplier_id')
|
|
if not supplier_id:
|
|
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 프로세서로 파일 처리
|
|
processor = ExcelProcessor()
|
|
if not processor.read_excel(filepath):
|
|
return jsonify({'success': False, 'error': 'Excel 파일을 읽을 수 없습니다'}), 400
|
|
|
|
# 형식 감지 및 처리
|
|
try:
|
|
df = processor.process()
|
|
except ValueError as e:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'지원하지 않는 Excel 형식입니다: {str(e)}'
|
|
}), 400
|
|
|
|
# 데이터 검증
|
|
valid, msg = processor.validate_data()
|
|
if not valid:
|
|
return jsonify({'success': False, 'error': f'데이터 검증 실패: {msg}'}), 400
|
|
|
|
# 표준 형식으로 변환
|
|
df = processor.export_to_standard()
|
|
|
|
# 처리 요약 정보
|
|
summary = processor.get_summary()
|
|
|
|
# 데이터 처리
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
processed_rows = 0
|
|
processed_items = set()
|
|
|
|
# 도매상 정보 확인
|
|
cursor.execute("SELECT name FROM suppliers WHERE supplier_id = ?", (supplier_id,))
|
|
supplier_info = cursor.fetchone()
|
|
if not supplier_info:
|
|
return jsonify({'success': False, 'error': '유효하지 않은 도매상입니다'}), 400
|
|
|
|
# 날짜별로 그룹화 (도매상은 이미 선택됨)
|
|
grouped = df.groupby(['receipt_date'])
|
|
|
|
for receipt_date, group in grouped:
|
|
|
|
# 입고장 헤더 생성
|
|
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))
|
|
|
|
processed_rows += 1
|
|
processed_items.add(row['herb_name'])
|
|
|
|
# 응답 메시지 생성
|
|
format_name = {
|
|
'hanisarang': '한의사랑',
|
|
'haninfo': '한의정보'
|
|
}.get(summary['format_type'], '알 수 없음')
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'{format_name} 형식 입고 데이터가 성공적으로 처리되었습니다',
|
|
'filename': filename,
|
|
'summary': {
|
|
'format': format_name,
|
|
'processed_rows': processed_rows,
|
|
'total_items': len(processed_items),
|
|
'total_quantity': f"{summary['total_quantity']:,.0f}g",
|
|
'total_amount': f"{summary['total_amount']:,.0f}원"
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 입고장 조회/관리 API ====================
|
|
|
|
@app.route('/api/purchase-receipts', methods=['GET'])
|
|
def get_purchase_receipts():
|
|
"""입고장 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 날짜 범위 파라미터
|
|
start_date = request.args.get('start_date')
|
|
end_date = request.args.get('end_date')
|
|
supplier_id = request.args.get('supplier_id')
|
|
|
|
query = """
|
|
SELECT
|
|
pr.receipt_id,
|
|
pr.receipt_date,
|
|
pr.receipt_no,
|
|
pr.source_file,
|
|
pr.created_at,
|
|
s.name as supplier_name,
|
|
s.supplier_id,
|
|
COUNT(prl.line_id) as line_count,
|
|
SUM(prl.quantity_g) as total_quantity,
|
|
SUM(prl.line_total) as total_amount
|
|
FROM purchase_receipts pr
|
|
JOIN suppliers s ON pr.supplier_id = s.supplier_id
|
|
LEFT JOIN purchase_receipt_lines prl ON pr.receipt_id = prl.receipt_id
|
|
WHERE 1=1
|
|
"""
|
|
params = []
|
|
|
|
if start_date:
|
|
query += " AND pr.receipt_date >= ?"
|
|
params.append(start_date)
|
|
if end_date:
|
|
query += " AND pr.receipt_date <= ?"
|
|
params.append(end_date)
|
|
if supplier_id:
|
|
query += " AND pr.supplier_id = ?"
|
|
params.append(supplier_id)
|
|
|
|
query += " GROUP BY pr.receipt_id ORDER BY pr.receipt_date DESC, pr.created_at DESC"
|
|
|
|
cursor.execute(query, params)
|
|
receipts = []
|
|
for row in cursor.fetchall():
|
|
receipt = dict(row)
|
|
# 타입 변환 (bytes 문제 해결)
|
|
for key, value in receipt.items():
|
|
if isinstance(value, bytes):
|
|
# bytes를 float로 변환 시도
|
|
try:
|
|
import struct
|
|
receipt[key] = struct.unpack('d', value)[0]
|
|
except:
|
|
receipt[key] = float(0)
|
|
elif key in ['receipt_date', 'created_at'] and value is not None:
|
|
receipt[key] = str(value)
|
|
|
|
# total_amount와 total_quantity 반올림
|
|
if 'total_amount' in receipt and receipt['total_amount'] is not None:
|
|
receipt['total_amount'] = round(float(receipt['total_amount']), 2)
|
|
if 'total_quantity' in receipt and receipt['total_quantity'] is not None:
|
|
receipt['total_quantity'] = round(float(receipt['total_quantity']), 2)
|
|
|
|
receipts.append(receipt)
|
|
|
|
return jsonify({'success': True, 'data': receipts})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>', methods=['GET'])
|
|
def get_purchase_receipt_detail(receipt_id):
|
|
"""입고장 상세 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 입고장 헤더 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
pr.*,
|
|
s.name as supplier_name,
|
|
s.business_no as supplier_business_no,
|
|
s.phone as supplier_phone
|
|
FROM purchase_receipts pr
|
|
JOIN suppliers s ON pr.supplier_id = s.supplier_id
|
|
WHERE pr.receipt_id = ?
|
|
""", (receipt_id,))
|
|
|
|
receipt = cursor.fetchone()
|
|
if not receipt:
|
|
return jsonify({'success': False, 'error': '입고장을 찾을 수 없습니다'}), 404
|
|
|
|
receipt_data = dict(receipt)
|
|
|
|
# 입고장 상세 라인 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
prl.*,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
il.lot_id,
|
|
il.quantity_onhand as current_stock
|
|
FROM purchase_receipt_lines prl
|
|
JOIN herb_items h ON prl.herb_item_id = h.herb_item_id
|
|
LEFT JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.receipt_id = ?
|
|
ORDER BY prl.line_id
|
|
""", (receipt_id,))
|
|
|
|
lines = [dict(row) for row in cursor.fetchall()]
|
|
receipt_data['lines'] = lines
|
|
|
|
return jsonify({'success': True, 'data': receipt_data})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>/lines/<int:line_id>', methods=['PUT'])
|
|
def update_purchase_receipt_line(receipt_id, line_id):
|
|
"""입고장 라인 수정"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 기존 라인 정보 조회
|
|
cursor.execute("""
|
|
SELECT prl.*, il.lot_id, il.quantity_onhand, il.quantity_received
|
|
FROM purchase_receipt_lines prl
|
|
LEFT JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.line_id = ? AND prl.receipt_id = ?
|
|
""", (line_id, receipt_id))
|
|
|
|
old_line = cursor.fetchone()
|
|
if not old_line:
|
|
return jsonify({'success': False, 'error': '입고 라인을 찾을 수 없습니다'}), 404
|
|
|
|
# 재고 사용 여부 확인
|
|
if old_line['quantity_onhand'] != old_line['quantity_received']:
|
|
used_qty = old_line['quantity_received'] - old_line['quantity_onhand']
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'이미 {used_qty}g이 사용되어 수정할 수 없습니다'
|
|
}), 400
|
|
|
|
# 수정 가능한 필드만 업데이트
|
|
update_fields = []
|
|
params = []
|
|
|
|
if 'quantity_g' in data:
|
|
update_fields.append('quantity_g = ?')
|
|
params.append(data['quantity_g'])
|
|
|
|
if 'unit_price_per_g' in data:
|
|
update_fields.append('unit_price_per_g = ?')
|
|
params.append(data['unit_price_per_g'])
|
|
|
|
if 'line_total' in data:
|
|
update_fields.append('line_total = ?')
|
|
params.append(data['line_total'])
|
|
elif 'quantity_g' in data and 'unit_price_per_g' in data:
|
|
# 자동 계산
|
|
line_total = float(data['quantity_g']) * float(data['unit_price_per_g'])
|
|
update_fields.append('line_total = ?')
|
|
params.append(line_total)
|
|
|
|
if 'origin_country' in data:
|
|
update_fields.append('origin_country = ?')
|
|
params.append(data['origin_country'])
|
|
|
|
if not update_fields:
|
|
return jsonify({'success': False, 'error': '수정할 내용이 없습니다'}), 400
|
|
|
|
# 입고장 라인 업데이트
|
|
params.append(line_id)
|
|
cursor.execute(f"""
|
|
UPDATE purchase_receipt_lines
|
|
SET {', '.join(update_fields)}
|
|
WHERE line_id = ?
|
|
""", params)
|
|
|
|
# 재고 로트 업데이트 (수량 변경시)
|
|
if 'quantity_g' in data and old_line['lot_id']:
|
|
cursor.execute("""
|
|
UPDATE inventory_lots
|
|
SET quantity_received = ?, quantity_onhand = ?
|
|
WHERE lot_id = ?
|
|
""", (data['quantity_g'], data['quantity_g'], old_line['lot_id']))
|
|
|
|
# 재고 원장에 조정 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger
|
|
(event_type, herb_item_id, lot_id, quantity_delta, notes, reference_table, reference_id)
|
|
VALUES ('ADJUST',
|
|
(SELECT herb_item_id FROM purchase_receipt_lines WHERE line_id = ?),
|
|
?, ?, '입고장 수정', 'purchase_receipt_lines', ?)
|
|
""", (line_id, old_line['lot_id'],
|
|
float(data['quantity_g']) - float(old_line['quantity_g']), line_id))
|
|
|
|
# 입고장 헤더의 총액 업데이트
|
|
cursor.execute("""
|
|
UPDATE purchase_receipts
|
|
SET total_amount = (
|
|
SELECT SUM(line_total)
|
|
FROM purchase_receipt_lines
|
|
WHERE receipt_id = ?
|
|
),
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE receipt_id = ?
|
|
""", (receipt_id, receipt_id))
|
|
|
|
return jsonify({'success': True, 'message': '입고 라인이 수정되었습니다'})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>', methods=['DELETE'])
|
|
def delete_purchase_receipt(receipt_id):
|
|
"""입고장 삭제 (재고 사용 확인 후)"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 재고 사용 여부 확인
|
|
cursor.execute("""
|
|
SELECT
|
|
COUNT(*) as used_count,
|
|
SUM(il.quantity_received - il.quantity_onhand) as used_quantity
|
|
FROM purchase_receipt_lines prl
|
|
JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.receipt_id = ?
|
|
AND il.quantity_onhand < il.quantity_received
|
|
""", (receipt_id,))
|
|
|
|
usage = cursor.fetchone()
|
|
if usage['used_count'] > 0:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'{usage["used_count"]}개 품목에서 {usage["used_quantity"]}g이 이미 사용되어 삭제할 수 없습니다'
|
|
}), 400
|
|
|
|
# 삭제 순서 중요: 참조하는 테이블부터 삭제
|
|
# 1. 재고 원장 기록 삭제 (lot_id를 참조)
|
|
cursor.execute("""
|
|
DELETE FROM stock_ledger
|
|
WHERE lot_id IN (
|
|
SELECT lot_id FROM inventory_lots
|
|
WHERE receipt_line_id IN (
|
|
SELECT line_id FROM purchase_receipt_lines WHERE receipt_id = ?
|
|
)
|
|
)
|
|
""", (receipt_id,))
|
|
|
|
# 2. 재고 로트 삭제 (receipt_line_id를 참조)
|
|
cursor.execute("""
|
|
DELETE FROM inventory_lots
|
|
WHERE receipt_line_id IN (
|
|
SELECT line_id FROM purchase_receipt_lines WHERE receipt_id = ?
|
|
)
|
|
""", (receipt_id,))
|
|
|
|
# 3. 입고장 라인 삭제 (receipt_id를 참조)
|
|
cursor.execute("DELETE FROM purchase_receipt_lines WHERE receipt_id = ?", (receipt_id,))
|
|
|
|
# 4. 입고장 헤더 삭제
|
|
cursor.execute("DELETE FROM purchase_receipts WHERE receipt_id = ?", (receipt_id,))
|
|
|
|
return jsonify({'success': True, 'message': '입고장이 삭제되었습니다'})
|
|
|
|
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']
|
|
origin_country = ingredient.get('origin_country') # 원산지 선택 정보
|
|
|
|
# 조제 약재 구성 기록
|
|
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
|
|
|
|
# 원산지가 지정된 경우 해당 원산지만, 아니면 전체에서 FIFO
|
|
if origin_country and origin_country != 'auto':
|
|
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
|
|
AND origin_country = ?
|
|
ORDER BY unit_price_per_g, received_date, lot_id
|
|
""", (herb_item_id, origin_country))
|
|
else:
|
|
# 자동 선택: 가격이 저렴한 것부터
|
|
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 unit_price_per_g, 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/herbs/<int:herb_item_id>/available-lots', methods=['GET'])
|
|
def get_available_lots(herb_item_id):
|
|
"""조제용 가용 로트 목록 - 원산지별로 그룹화"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 약재 정보
|
|
cursor.execute("""
|
|
SELECT herb_name, insurance_code
|
|
FROM herb_items
|
|
WHERE herb_item_id = ?
|
|
""", (herb_item_id,))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
return jsonify({'success': False, 'error': '약재를 찾을 수 없습니다'}), 404
|
|
|
|
# 가용 로트 목록 (소진되지 않은 재고)
|
|
cursor.execute("""
|
|
SELECT
|
|
lot_id,
|
|
origin_country,
|
|
quantity_onhand,
|
|
unit_price_per_g,
|
|
received_date,
|
|
supplier_id
|
|
FROM inventory_lots
|
|
WHERE herb_item_id = ?
|
|
AND is_depleted = 0
|
|
AND quantity_onhand > 0
|
|
ORDER BY origin_country, unit_price_per_g, received_date
|
|
""", (herb_item_id,))
|
|
|
|
lots = []
|
|
for row in cursor.fetchall():
|
|
lots.append({
|
|
'lot_id': row[0],
|
|
'origin_country': row[1] or '미지정',
|
|
'quantity_onhand': row[2],
|
|
'unit_price_per_g': row[3],
|
|
'received_date': row[4],
|
|
'supplier_id': row[5]
|
|
})
|
|
|
|
# 원산지별 요약
|
|
origin_summary = {}
|
|
for lot in lots:
|
|
origin = lot['origin_country']
|
|
if origin not in origin_summary:
|
|
origin_summary[origin] = {
|
|
'origin_country': origin,
|
|
'total_quantity': 0,
|
|
'min_price': float('inf'),
|
|
'max_price': 0,
|
|
'lot_count': 0,
|
|
'lots': []
|
|
}
|
|
|
|
origin_summary[origin]['total_quantity'] += lot['quantity_onhand']
|
|
origin_summary[origin]['min_price'] = min(origin_summary[origin]['min_price'], lot['unit_price_per_g'])
|
|
origin_summary[origin]['max_price'] = max(origin_summary[origin]['max_price'], lot['unit_price_per_g'])
|
|
origin_summary[origin]['lot_count'] += 1
|
|
origin_summary[origin]['lots'].append(lot)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'herb_name': herb[0],
|
|
'insurance_code': herb[1],
|
|
'origins': list(origin_summary.values()),
|
|
'total_quantity': sum(lot['quantity_onhand'] for lot in lots)
|
|
}
|
|
})
|
|
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,
|
|
COUNT(DISTINCT il.origin_country) as origin_count,
|
|
AVG(il.unit_price_per_g) as avg_price,
|
|
MIN(il.unit_price_per_g) as min_price,
|
|
MAX(il.unit_price_per_g) as max_price,
|
|
COALESCE(SUM(il.quantity_onhand * il.unit_price_per_g), 0) as total_value,
|
|
GROUP_CONCAT(DISTINCT et.tag_name) as efficacy_tags
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id AND il.is_depleted = 0
|
|
LEFT JOIN herb_item_tags hit ON h.herb_item_id = hit.herb_item_id
|
|
LEFT JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
|
|
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()]
|
|
|
|
# 태그를 리스트로 변환
|
|
for item in inventory:
|
|
if item['efficacy_tags']:
|
|
item['efficacy_tags'] = item['efficacy_tags'].split(',')
|
|
else:
|
|
item['efficacy_tags'] = []
|
|
|
|
# 전체 요약
|
|
total_value = sum(item['total_value'] for item in inventory)
|
|
total_items = len(inventory)
|
|
|
|
# 주성분코드 기준 보유 현황 추가
|
|
cursor.execute("""
|
|
SELECT COUNT(DISTINCT ingredient_code)
|
|
FROM herb_masters
|
|
""")
|
|
total_ingredient_codes = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
SELECT COUNT(DISTINCT h.ingredient_code)
|
|
FROM herb_items h
|
|
INNER JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
WHERE il.quantity_onhand > 0 AND il.is_depleted = 0
|
|
AND h.ingredient_code IS NOT NULL
|
|
""")
|
|
owned_ingredient_codes = cursor.fetchone()[0]
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': inventory,
|
|
'summary': {
|
|
'total_items': total_items,
|
|
'total_value': total_value,
|
|
'total_ingredient_codes': total_ingredient_codes, # 전체 급여 약재 수
|
|
'owned_ingredient_codes': owned_ingredient_codes, # 보유 약재 수
|
|
'coverage_rate': round(owned_ingredient_codes * 100 / total_ingredient_codes, 1) if total_ingredient_codes > 0 else 0 # 보유율
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/inventory/detail/<int:herb_item_id>', methods=['GET'])
|
|
def get_inventory_detail(herb_item_id):
|
|
"""약재별 재고 상세 - 원산지별로 구분"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 약재 기본 정보
|
|
cursor.execute("""
|
|
SELECT herb_item_id, insurance_code, herb_name
|
|
FROM herb_items
|
|
WHERE herb_item_id = ?
|
|
""", (herb_item_id,))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
return jsonify({'success': False, 'error': '약재를 찾을 수 없습니다'}), 404
|
|
|
|
herb_data = dict(herb)
|
|
|
|
# 원산지별 재고 정보
|
|
cursor.execute("""
|
|
SELECT
|
|
il.lot_id,
|
|
il.origin_country,
|
|
il.quantity_onhand,
|
|
il.unit_price_per_g,
|
|
il.received_date,
|
|
il.supplier_id,
|
|
s.name as supplier_name,
|
|
il.quantity_onhand * il.unit_price_per_g as lot_value
|
|
FROM inventory_lots il
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
WHERE il.herb_item_id = ? AND il.is_depleted = 0
|
|
ORDER BY il.origin_country, il.unit_price_per_g, il.received_date
|
|
""", (herb_item_id,))
|
|
|
|
lots = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 원산지별 그룹화
|
|
by_origin = {}
|
|
for lot in lots:
|
|
origin = lot['origin_country'] or '미지정'
|
|
if origin not in by_origin:
|
|
by_origin[origin] = {
|
|
'origin_country': origin,
|
|
'lots': [],
|
|
'total_quantity': 0,
|
|
'total_value': 0,
|
|
'min_price': float('inf'),
|
|
'max_price': 0,
|
|
'avg_price': 0
|
|
}
|
|
|
|
by_origin[origin]['lots'].append(lot)
|
|
by_origin[origin]['total_quantity'] += lot['quantity_onhand']
|
|
by_origin[origin]['total_value'] += lot['lot_value']
|
|
by_origin[origin]['min_price'] = min(by_origin[origin]['min_price'], lot['unit_price_per_g'])
|
|
by_origin[origin]['max_price'] = max(by_origin[origin]['max_price'], lot['unit_price_per_g'])
|
|
|
|
# 평균 단가 계산
|
|
for origin_data in by_origin.values():
|
|
if origin_data['total_quantity'] > 0:
|
|
origin_data['avg_price'] = origin_data['total_value'] / origin_data['total_quantity']
|
|
|
|
herb_data['origins'] = list(by_origin.values())
|
|
herb_data['total_origins'] = len(by_origin)
|
|
|
|
return jsonify({'success': True, 'data': herb_data})
|
|
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) |