시골약사
1441c01fb4
- 프론트엔드: 조제 시 실시간 커스텀 처방 감지 - 처방 선택 시 원래 구성 약재 저장 - 약재 추가/삭제/변경 시 즉시 감지 - 가감방 뱃지 및 변경 내용 표시 - 백엔드: 커스텀 처방 자동 감지 및 저장 - compounds 테이블에 커스텀 관련 필드 추가 - 조제 시 원 처방과 비교하여 변경사항 자동 감지 - 커스텀 처방 정보 저장 (추가/제거/변경된 약재) - 환자 조제 내역에 커스텀 처방 표시 - 가감방 뱃지 표시 - 변경 내용 상세 표시 - DB 마이그레이션 스크립트 추가 - is_custom, custom_summary, custom_type 필드 추가 - compound_ingredients에 modification_type, original_grams 필드 추가 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
2114 lines
85 KiB
Python
2114 lines
85 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')
|
|
|
|
@app.route('/survey/<survey_token>')
|
|
def survey_page(survey_token):
|
|
"""문진표 페이지 (모바일)"""
|
|
return render_template('survey.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/<int:patient_id>', methods=['GET'])
|
|
def get_patient(patient_id):
|
|
"""환자 개별 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT patient_id, name, phone, jumin_no, gender, birth_date, address, notes
|
|
FROM patients
|
|
WHERE patient_id = ? AND is_active = 1
|
|
""", (patient_id,))
|
|
patient_row = cursor.fetchone()
|
|
if patient_row:
|
|
return jsonify({'success': True, 'data': dict(patient_row)})
|
|
else:
|
|
return jsonify({'success': False, 'error': '환자를 찾을 수 없습니다'}), 404
|
|
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
|
|
|
|
@app.route('/api/herbs/masters', methods=['GET'])
|
|
def get_herb_masters():
|
|
"""주성분코드 기준 전체 약재 목록 조회 (454개)"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
m.ingredient_code,
|
|
m.herb_name,
|
|
m.herb_name_hanja,
|
|
m.herb_name_latin,
|
|
-- 재고 정보
|
|
COALESCE(inv.total_quantity, 0) as stock_quantity,
|
|
COALESCE(inv.lot_count, 0) as lot_count,
|
|
COALESCE(inv.avg_price, 0) as avg_price,
|
|
CASE WHEN inv.total_quantity > 0 THEN 1 ELSE 0 END as has_stock,
|
|
-- 효능 태그
|
|
GROUP_CONCAT(DISTINCT et.tag_name) as efficacy_tags,
|
|
-- 제품 정보
|
|
COUNT(DISTINCT p.company_name) as company_count,
|
|
COUNT(DISTINCT p.product_id) as product_count
|
|
FROM herb_masters m
|
|
LEFT JOIN (
|
|
-- 재고 정보 서브쿼리
|
|
SELECT
|
|
h.ingredient_code,
|
|
SUM(il.quantity_onhand) as total_quantity,
|
|
COUNT(DISTINCT il.lot_id) as lot_count,
|
|
AVG(il.unit_price_per_g) as avg_price
|
|
FROM herb_items h
|
|
INNER JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
WHERE il.is_depleted = 0 AND il.quantity_onhand > 0
|
|
GROUP BY h.ingredient_code
|
|
) inv ON m.ingredient_code = inv.ingredient_code
|
|
LEFT JOIN herb_products p ON m.ingredient_code = p.ingredient_code
|
|
LEFT JOIN herb_items hi ON m.ingredient_code = hi.ingredient_code
|
|
LEFT JOIN herb_item_tags hit ON hi.herb_item_id = hit.herb_item_id
|
|
LEFT JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
|
|
WHERE m.is_active = 1
|
|
GROUP BY m.ingredient_code, m.herb_name, inv.total_quantity, inv.lot_count, inv.avg_price
|
|
ORDER BY has_stock DESC, m.herb_name
|
|
""")
|
|
|
|
herbs = []
|
|
for row in cursor.fetchall():
|
|
herb = dict(row)
|
|
# 효능 태그를 리스트로 변환
|
|
if herb['efficacy_tags']:
|
|
herb['efficacy_tags'] = herb['efficacy_tags'].split(',')
|
|
else:
|
|
herb['efficacy_tags'] = []
|
|
herbs.append(herb)
|
|
|
|
# 통계 정보
|
|
total_herbs = len(herbs)
|
|
herbs_with_stock = sum(1 for h in herbs if h['has_stock'])
|
|
coverage_rate = round(herbs_with_stock * 100 / total_herbs, 1) if total_herbs > 0 else 0
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': herbs,
|
|
'summary': {
|
|
'total_herbs': total_herbs,
|
|
'herbs_with_stock': herbs_with_stock,
|
|
'herbs_without_stock': total_herbs - herbs_with_stock,
|
|
'coverage_rate': coverage_rate
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/herbs/by-ingredient/<ingredient_code>', methods=['GET'])
|
|
def get_herbs_by_ingredient(ingredient_code):
|
|
"""특정 ingredient_code에 해당하는 제품 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 먼저 마스터 약재명 조회
|
|
cursor.execute("""
|
|
SELECT herb_name
|
|
FROM herb_masters
|
|
WHERE ingredient_code = ?
|
|
""", (ingredient_code,))
|
|
master_row = cursor.fetchone()
|
|
master_herb_name = master_row[0] if master_row else None
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
h.herb_item_id,
|
|
h.insurance_code,
|
|
h.herb_name as product_name,
|
|
h.specification,
|
|
CASE
|
|
WHEN h.specification LIKE '%신흥%' THEN '신흥'
|
|
WHEN h.specification LIKE '%세화%' THEN '세화'
|
|
WHEN h.specification LIKE '%한동%' THEN '한동'
|
|
WHEN h.specification IS NULL OR h.specification = '' THEN '일반'
|
|
ELSE h.specification
|
|
END as company_name,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as stock_quantity,
|
|
COUNT(DISTINCT il.lot_id) as lot_count,
|
|
AVG(il.unit_price_per_g) as avg_price
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
AND il.is_depleted = 0
|
|
WHERE h.ingredient_code = ?
|
|
AND h.is_active = 1
|
|
GROUP BY h.herb_item_id
|
|
ORDER BY stock_quantity DESC, h.herb_name
|
|
""", (ingredient_code,))
|
|
|
|
products = []
|
|
for row in cursor.fetchall():
|
|
product = dict(row)
|
|
# 마스터 약재명 추가
|
|
product['herb_name'] = master_herb_name or product['product_name']
|
|
products.append(product)
|
|
|
|
return jsonify({'success': True, 'data': products})
|
|
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):
|
|
"""처방 구성 약재 조회 (ingredient_code 기반, 사용 가능한 모든 제품 포함)"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 처방 구성 약재 조회 (ingredient_code 기반)
|
|
cursor.execute("""
|
|
SELECT
|
|
fi.ingredient_id,
|
|
fi.formula_id,
|
|
fi.ingredient_code,
|
|
fi.grams_per_cheop,
|
|
fi.notes,
|
|
fi.sort_order,
|
|
hm.herb_name,
|
|
hm.herb_name_hanja
|
|
FROM formula_ingredients fi
|
|
LEFT JOIN herb_masters hm ON fi.ingredient_code = hm.ingredient_code
|
|
WHERE fi.formula_id = ?
|
|
ORDER BY fi.sort_order
|
|
""", (formula_id,))
|
|
|
|
ingredients = []
|
|
for row in cursor.fetchall():
|
|
ingredient = dict(row)
|
|
ingredient_code = ingredient['ingredient_code']
|
|
|
|
# 해당 주성분을 가진 사용 가능한 모든 제품 찾기
|
|
cursor.execute("""
|
|
SELECT
|
|
h.herb_item_id,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
h.specification,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as stock,
|
|
COALESCE(AVG(il.unit_price_per_g), 0) as avg_price
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id AND il.is_depleted = 0
|
|
WHERE h.ingredient_code = ?
|
|
GROUP BY h.herb_item_id
|
|
HAVING stock > 0
|
|
ORDER BY stock DESC
|
|
""", (ingredient_code,))
|
|
|
|
available_products = [dict(row) for row in cursor.fetchall()]
|
|
ingredient['available_products'] = available_products
|
|
ingredient['total_available_stock'] = sum(p['stock'] for p in available_products)
|
|
ingredient['product_count'] = len(available_products)
|
|
|
|
ingredients.append(ingredient)
|
|
|
|
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:
|
|
# receipt_date가 튜플인 경우 처리
|
|
if isinstance(receipt_date, tuple):
|
|
receipt_date = receipt_date[0]
|
|
|
|
# receipt_date를 문자열로 확실히 변환
|
|
receipt_date_str = str(receipt_date)
|
|
|
|
# 입고장 번호 생성 (PR-YYYYMMDD-XXXX)
|
|
date_str = receipt_date_str.replace('-', '')
|
|
|
|
# 해당 날짜의 최대 번호 찾기
|
|
cursor.execute("""
|
|
SELECT MAX(CAST(SUBSTR(receipt_no, -4) AS INTEGER))
|
|
FROM purchase_receipts
|
|
WHERE receipt_no LIKE ?
|
|
""", (f'PR-{date_str}-%',))
|
|
|
|
max_num = cursor.fetchone()[0]
|
|
next_num = (max_num or 0) + 1
|
|
receipt_no = f"PR-{date_str}-{next_num:04d}"
|
|
|
|
# 입고장 헤더 생성
|
|
total_amount = float(group['total_amount'].sum()) # float로 변환하여 numpy 타입 문제 해결
|
|
cursor.execute("""
|
|
INSERT INTO purchase_receipts (supplier_id, receipt_date, receipt_no, total_amount, source_file)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (supplier_id, receipt_date_str, receipt_no, total_amount, filename))
|
|
receipt_id = cursor.lastrowid
|
|
|
|
# 입고장 라인 생성
|
|
for _, row in group.iterrows():
|
|
insurance_code = row.get('insurance_code')
|
|
|
|
# 보험코드가 있는 경우 herb_products에서 정보 가져오기
|
|
if insurance_code:
|
|
cursor.execute("""
|
|
SELECT DISTINCT
|
|
hp.ingredient_code,
|
|
hp.product_name,
|
|
hp.company_name
|
|
FROM herb_products hp
|
|
WHERE hp.product_code = ?
|
|
""", (insurance_code,))
|
|
product_info = cursor.fetchone()
|
|
|
|
if product_info:
|
|
ingredient_code = product_info[0]
|
|
product_name = product_info[1]
|
|
company_name = product_info[2]
|
|
|
|
# herb_items에서 해당 보험코드 제품 확인
|
|
cursor.execute("""
|
|
SELECT herb_item_id FROM herb_items
|
|
WHERE insurance_code = ?
|
|
""", (insurance_code,))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
# 새 제품 생성 (ingredient_code, company_name 포함)
|
|
cursor.execute("""
|
|
INSERT INTO herb_items (
|
|
ingredient_code,
|
|
insurance_code,
|
|
herb_name,
|
|
specification
|
|
) VALUES (?, ?, ?, ?)
|
|
""", (ingredient_code, insurance_code, product_name, company_name))
|
|
herb_item_id = cursor.lastrowid
|
|
else:
|
|
herb_item_id = herb[0]
|
|
# 기존 제품의 ingredient_code가 없으면 업데이트
|
|
cursor.execute("""
|
|
UPDATE herb_items
|
|
SET ingredient_code = COALESCE(ingredient_code, ?),
|
|
specification = COALESCE(specification, ?)
|
|
WHERE herb_item_id = ?
|
|
""", (ingredient_code, company_name, herb_item_id))
|
|
else:
|
|
# herb_products에 없는 경우 기존 로직
|
|
cursor.execute("""
|
|
SELECT herb_item_id FROM herb_items
|
|
WHERE insurance_code = ? OR herb_name = ?
|
|
""", (insurance_code, row['herb_name']))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
cursor.execute("""
|
|
INSERT INTO herb_items (insurance_code, herb_name)
|
|
VALUES (?, ?)
|
|
""", (insurance_code, row['herb_name']))
|
|
herb_item_id = cursor.lastrowid
|
|
else:
|
|
herb_item_id = herb[0]
|
|
else:
|
|
# 보험코드가 없는 경우 약재명으로만 처리
|
|
cursor.execute("""
|
|
SELECT herb_item_id FROM herb_items
|
|
WHERE herb_name = ?
|
|
""", (row['herb_name'],))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
cursor.execute("""
|
|
INSERT INTO herb_items (herb_name)
|
|
VALUES (?)
|
|
""", (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)
|
|
|
|
# 입고장 상세 라인 조회 (display_name 포함)
|
|
cursor.execute("""
|
|
SELECT
|
|
prl.*,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
il.lot_id,
|
|
il.quantity_onhand as current_stock,
|
|
il.display_name,
|
|
lv.form,
|
|
lv.processing,
|
|
lv.selection_state,
|
|
lv.grade
|
|
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
|
|
LEFT JOIN lot_variants lv ON il.lot_id = lv.lot_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=['GET'])
|
|
def get_compounds():
|
|
"""조제 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
c.compound_id,
|
|
c.patient_id,
|
|
p.name as patient_name,
|
|
p.phone as patient_phone,
|
|
c.formula_id,
|
|
f.formula_name,
|
|
f.formula_code,
|
|
c.compound_date,
|
|
c.je_count,
|
|
c.cheop_total,
|
|
c.pouch_total,
|
|
c.cost_total,
|
|
c.sell_price_total,
|
|
c.prescription_no,
|
|
c.status,
|
|
c.notes,
|
|
c.created_at,
|
|
c.created_by
|
|
FROM compounds c
|
|
LEFT JOIN patients p ON c.patient_id = p.patient_id
|
|
LEFT JOIN formulas f ON c.formula_id = f.formula_id
|
|
ORDER BY c.created_at DESC
|
|
LIMIT 100
|
|
""")
|
|
compounds = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': compounds})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/compounds/<int:compound_id>', methods=['GET'])
|
|
def get_compound_detail(compound_id):
|
|
"""조제 상세 정보 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 조제 마스터 정보
|
|
cursor.execute("""
|
|
SELECT
|
|
c.*,
|
|
p.name as patient_name,
|
|
p.phone as patient_phone,
|
|
f.formula_name,
|
|
f.formula_code
|
|
FROM compounds c
|
|
LEFT JOIN patients p ON c.patient_id = p.patient_id
|
|
LEFT JOIN formulas f ON c.formula_id = f.formula_id
|
|
WHERE c.compound_id = ?
|
|
""", (compound_id,))
|
|
compound = dict(cursor.fetchone())
|
|
|
|
# 조제 약재 구성
|
|
cursor.execute("""
|
|
SELECT
|
|
ci.*,
|
|
h.herb_name,
|
|
h.insurance_code
|
|
FROM compound_ingredients ci
|
|
JOIN herb_items h ON ci.herb_item_id = h.herb_item_id
|
|
WHERE ci.compound_id = ?
|
|
ORDER BY ci.compound_ingredient_id
|
|
""", (compound_id,))
|
|
ingredients = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 소비 내역
|
|
cursor.execute("""
|
|
SELECT
|
|
cc.*,
|
|
h.herb_name,
|
|
il.origin_country,
|
|
il.supplier_id,
|
|
s.name as supplier_name
|
|
FROM compound_consumptions cc
|
|
JOIN herb_items h ON cc.herb_item_id = h.herb_item_id
|
|
JOIN inventory_lots il ON cc.lot_id = il.lot_id
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
WHERE cc.compound_id = ?
|
|
ORDER BY cc.consumption_id
|
|
""", (compound_id,))
|
|
consumptions = [dict(row) for row in cursor.fetchall()]
|
|
|
|
compound['ingredients'] = ingredients
|
|
compound['consumptions'] = consumptions
|
|
|
|
return jsonify({'success': True, 'data': compound})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/patients/<int:patient_id>/compounds', methods=['GET'])
|
|
def get_patient_compounds(patient_id):
|
|
"""환자별 처방 기록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
c.compound_id,
|
|
c.formula_id,
|
|
f.formula_name,
|
|
f.formula_code,
|
|
c.compound_date,
|
|
c.je_count,
|
|
c.cheop_total,
|
|
c.pouch_total,
|
|
c.cost_total,
|
|
c.sell_price_total,
|
|
c.prescription_no,
|
|
c.status,
|
|
c.notes,
|
|
c.created_at,
|
|
c.created_by,
|
|
c.is_custom,
|
|
c.custom_summary,
|
|
c.custom_type
|
|
FROM compounds c
|
|
LEFT JOIN formulas f ON c.formula_id = f.formula_id
|
|
WHERE c.patient_id = ?
|
|
ORDER BY c.compound_date DESC, c.created_at DESC
|
|
""", (patient_id,))
|
|
compounds = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 환자 정보도 함께 반환
|
|
cursor.execute("""
|
|
SELECT patient_id, name, phone, gender, birth_date, notes
|
|
FROM patients
|
|
WHERE patient_id = ?
|
|
""", (patient_id,))
|
|
patient_row = cursor.fetchone()
|
|
patient = dict(patient_row) if patient_row else None
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'patient': patient,
|
|
'compounds': compounds
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/compounds', methods=['POST'])
|
|
def create_compound():
|
|
"""조제 실행 - 커스텀 처방 감지 포함"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
formula_id = data.get('formula_id')
|
|
|
|
# 커스텀 처방 감지를 위한 준비
|
|
is_custom = False
|
|
custom_summary = ""
|
|
custom_details = {
|
|
'added': [],
|
|
'removed': [],
|
|
'modified': []
|
|
}
|
|
|
|
# formula_id가 있는 경우 원 처방과 비교
|
|
if formula_id:
|
|
# 원 처방 구성 조회 (ingredient_code 기반)
|
|
cursor.execute("""
|
|
SELECT fi.ingredient_code, hm.herb_name, fi.grams_per_cheop
|
|
FROM formula_ingredients fi
|
|
JOIN herb_masters hm ON fi.ingredient_code = hm.ingredient_code
|
|
WHERE fi.formula_id = ?
|
|
""", (formula_id,))
|
|
|
|
# ingredient_code -> herb_item_id 매핑
|
|
original_by_code = {}
|
|
for row in cursor.fetchall():
|
|
ingredient_code = row[0]
|
|
herb_name = row[1]
|
|
grams = row[2]
|
|
|
|
# 해당 ingredient_code를 가진 herb_item_id들 조회
|
|
cursor.execute("SELECT herb_item_id FROM herb_items WHERE ingredient_code = ?",
|
|
(ingredient_code,))
|
|
herb_ids = [r[0] for r in cursor.fetchall()]
|
|
|
|
for herb_id in herb_ids:
|
|
original_by_code[herb_id] = {
|
|
'herb_name': herb_name,
|
|
'grams': grams,
|
|
'ingredient_code': ingredient_code
|
|
}
|
|
|
|
original_ingredients = original_by_code
|
|
|
|
# 실제 조제 구성과 비교
|
|
actual_ingredients = {ing['herb_item_id']: ing['grams_per_cheop']
|
|
for ing in data['ingredients']}
|
|
|
|
# 추가된 약재 확인
|
|
for ing in data['ingredients']:
|
|
herb_id = ing['herb_item_id']
|
|
if herb_id not in original_ingredients:
|
|
# 약재명 조회
|
|
cursor.execute("SELECT herb_name FROM herb_items WHERE herb_item_id = ?", (herb_id,))
|
|
herb_name = cursor.fetchone()[0]
|
|
custom_details['added'].append(f"{herb_name} {ing['grams_per_cheop']}g")
|
|
is_custom = True
|
|
|
|
# 제거된 약재 확인
|
|
for herb_id, info in original_ingredients.items():
|
|
if herb_id not in actual_ingredients:
|
|
custom_details['removed'].append(info['herb_name'])
|
|
is_custom = True
|
|
|
|
# 용량 변경된 약재 확인
|
|
for herb_id, original_info in original_ingredients.items():
|
|
if herb_id in actual_ingredients:
|
|
original_grams = original_info['grams']
|
|
actual_grams = actual_ingredients[herb_id]
|
|
if abs(original_grams - actual_grams) > 0.01:
|
|
custom_details['modified'].append(
|
|
f"{original_info['herb_name']} {original_grams}g→{actual_grams}g"
|
|
)
|
|
is_custom = True
|
|
|
|
# 커스텀 요약 생성
|
|
summary_parts = []
|
|
if custom_details['added']:
|
|
summary_parts.append(f"추가: {', '.join(custom_details['added'])}")
|
|
if custom_details['removed']:
|
|
summary_parts.append(f"제거: {', '.join(custom_details['removed'])}")
|
|
if custom_details['modified']:
|
|
summary_parts.append(f"변경: {', '.join(custom_details['modified'])}")
|
|
|
|
custom_summary = " | ".join(summary_parts) if summary_parts else ""
|
|
|
|
# 조제 마스터 생성 (커스텀 정보 포함)
|
|
cursor.execute("""
|
|
INSERT INTO compounds (patient_id, formula_id, compound_date,
|
|
je_count, cheop_total, pouch_total,
|
|
prescription_no, notes, created_by,
|
|
is_custom, custom_summary, custom_type)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
data.get('patient_id'),
|
|
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'),
|
|
1 if is_custom else 0,
|
|
custom_summary if is_custom else None,
|
|
'custom' if is_custom else 'standard'
|
|
))
|
|
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') # 원산지 선택 정보
|
|
|
|
# modification_type 결정
|
|
modification_type = 'original'
|
|
original_grams = None
|
|
|
|
if formula_id and herb_item_id in original_ingredients:
|
|
orig_g = original_ingredients[herb_item_id]['grams']
|
|
if abs(orig_g - ingredient['grams_per_cheop']) > 0.01:
|
|
modification_type = 'modified'
|
|
original_grams = orig_g
|
|
elif formula_id and herb_item_id not in original_ingredients:
|
|
modification_type = 'added'
|
|
|
|
# 조제 약재 구성 기록 (커스텀 정보 포함)
|
|
cursor.execute("""
|
|
INSERT INTO compound_ingredients (compound_id, herb_item_id,
|
|
grams_per_cheop, total_grams,
|
|
modification_type, original_grams)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (compound_id, herb_item_id,
|
|
ingredient['grams_per_cheop'], total_grams,
|
|
modification_type, original_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/stock-ledger', methods=['GET'])
|
|
def get_stock_ledger():
|
|
"""재고 원장 (입출고 내역) 조회"""
|
|
try:
|
|
herb_id = request.args.get('herb_id')
|
|
limit = request.args.get('limit', 100, type=int)
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
if herb_id:
|
|
cursor.execute("""
|
|
SELECT
|
|
sl.ledger_id,
|
|
sl.event_type,
|
|
sl.event_time,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
sl.quantity_delta,
|
|
sl.unit_cost_per_g,
|
|
sl.reference_table,
|
|
sl.reference_id,
|
|
il.origin_country,
|
|
s.name as supplier_name,
|
|
CASE
|
|
WHEN sl.event_type = 'PURCHASE' OR sl.event_type = 'RECEIPT' THEN pr.receipt_no
|
|
WHEN sl.event_type = 'CONSUME' THEN
|
|
CASE
|
|
WHEN c.prescription_no IS NOT NULL THEN c.prescription_no || ' (' || COALESCE(f.formula_name, '직접조제') || ')'
|
|
ELSE '조제#' || c.compound_id || ' (' || COALESCE(f.formula_name, '직접조제') || ')'
|
|
END
|
|
WHEN sl.event_type = 'ADJUST' THEN adj.adjustment_no
|
|
ELSE NULL
|
|
END as reference_no,
|
|
CASE
|
|
WHEN sl.event_type = 'CONSUME' THEN p.name
|
|
ELSE NULL
|
|
END as patient_name
|
|
FROM stock_ledger sl
|
|
JOIN herb_items h ON sl.herb_item_id = h.herb_item_id
|
|
LEFT JOIN inventory_lots il ON sl.lot_id = il.lot_id
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
LEFT JOIN purchase_receipts pr ON sl.reference_table = 'purchase_receipts' AND sl.reference_id = pr.receipt_id
|
|
LEFT JOIN compounds c ON sl.reference_table = 'compounds' AND sl.reference_id = c.compound_id
|
|
LEFT JOIN patients p ON c.patient_id = p.patient_id
|
|
LEFT JOIN formulas f ON c.formula_id = f.formula_id
|
|
LEFT JOIN stock_adjustments adj ON sl.reference_table = 'stock_adjustments' AND sl.reference_id = adj.adjustment_id
|
|
WHERE sl.herb_item_id = ?
|
|
ORDER BY sl.event_time DESC
|
|
LIMIT ?
|
|
""", (herb_id, limit))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT
|
|
sl.ledger_id,
|
|
sl.event_type,
|
|
sl.event_time,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
sl.quantity_delta,
|
|
sl.unit_cost_per_g,
|
|
sl.reference_table,
|
|
sl.reference_id,
|
|
il.origin_country,
|
|
s.name as supplier_name,
|
|
CASE
|
|
WHEN sl.event_type = 'PURCHASE' OR sl.event_type = 'RECEIPT' THEN pr.receipt_no
|
|
WHEN sl.event_type = 'CONSUME' THEN
|
|
CASE
|
|
WHEN c.prescription_no IS NOT NULL THEN c.prescription_no || ' (' || COALESCE(f.formula_name, '직접조제') || ')'
|
|
ELSE '조제#' || c.compound_id || ' (' || COALESCE(f.formula_name, '직접조제') || ')'
|
|
END
|
|
WHEN sl.event_type = 'ADJUST' THEN adj.adjustment_no
|
|
ELSE NULL
|
|
END as reference_no,
|
|
CASE
|
|
WHEN sl.event_type = 'CONSUME' THEN p.name
|
|
ELSE NULL
|
|
END as patient_name
|
|
FROM stock_ledger sl
|
|
JOIN herb_items h ON sl.herb_item_id = h.herb_item_id
|
|
LEFT JOIN inventory_lots il ON sl.lot_id = il.lot_id
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
LEFT JOIN purchase_receipts pr ON sl.reference_table = 'purchase_receipts' AND sl.reference_id = pr.receipt_id
|
|
LEFT JOIN compounds c ON sl.reference_table = 'compounds' AND sl.reference_id = c.compound_id
|
|
LEFT JOIN patients p ON c.patient_id = p.patient_id
|
|
LEFT JOIN formulas f ON c.formula_id = f.formula_id
|
|
LEFT JOIN stock_adjustments adj ON sl.reference_table = 'stock_adjustments' AND sl.reference_id = adj.adjustment_id
|
|
ORDER BY sl.event_time DESC
|
|
LIMIT ?
|
|
""", (limit,))
|
|
|
|
ledger_entries = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 약재별 현재 재고 요약
|
|
cursor.execute("""
|
|
SELECT
|
|
h.herb_item_id,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as total_stock,
|
|
COUNT(DISTINCT il.lot_id) as active_lots,
|
|
AVG(il.unit_price_per_g) as avg_price
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
AND il.is_depleted = 0
|
|
WHERE h.is_active = 1
|
|
GROUP BY h.herb_item_id
|
|
HAVING total_stock > 0
|
|
ORDER BY h.herb_name
|
|
""")
|
|
|
|
stock_summary = [dict(row) for row in cursor.fetchall()]
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'ledger': ledger_entries,
|
|
'summary': stock_summary
|
|
})
|
|
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
|
|
|
|
# 가용 로트 목록 (소진되지 않은 재고) - display_name 포함
|
|
cursor.execute("""
|
|
SELECT
|
|
il.lot_id,
|
|
il.origin_country,
|
|
il.quantity_onhand,
|
|
il.unit_price_per_g,
|
|
il.received_date,
|
|
il.supplier_id,
|
|
il.display_name,
|
|
lv.form,
|
|
lv.processing,
|
|
lv.selection_state,
|
|
lv.grade
|
|
FROM inventory_lots il
|
|
LEFT JOIN lot_variants lv ON il.lot_id = lv.lot_id
|
|
WHERE il.herb_item_id = ?
|
|
AND il.is_depleted = 0
|
|
AND il.quantity_onhand > 0
|
|
ORDER BY il.origin_country, il.unit_price_per_g, il.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],
|
|
'display_name': row[6],
|
|
'form': row[7],
|
|
'processing': row[8],
|
|
'selection_state': row[9],
|
|
'grade': row[10]
|
|
})
|
|
|
|
# 원산지별 요약
|
|
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)
|
|
|
|
# 원산지별 재고 정보 (display_name 포함)
|
|
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,
|
|
il.display_name,
|
|
lv.form,
|
|
lv.processing,
|
|
lv.selection_state,
|
|
lv.grade
|
|
FROM inventory_lots il
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
LEFT JOIN lot_variants lv ON il.lot_id = lv.lot_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
|
|
|
|
# 서버 실행
|
|
# ==================== 재고 보정 API ====================
|
|
|
|
@app.route('/api/stock-adjustments', methods=['GET'])
|
|
def get_stock_adjustments():
|
|
"""재고 보정 내역 조회"""
|
|
try:
|
|
limit = request.args.get('limit', 100, type=int)
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
sa.adjustment_id,
|
|
sa.adjustment_date,
|
|
sa.adjustment_no,
|
|
sa.adjustment_type,
|
|
sa.notes,
|
|
sa.created_by,
|
|
sa.created_at,
|
|
COUNT(sad.detail_id) as detail_count,
|
|
SUM(ABS(sad.quantity_delta)) as total_adjusted
|
|
FROM stock_adjustments sa
|
|
LEFT JOIN stock_adjustment_details sad ON sa.adjustment_id = sad.adjustment_id
|
|
GROUP BY sa.adjustment_id
|
|
ORDER BY sa.adjustment_date DESC, sa.created_at DESC
|
|
LIMIT ?
|
|
""", (limit,))
|
|
|
|
adjustments = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': adjustments})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/stock-adjustments/<int:adjustment_id>', methods=['GET'])
|
|
def get_stock_adjustment_detail(adjustment_id):
|
|
"""재고 보정 상세 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 보정 헤더
|
|
cursor.execute("""
|
|
SELECT * FROM stock_adjustments
|
|
WHERE adjustment_id = ?
|
|
""", (adjustment_id,))
|
|
adjustment = dict(cursor.fetchone())
|
|
|
|
# 보정 상세
|
|
cursor.execute("""
|
|
SELECT
|
|
sad.*,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
il.origin_country,
|
|
s.name as supplier_name
|
|
FROM stock_adjustment_details sad
|
|
JOIN herb_items h ON sad.herb_item_id = h.herb_item_id
|
|
JOIN inventory_lots il ON sad.lot_id = il.lot_id
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
WHERE sad.adjustment_id = ?
|
|
""", (adjustment_id,))
|
|
|
|
details = [dict(row) for row in cursor.fetchall()]
|
|
adjustment['details'] = details
|
|
|
|
return jsonify({'success': True, 'data': adjustment})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/stock-adjustments', methods=['POST'])
|
|
def create_stock_adjustment():
|
|
"""재고 보정 생성"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 보정 번호 생성 (ADJ-YYYYMMDD-XXXX)
|
|
adjustment_date = data.get('adjustment_date', datetime.now().strftime('%Y-%m-%d'))
|
|
date_str = adjustment_date.replace('-', '')
|
|
|
|
cursor.execute("""
|
|
SELECT MAX(CAST(SUBSTR(adjustment_no, -4) AS INTEGER))
|
|
FROM stock_adjustments
|
|
WHERE adjustment_no LIKE ?
|
|
""", (f'ADJ-{date_str}-%',))
|
|
|
|
max_num = cursor.fetchone()[0]
|
|
next_num = (max_num or 0) + 1
|
|
adjustment_no = f"ADJ-{date_str}-{next_num:04d}"
|
|
|
|
# 보정 헤더 생성
|
|
cursor.execute("""
|
|
INSERT INTO stock_adjustments (adjustment_date, adjustment_no, adjustment_type, notes, created_by)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (
|
|
adjustment_date,
|
|
adjustment_no,
|
|
data['adjustment_type'],
|
|
data.get('notes'),
|
|
data.get('created_by', 'system')
|
|
))
|
|
adjustment_id = cursor.lastrowid
|
|
|
|
# 보정 상세 처리
|
|
for detail in data['details']:
|
|
herb_item_id = detail['herb_item_id']
|
|
lot_id = detail['lot_id']
|
|
quantity_before = detail['quantity_before']
|
|
quantity_after = detail['quantity_after']
|
|
quantity_delta = quantity_after - quantity_before
|
|
|
|
# 보정 상세 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_adjustment_details (adjustment_id, herb_item_id, lot_id,
|
|
quantity_before, quantity_after, quantity_delta, reason)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
adjustment_id, herb_item_id, lot_id,
|
|
quantity_before, quantity_after, quantity_delta,
|
|
detail.get('reason')
|
|
))
|
|
|
|
# 재고 로트 업데이트
|
|
cursor.execute("""
|
|
UPDATE inventory_lots
|
|
SET quantity_onhand = ?,
|
|
is_depleted = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE lot_id = ?
|
|
""", (quantity_after, 1 if quantity_after == 0 else 0, lot_id))
|
|
|
|
# 재고 원장 기록
|
|
cursor.execute("""
|
|
SELECT unit_price_per_g FROM inventory_lots WHERE lot_id = ?
|
|
""", (lot_id,))
|
|
unit_price = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger (event_type, herb_item_id, lot_id,
|
|
quantity_delta, unit_cost_per_g,
|
|
reference_table, reference_id, notes, created_by)
|
|
VALUES ('ADJUST', ?, ?, ?, ?, 'stock_adjustments', ?, ?, ?)
|
|
""", (
|
|
herb_item_id, lot_id, quantity_delta, unit_price,
|
|
adjustment_id, detail.get('reason'), data.get('created_by', 'system')
|
|
))
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '재고 보정이 완료되었습니다',
|
|
'adjustment_id': adjustment_id,
|
|
'adjustment_no': adjustment_no
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 문진표 API ====================
|
|
|
|
@app.route('/api/surveys/templates', methods=['GET'])
|
|
def get_survey_templates():
|
|
"""문진표 템플릿 조회"""
|
|
try:
|
|
category = request.args.get('category')
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
if category:
|
|
cursor.execute("""
|
|
SELECT * FROM survey_templates
|
|
WHERE category = ? AND is_active = 1
|
|
ORDER BY sort_order
|
|
""", (category,))
|
|
else:
|
|
cursor.execute("""
|
|
SELECT * FROM survey_templates
|
|
WHERE is_active = 1
|
|
ORDER BY sort_order
|
|
""")
|
|
|
|
templates = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# JSON 파싱
|
|
for template in templates:
|
|
if template['options']:
|
|
template['options'] = json.loads(template['options'])
|
|
|
|
return jsonify({'success': True, 'data': templates})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/surveys/categories', methods=['GET'])
|
|
def get_survey_categories():
|
|
"""문진표 카테고리 목록"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT DISTINCT category, category_name,
|
|
MIN(sort_order) as min_order,
|
|
COUNT(*) as question_count
|
|
FROM survey_templates
|
|
WHERE is_active = 1
|
|
GROUP BY category
|
|
ORDER BY min_order
|
|
""")
|
|
|
|
categories = [dict(row) for row in cursor.fetchall()]
|
|
return jsonify({'success': True, 'data': categories})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/surveys', methods=['POST'])
|
|
def create_survey():
|
|
"""새 문진표 생성"""
|
|
try:
|
|
data = request.json
|
|
patient_id = data.get('patient_id')
|
|
|
|
# 고유 토큰 생성
|
|
import secrets
|
|
survey_token = secrets.token_urlsafe(16)
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
INSERT INTO patient_surveys (patient_id, survey_token, status)
|
|
VALUES (?, ?, 'PENDING')
|
|
""", (patient_id, survey_token))
|
|
|
|
survey_id = cursor.lastrowid
|
|
conn.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'survey_id': survey_id,
|
|
'survey_token': survey_token,
|
|
'survey_url': f'/survey/{survey_token}'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/surveys/<survey_token>', methods=['GET'])
|
|
def get_survey(survey_token):
|
|
"""문진표 조회 (토큰으로)"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 문진표 기본 정보
|
|
cursor.execute("""
|
|
SELECT s.*, p.name as patient_name, p.phone as patient_phone
|
|
FROM patient_surveys s
|
|
LEFT JOIN patients p ON s.patient_id = p.patient_id
|
|
WHERE s.survey_token = ?
|
|
""", (survey_token,))
|
|
|
|
survey_row = cursor.fetchone()
|
|
if not survey_row:
|
|
return jsonify({'success': False, 'error': '문진표를 찾을 수 없습니다'}), 404
|
|
|
|
survey = dict(survey_row)
|
|
|
|
# 진행 상태 조회
|
|
cursor.execute("""
|
|
SELECT * FROM survey_progress
|
|
WHERE survey_id = ?
|
|
""", (survey['survey_id'],))
|
|
|
|
progress = [dict(row) for row in cursor.fetchall()]
|
|
survey['progress'] = progress
|
|
|
|
# 기존 응답 조회
|
|
cursor.execute("""
|
|
SELECT * FROM survey_responses
|
|
WHERE survey_id = ?
|
|
""", (survey['survey_id'],))
|
|
|
|
responses = [dict(row) for row in cursor.fetchall()]
|
|
survey['responses'] = responses
|
|
|
|
return jsonify({'success': True, 'data': survey})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/surveys/<survey_token>/responses', methods=['POST'])
|
|
def save_survey_responses(survey_token):
|
|
"""문진 응답 저장"""
|
|
try:
|
|
data = request.json
|
|
responses = data.get('responses', [])
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 문진표 확인
|
|
cursor.execute("""
|
|
SELECT survey_id FROM patient_surveys
|
|
WHERE survey_token = ?
|
|
""", (survey_token,))
|
|
|
|
survey_row = cursor.fetchone()
|
|
if not survey_row:
|
|
return jsonify({'success': False, 'error': '문진표를 찾을 수 없습니다'}), 404
|
|
|
|
survey_id = survey_row['survey_id']
|
|
|
|
# 기존 응답 삭제 후 새로 저장 (upsert 방식)
|
|
for response in responses:
|
|
# 기존 응답 삭제
|
|
cursor.execute("""
|
|
DELETE FROM survey_responses
|
|
WHERE survey_id = ? AND question_code = ?
|
|
""", (survey_id, response['question_code']))
|
|
|
|
# 새 응답 저장
|
|
cursor.execute("""
|
|
INSERT INTO survey_responses
|
|
(survey_id, category, question_code, question_text, answer_value, answer_type)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
survey_id,
|
|
response['category'],
|
|
response['question_code'],
|
|
response.get('question_text'),
|
|
json.dumps(response['answer_value'], ensure_ascii=False) if isinstance(response['answer_value'], (list, dict)) else response['answer_value'],
|
|
response.get('answer_type', 'SINGLE')
|
|
))
|
|
|
|
# 상태 업데이트
|
|
cursor.execute("""
|
|
UPDATE patient_surveys
|
|
SET status = 'IN_PROGRESS',
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE survey_id = ?
|
|
""", (survey_id,))
|
|
|
|
conn.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'{len(responses)}개 응답 저장 완료'
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/surveys/<survey_token>/complete', methods=['POST'])
|
|
def complete_survey(survey_token):
|
|
"""문진표 제출 완료"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
UPDATE patient_surveys
|
|
SET status = 'COMPLETED',
|
|
completed_at = CURRENT_TIMESTAMP
|
|
WHERE survey_token = ?
|
|
""", (survey_token,))
|
|
|
|
conn.commit()
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '문진표가 제출되었습니다'
|
|
})
|
|
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) |