시골약사
13b56bc1e9
- herb_item_id ≠ herb_id 문제 해결 - herb_products를 통한 올바른 JOIN 경로 구현 (herb_items → herb_products → herb_masters → herb_master_extended → tags) - /api/herbs, /api/herbs/masters, /api/inventory/summary 모두 정상 작동 - 감초에 효능 태그 추가 (보기, 청열, 해독, 거담, 항염) 이제 조제 페이지 약재 추가 드롭다운 정상 작동 재고 현황 페이지 정상 표시
2507 lines
101 KiB
Python
2507 lines
101 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 het.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
|
|
-- herb_products를 통해 ingredient_code 연결
|
|
LEFT JOIN herb_products hp ON h.insurance_code = hp.product_code
|
|
LEFT JOIN herb_masters hm ON hp.ingredient_code = hm.ingredient_code
|
|
LEFT JOIN herb_master_extended hme ON hm.ingredient_code = hme.ingredient_code
|
|
LEFT JOIN herb_item_tags hit ON hme.herb_id = hit.herb_id
|
|
LEFT JOIN herb_efficacy_tags het ON hit.tag_id = het.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 = []
|
|
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)
|
|
|
|
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,
|
|
-- 제품 정보
|
|
COUNT(DISTINCT p.company_name) as company_count,
|
|
COUNT(DISTINCT p.product_id) as product_count,
|
|
-- 효능 태그
|
|
GROUP_CONCAT(DISTINCT et.tag_name) as efficacy_tags
|
|
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_master_extended hme ON m.ingredient_code = hme.ingredient_code
|
|
LEFT JOIN herb_item_tags hit ON hme.herb_id = hit.herb_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 기준으로 저장
|
|
original_by_code = {}
|
|
original_ingredients = {} # herb_item_id 기준
|
|
|
|
for row in cursor.fetchall():
|
|
ingredient_code = row[0]
|
|
herb_name = row[1]
|
|
grams = row[2]
|
|
|
|
# ingredient_code 기준으로 저장
|
|
original_by_code[ingredient_code] = {
|
|
'herb_name': herb_name,
|
|
'grams': grams
|
|
}
|
|
|
|
# 해당 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_ingredients[herb_id] = {
|
|
'herb_name': herb_name,
|
|
'grams': grams,
|
|
'ingredient_code': ingredient_code
|
|
}
|
|
|
|
# 실제 조제 구성과 비교
|
|
actual_ingredients = {ing['herb_item_id']: ing['grams_per_cheop']
|
|
for ing in data['ingredients']}
|
|
|
|
# 실제 조제의 ingredient_code 수집
|
|
actual_by_code = {}
|
|
for ing in data['ingredients']:
|
|
cursor.execute("SELECT ingredient_code FROM herb_items WHERE herb_item_id = ?",
|
|
(ing['herb_item_id'],))
|
|
result = cursor.fetchone()
|
|
if result:
|
|
ingredient_code = result[0]
|
|
if ingredient_code not in actual_by_code:
|
|
actual_by_code[ingredient_code] = ing['grams_per_cheop']
|
|
|
|
# 추가된 약재 확인
|
|
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
|
|
|
|
# 제거된 약재 확인 (ingredient_code 기준)
|
|
for code, info in original_by_code.items():
|
|
if code not in actual_by_code:
|
|
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))
|
|
|
|
# 재고 차감 처리
|
|
remaining_qty = total_grams
|
|
|
|
# 수동 로트 배분이 있는 경우 (lot_assignments 배열 사용)
|
|
if 'lot_assignments' in ingredient and ingredient['lot_assignments']:
|
|
# 수동 로트 배분 검증
|
|
assigned_total = sum(la['quantity'] for la in ingredient['lot_assignments'])
|
|
if abs(assigned_total - total_grams) > 0.01:
|
|
raise ValueError(f"로트 배분 합계({assigned_total}g)와 필요량({total_grams}g)이 일치하지 않습니다")
|
|
|
|
# 각 로트별로 처리
|
|
for assignment in ingredient['lot_assignments']:
|
|
lot_id = assignment['lot_id']
|
|
requested_qty = assignment['quantity']
|
|
|
|
# 로트 정보 조회
|
|
cursor.execute("""
|
|
SELECT quantity_onhand, unit_price_per_g
|
|
FROM inventory_lots
|
|
WHERE lot_id = ? AND herb_item_id = ? AND is_depleted = 0
|
|
""", (lot_id, herb_item_id))
|
|
|
|
lot_info = cursor.fetchone()
|
|
if not lot_info:
|
|
raise ValueError(f"로트 #{lot_id}를 찾을 수 없거나 사용 불가능합니다")
|
|
|
|
available = lot_info[0]
|
|
unit_price = lot_info[1]
|
|
|
|
if requested_qty > available:
|
|
raise ValueError(f"로트 #{lot_id}의 재고({available}g)가 부족합니다 (요청: {requested_qty}g)")
|
|
|
|
cost = requested_qty * 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, requested_qty, unit_price, cost))
|
|
|
|
# 로트 재고 감소
|
|
new_qty = available - requested_qty
|
|
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, -requested_qty, unit_price, compound_id))
|
|
|
|
# remaining_qty 감소 (중요!)
|
|
remaining_qty -= requested_qty
|
|
|
|
# 자동 로트 선택 (기존 로직)
|
|
else:
|
|
# 원산지가 지정된 경우 해당 원산지만, 아니면 전체에서 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
|
|
-- 효능 태그 조인 (herb_products 경유)
|
|
LEFT JOIN herb_products hp ON h.insurance_code = hp.product_code
|
|
LEFT JOIN herb_masters hm ON COALESCE(h.ingredient_code, hp.ingredient_code) = hm.ingredient_code
|
|
LEFT JOIN herb_master_extended hme ON hm.ingredient_code = hme.ingredient_code
|
|
LEFT JOIN herb_item_tags hit ON hme.herb_id = hit.herb_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 = []
|
|
for row in cursor.fetchall():
|
|
item = dict(row)
|
|
# 효능 태그를 리스트로 변환
|
|
if item['efficacy_tags']:
|
|
item['efficacy_tags'] = item['efficacy_tags'].split(',')
|
|
else:
|
|
item['efficacy_tags'] = []
|
|
inventory.append(item)
|
|
|
|
# 전체 요약
|
|
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
|
|
|
|
# ================ 한약재 확장 정보 API ================
|
|
|
|
@app.route('/api/herbs/<int:herb_id>/extended', methods=['GET'])
|
|
def get_herb_extended_info(herb_id):
|
|
"""약재 확장 정보 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 기본 정보 + 확장 정보 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
hme.*,
|
|
hm.herb_name,
|
|
hm.herb_name_hanja
|
|
FROM herb_master_extended hme
|
|
LEFT JOIN herb_masters hm ON hme.ingredient_code = hm.ingredient_code
|
|
WHERE hme.herb_id = ?
|
|
""", (herb_id,))
|
|
|
|
herb_info = cursor.fetchone()
|
|
|
|
if not herb_info:
|
|
return jsonify({'error': '약재 정보를 찾을 수 없습니다'}), 404
|
|
|
|
# 효능 태그 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
het.tag_name,
|
|
het.tag_category,
|
|
het.description,
|
|
hit.strength
|
|
FROM herb_item_tags hit
|
|
JOIN herb_efficacy_tags het ON hit.tag_id = het.tag_id
|
|
WHERE hit.herb_id = ?
|
|
ORDER BY hit.strength DESC, het.tag_category
|
|
""", (herb_id,))
|
|
|
|
tags = []
|
|
for row in cursor.fetchall():
|
|
tags.append({
|
|
'name': row['tag_name'],
|
|
'category': row['tag_category'],
|
|
'description': row['description'],
|
|
'strength': row['strength']
|
|
})
|
|
|
|
# 안전성 정보 조회
|
|
cursor.execute("""
|
|
SELECT * FROM herb_safety_info
|
|
WHERE herb_id = ?
|
|
""", (herb_id,))
|
|
|
|
safety_info = cursor.fetchone()
|
|
|
|
result = dict(herb_info)
|
|
result['efficacy_tags'] = tags
|
|
result['safety_info'] = dict(safety_info) if safety_info else None
|
|
|
|
return jsonify(result)
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/herbs/<int:herb_id>/extended', methods=['PUT'])
|
|
def update_herb_extended_info(herb_id):
|
|
"""약재 확장 정보 수정"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 업데이트할 필드 동적 생성
|
|
update_fields = []
|
|
update_values = []
|
|
|
|
allowed_fields = [
|
|
'property', 'taste', 'meridian_tropism',
|
|
'main_effects', 'indications', 'contraindications',
|
|
'precautions', 'dosage_range', 'dosage_max',
|
|
'preparation_method', 'active_compounds',
|
|
'pharmacological_effects', 'clinical_applications'
|
|
]
|
|
|
|
for field in allowed_fields:
|
|
if field in data:
|
|
update_fields.append(f"{field} = ?")
|
|
update_values.append(data[field])
|
|
|
|
if update_fields:
|
|
update_values.append(herb_id)
|
|
cursor.execute(f"""
|
|
UPDATE herb_master_extended
|
|
SET {', '.join(update_fields)},
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE herb_id = ?
|
|
""", update_values)
|
|
|
|
# 변경 로그 기록
|
|
cursor.execute("""
|
|
INSERT INTO data_update_logs
|
|
(update_type, source, target_table, target_id, after_data)
|
|
VALUES ('MANUAL', 'API', 'herb_master_extended', ?, ?)
|
|
""", (herb_id, json.dumps(data, ensure_ascii=False)))
|
|
|
|
conn.commit()
|
|
|
|
return jsonify({'success': True, 'message': '정보가 업데이트되었습니다'})
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/herbs/<int:herb_id>/tags', methods=['GET'])
|
|
def get_herb_tags(herb_id):
|
|
"""약재 효능 태그 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT
|
|
het.tag_id,
|
|
het.tag_name,
|
|
het.tag_category,
|
|
het.description,
|
|
hit.strength
|
|
FROM herb_item_tags hit
|
|
JOIN herb_efficacy_tags het ON hit.tag_id = het.tag_id
|
|
WHERE hit.herb_id = ?
|
|
ORDER BY hit.strength DESC
|
|
""", (herb_id,))
|
|
|
|
tags = []
|
|
for row in cursor.fetchall():
|
|
tags.append({
|
|
'tag_id': row['tag_id'],
|
|
'name': row['tag_name'],
|
|
'category': row['tag_category'],
|
|
'description': row['description'],
|
|
'strength': row['strength']
|
|
})
|
|
|
|
return jsonify(tags)
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/herbs/<int:herb_id>/tags', methods=['POST'])
|
|
def add_herb_tag(herb_id):
|
|
"""약재에 효능 태그 추가"""
|
|
try:
|
|
data = request.json
|
|
tag_id = data.get('tag_id')
|
|
strength = data.get('strength', 3)
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
INSERT OR REPLACE INTO herb_item_tags
|
|
(herb_id, tag_id, strength)
|
|
VALUES (?, ?, ?)
|
|
""", (herb_id, tag_id, strength))
|
|
|
|
conn.commit()
|
|
|
|
return jsonify({'success': True, 'message': '태그가 추가되었습니다'})
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/herbs/search-by-efficacy', methods=['GET'])
|
|
def search_herbs_by_efficacy():
|
|
"""효능별 약재 검색"""
|
|
try:
|
|
tag_names = request.args.getlist('tags')
|
|
|
|
if not tag_names:
|
|
return jsonify({'error': '검색할 태그를 지정해주세요'}), 400
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
placeholders = ','.join('?' * len(tag_names))
|
|
cursor.execute(f"""
|
|
SELECT DISTINCT
|
|
hme.herb_id,
|
|
hme.name_korean,
|
|
hme.name_hanja,
|
|
hme.main_effects,
|
|
GROUP_CONCAT(het.tag_name) as tags
|
|
FROM herb_master_extended hme
|
|
JOIN herb_item_tags hit ON hme.herb_id = hit.herb_id
|
|
JOIN herb_efficacy_tags het ON hit.tag_id = het.tag_id
|
|
WHERE het.tag_name IN ({placeholders})
|
|
GROUP BY hme.herb_id
|
|
ORDER BY hme.name_korean
|
|
""", tag_names)
|
|
|
|
results = []
|
|
for row in cursor.fetchall():
|
|
results.append({
|
|
'herb_id': row['herb_id'],
|
|
'name_korean': row['name_korean'],
|
|
'name_hanja': row['name_hanja'],
|
|
'main_effects': row['main_effects'],
|
|
'tags': row['tags'].split(',') if row['tags'] else []
|
|
})
|
|
|
|
return jsonify(results)
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/prescription-check', methods=['POST'])
|
|
def check_prescription_safety():
|
|
"""처방 안전성 검증"""
|
|
try:
|
|
data = request.json
|
|
herb_ids = data.get('herb_ids', [])
|
|
|
|
if len(herb_ids) < 2:
|
|
return jsonify({'safe': True, 'warnings': []})
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
warnings = []
|
|
|
|
# 약재 조합 규칙 확인
|
|
for i in range(len(herb_ids)):
|
|
for j in range(i + 1, len(herb_ids)):
|
|
cursor.execute("""
|
|
SELECT
|
|
relationship_type,
|
|
description,
|
|
severity_level,
|
|
is_absolute
|
|
FROM prescription_rules
|
|
WHERE (herb1_id = ? AND herb2_id = ?)
|
|
OR (herb1_id = ? AND herb2_id = ?)
|
|
""", (herb_ids[i], herb_ids[j], herb_ids[j], herb_ids[i]))
|
|
|
|
rule = cursor.fetchone()
|
|
if rule:
|
|
# 상반(相反), 상살(相殺) 등 위험한 관계 체크
|
|
if rule['relationship_type'] in ['상반', '상살']:
|
|
warnings.append({
|
|
'type': 'danger',
|
|
'herbs': [herb_ids[i], herb_ids[j]],
|
|
'relationship': rule['relationship_type'],
|
|
'description': rule['description'],
|
|
'is_absolute': rule['is_absolute']
|
|
})
|
|
elif rule['relationship_type'] == '상외':
|
|
warnings.append({
|
|
'type': 'warning',
|
|
'herbs': [herb_ids[i], herb_ids[j]],
|
|
'relationship': rule['relationship_type'],
|
|
'description': rule['description']
|
|
})
|
|
|
|
# 절대 금기 사항이 있으면 안전하지 않음
|
|
is_safe = not any(w.get('is_absolute') for w in warnings)
|
|
|
|
return jsonify({
|
|
'safe': is_safe,
|
|
'warnings': warnings
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'error': str(e)}), 500
|
|
|
|
@app.route('/api/efficacy-tags', methods=['GET'])
|
|
def get_all_efficacy_tags():
|
|
"""모든 효능 태그 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT * FROM herb_efficacy_tags
|
|
ORDER BY tag_category, tag_name
|
|
""")
|
|
|
|
tags = []
|
|
for row in cursor.fetchall():
|
|
tags.append({
|
|
'tag_id': row['tag_id'],
|
|
'name': row['tag_name'],
|
|
'category': row['tag_category'],
|
|
'description': row['description']
|
|
})
|
|
|
|
return jsonify(tags)
|
|
|
|
except Exception as e:
|
|
return jsonify({'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) |