시골약사
9fa1f7a031
- formula_ingredients 테이블이 herb_item_id 대신 ingredient_code 사용하도록 변경 - GET /api/formulas/<id>/ingredients API 개선 - ingredient_code 기반 조회로 변경 - available_products 배열 추가 (재고 있는 모든 제품 포함) - total_available_stock, product_count 필드 추가 - 같은 주성분을 가진 모든 제품의 재고 정보를 반환하도록 수정 이제 처방에서 특정 제품이 아닌 주성분을 지정하여 재고가 있는 대체 제품을 자동으로 선택 가능
1865 lines
74 KiB
Python
1865 lines
74 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
|
|
|
|
# ==================== 처방 관리 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:
|
|
|
|
# 입고장 번호 생성 (PR-YYYYMMDD-XXXX)
|
|
date_str = str(receipt_date).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 = group['total_amount'].sum()
|
|
cursor.execute("""
|
|
INSERT INTO purchase_receipts (supplier_id, receipt_date, receipt_no, total_amount, source_file)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (supplier_id, str(receipt_date), receipt_no, total_amount, filename))
|
|
receipt_id = cursor.lastrowid
|
|
|
|
# 입고장 라인 생성
|
|
for _, row in group.iterrows():
|
|
# 약재 확인/생성
|
|
cursor.execute("""
|
|
SELECT herb_item_id FROM herb_items
|
|
WHERE insurance_code = ? OR herb_name = ?
|
|
""", (row.get('insurance_code'), row['herb_name']))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
cursor.execute("""
|
|
INSERT INTO herb_items (insurance_code, herb_name)
|
|
VALUES (?, ?)
|
|
""", (row.get('insurance_code'), row['herb_name']))
|
|
herb_item_id = cursor.lastrowid
|
|
else:
|
|
herb_item_id = herb[0]
|
|
|
|
# 단가 계산 (총액 / 수량)
|
|
quantity = float(row['quantity'])
|
|
total = float(row['total_amount'])
|
|
unit_price = total / quantity if quantity > 0 else 0
|
|
|
|
# 입고장 라인 생성
|
|
cursor.execute("""
|
|
INSERT INTO purchase_receipt_lines
|
|
(receipt_id, herb_item_id, origin_country, quantity_g, unit_price_per_g, line_total)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (receipt_id, herb_item_id, row.get('origin_country'),
|
|
quantity, unit_price, total))
|
|
line_id = cursor.lastrowid
|
|
|
|
# 재고 로트 생성
|
|
cursor.execute("""
|
|
INSERT INTO inventory_lots
|
|
(herb_item_id, supplier_id, receipt_line_id, received_date, origin_country,
|
|
unit_price_per_g, quantity_received, quantity_onhand)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (herb_item_id, supplier_id, line_id, str(receipt_date),
|
|
row.get('origin_country'), unit_price, quantity, quantity))
|
|
lot_id = cursor.lastrowid
|
|
|
|
# 재고 원장 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger
|
|
(event_type, herb_item_id, lot_id, quantity_delta, unit_cost_per_g,
|
|
reference_table, reference_id)
|
|
VALUES ('RECEIPT', ?, ?, ?, ?, 'purchase_receipts', ?)
|
|
""", (herb_item_id, lot_id, quantity, unit_price, receipt_id))
|
|
|
|
processed_rows += 1
|
|
processed_items.add(row['herb_name'])
|
|
|
|
# 응답 메시지 생성
|
|
format_name = {
|
|
'hanisarang': '한의사랑',
|
|
'haninfo': '한의정보'
|
|
}.get(summary['format_type'], '알 수 없음')
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'{format_name} 형식 입고 데이터가 성공적으로 처리되었습니다',
|
|
'filename': filename,
|
|
'summary': {
|
|
'format': format_name,
|
|
'processed_rows': processed_rows,
|
|
'total_items': len(processed_items),
|
|
'total_quantity': f"{summary['total_quantity']:,.0f}g",
|
|
'total_amount': f"{summary['total_amount']:,.0f}원"
|
|
}
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 입고장 조회/관리 API ====================
|
|
|
|
@app.route('/api/purchase-receipts', methods=['GET'])
|
|
def get_purchase_receipts():
|
|
"""입고장 목록 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 날짜 범위 파라미터
|
|
start_date = request.args.get('start_date')
|
|
end_date = request.args.get('end_date')
|
|
supplier_id = request.args.get('supplier_id')
|
|
|
|
query = """
|
|
SELECT
|
|
pr.receipt_id,
|
|
pr.receipt_date,
|
|
pr.receipt_no,
|
|
pr.source_file,
|
|
pr.created_at,
|
|
s.name as supplier_name,
|
|
s.supplier_id,
|
|
COUNT(prl.line_id) as line_count,
|
|
SUM(prl.quantity_g) as total_quantity,
|
|
SUM(prl.line_total) as total_amount
|
|
FROM purchase_receipts pr
|
|
JOIN suppliers s ON pr.supplier_id = s.supplier_id
|
|
LEFT JOIN purchase_receipt_lines prl ON pr.receipt_id = prl.receipt_id
|
|
WHERE 1=1
|
|
"""
|
|
params = []
|
|
|
|
if start_date:
|
|
query += " AND pr.receipt_date >= ?"
|
|
params.append(start_date)
|
|
if end_date:
|
|
query += " AND pr.receipt_date <= ?"
|
|
params.append(end_date)
|
|
if supplier_id:
|
|
query += " AND pr.supplier_id = ?"
|
|
params.append(supplier_id)
|
|
|
|
query += " GROUP BY pr.receipt_id ORDER BY pr.receipt_date DESC, pr.created_at DESC"
|
|
|
|
cursor.execute(query, params)
|
|
receipts = []
|
|
for row in cursor.fetchall():
|
|
receipt = dict(row)
|
|
# 타입 변환 (bytes 문제 해결)
|
|
for key, value in receipt.items():
|
|
if isinstance(value, bytes):
|
|
# bytes를 float로 변환 시도
|
|
try:
|
|
import struct
|
|
receipt[key] = struct.unpack('d', value)[0]
|
|
except:
|
|
receipt[key] = float(0)
|
|
elif key in ['receipt_date', 'created_at'] and value is not None:
|
|
receipt[key] = str(value)
|
|
|
|
# total_amount와 total_quantity 반올림
|
|
if 'total_amount' in receipt and receipt['total_amount'] is not None:
|
|
receipt['total_amount'] = round(float(receipt['total_amount']), 2)
|
|
if 'total_quantity' in receipt and receipt['total_quantity'] is not None:
|
|
receipt['total_quantity'] = round(float(receipt['total_quantity']), 2)
|
|
|
|
receipts.append(receipt)
|
|
|
|
return jsonify({'success': True, 'data': receipts})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>', methods=['GET'])
|
|
def get_purchase_receipt_detail(receipt_id):
|
|
"""입고장 상세 조회"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 입고장 헤더 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
pr.*,
|
|
s.name as supplier_name,
|
|
s.business_no as supplier_business_no,
|
|
s.phone as supplier_phone
|
|
FROM purchase_receipts pr
|
|
JOIN suppliers s ON pr.supplier_id = s.supplier_id
|
|
WHERE pr.receipt_id = ?
|
|
""", (receipt_id,))
|
|
|
|
receipt = cursor.fetchone()
|
|
if not receipt:
|
|
return jsonify({'success': False, 'error': '입고장을 찾을 수 없습니다'}), 404
|
|
|
|
receipt_data = dict(receipt)
|
|
|
|
# 입고장 상세 라인 조회
|
|
cursor.execute("""
|
|
SELECT
|
|
prl.*,
|
|
h.herb_name,
|
|
h.insurance_code,
|
|
il.lot_id,
|
|
il.quantity_onhand as current_stock
|
|
FROM purchase_receipt_lines prl
|
|
JOIN herb_items h ON prl.herb_item_id = h.herb_item_id
|
|
LEFT JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.receipt_id = ?
|
|
ORDER BY prl.line_id
|
|
""", (receipt_id,))
|
|
|
|
lines = [dict(row) for row in cursor.fetchall()]
|
|
receipt_data['lines'] = lines
|
|
|
|
return jsonify({'success': True, 'data': receipt_data})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>/lines/<int:line_id>', methods=['PUT'])
|
|
def update_purchase_receipt_line(receipt_id, line_id):
|
|
"""입고장 라인 수정"""
|
|
try:
|
|
data = request.json
|
|
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 기존 라인 정보 조회
|
|
cursor.execute("""
|
|
SELECT prl.*, il.lot_id, il.quantity_onhand, il.quantity_received
|
|
FROM purchase_receipt_lines prl
|
|
LEFT JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.line_id = ? AND prl.receipt_id = ?
|
|
""", (line_id, receipt_id))
|
|
|
|
old_line = cursor.fetchone()
|
|
if not old_line:
|
|
return jsonify({'success': False, 'error': '입고 라인을 찾을 수 없습니다'}), 404
|
|
|
|
# 재고 사용 여부 확인
|
|
if old_line['quantity_onhand'] != old_line['quantity_received']:
|
|
used_qty = old_line['quantity_received'] - old_line['quantity_onhand']
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'이미 {used_qty}g이 사용되어 수정할 수 없습니다'
|
|
}), 400
|
|
|
|
# 수정 가능한 필드만 업데이트
|
|
update_fields = []
|
|
params = []
|
|
|
|
if 'quantity_g' in data:
|
|
update_fields.append('quantity_g = ?')
|
|
params.append(data['quantity_g'])
|
|
|
|
if 'unit_price_per_g' in data:
|
|
update_fields.append('unit_price_per_g = ?')
|
|
params.append(data['unit_price_per_g'])
|
|
|
|
if 'line_total' in data:
|
|
update_fields.append('line_total = ?')
|
|
params.append(data['line_total'])
|
|
elif 'quantity_g' in data and 'unit_price_per_g' in data:
|
|
# 자동 계산
|
|
line_total = float(data['quantity_g']) * float(data['unit_price_per_g'])
|
|
update_fields.append('line_total = ?')
|
|
params.append(line_total)
|
|
|
|
if 'origin_country' in data:
|
|
update_fields.append('origin_country = ?')
|
|
params.append(data['origin_country'])
|
|
|
|
if not update_fields:
|
|
return jsonify({'success': False, 'error': '수정할 내용이 없습니다'}), 400
|
|
|
|
# 입고장 라인 업데이트
|
|
params.append(line_id)
|
|
cursor.execute(f"""
|
|
UPDATE purchase_receipt_lines
|
|
SET {', '.join(update_fields)}
|
|
WHERE line_id = ?
|
|
""", params)
|
|
|
|
# 재고 로트 업데이트 (수량 변경시)
|
|
if 'quantity_g' in data and old_line['lot_id']:
|
|
cursor.execute("""
|
|
UPDATE inventory_lots
|
|
SET quantity_received = ?, quantity_onhand = ?
|
|
WHERE lot_id = ?
|
|
""", (data['quantity_g'], data['quantity_g'], old_line['lot_id']))
|
|
|
|
# 재고 원장에 조정 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger
|
|
(event_type, herb_item_id, lot_id, quantity_delta, notes, reference_table, reference_id)
|
|
VALUES ('ADJUST',
|
|
(SELECT herb_item_id FROM purchase_receipt_lines WHERE line_id = ?),
|
|
?, ?, '입고장 수정', 'purchase_receipt_lines', ?)
|
|
""", (line_id, old_line['lot_id'],
|
|
float(data['quantity_g']) - float(old_line['quantity_g']), line_id))
|
|
|
|
# 입고장 헤더의 총액 업데이트
|
|
cursor.execute("""
|
|
UPDATE purchase_receipts
|
|
SET total_amount = (
|
|
SELECT SUM(line_total)
|
|
FROM purchase_receipt_lines
|
|
WHERE receipt_id = ?
|
|
),
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE receipt_id = ?
|
|
""", (receipt_id, receipt_id))
|
|
|
|
return jsonify({'success': True, 'message': '입고 라인이 수정되었습니다'})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/purchase-receipts/<int:receipt_id>', methods=['DELETE'])
|
|
def delete_purchase_receipt(receipt_id):
|
|
"""입고장 삭제 (재고 사용 확인 후)"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 재고 사용 여부 확인
|
|
cursor.execute("""
|
|
SELECT
|
|
COUNT(*) as used_count,
|
|
SUM(il.quantity_received - il.quantity_onhand) as used_quantity
|
|
FROM purchase_receipt_lines prl
|
|
JOIN inventory_lots il ON prl.line_id = il.receipt_line_id
|
|
WHERE prl.receipt_id = ?
|
|
AND il.quantity_onhand < il.quantity_received
|
|
""", (receipt_id,))
|
|
|
|
usage = cursor.fetchone()
|
|
if usage['used_count'] > 0:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'{usage["used_count"]}개 품목에서 {usage["used_quantity"]}g이 이미 사용되어 삭제할 수 없습니다'
|
|
}), 400
|
|
|
|
# 삭제 순서 중요: 참조하는 테이블부터 삭제
|
|
# 1. 재고 원장 기록 삭제 (lot_id를 참조)
|
|
cursor.execute("""
|
|
DELETE FROM stock_ledger
|
|
WHERE lot_id IN (
|
|
SELECT lot_id FROM inventory_lots
|
|
WHERE receipt_line_id IN (
|
|
SELECT line_id FROM purchase_receipt_lines WHERE receipt_id = ?
|
|
)
|
|
)
|
|
""", (receipt_id,))
|
|
|
|
# 2. 재고 로트 삭제 (receipt_line_id를 참조)
|
|
cursor.execute("""
|
|
DELETE FROM inventory_lots
|
|
WHERE receipt_line_id IN (
|
|
SELECT line_id FROM purchase_receipt_lines WHERE receipt_id = ?
|
|
)
|
|
""", (receipt_id,))
|
|
|
|
# 3. 입고장 라인 삭제 (receipt_id를 참조)
|
|
cursor.execute("DELETE FROM purchase_receipt_lines WHERE receipt_id = ?", (receipt_id,))
|
|
|
|
# 4. 입고장 헤더 삭제
|
|
cursor.execute("DELETE FROM purchase_receipts WHERE receipt_id = ?", (receipt_id,))
|
|
|
|
return jsonify({'success': True, 'message': '입고장이 삭제되었습니다'})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 조제 관리 API ====================
|
|
|
|
@app.route('/api/compounds', methods=['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
|
|
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()
|
|
|
|
# 조제 마스터 생성
|
|
cursor.execute("""
|
|
INSERT INTO compounds (patient_id, formula_id, compound_date,
|
|
je_count, cheop_total, pouch_total,
|
|
prescription_no, notes, created_by)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""", (
|
|
data.get('patient_id'),
|
|
data.get('formula_id'),
|
|
data.get('compound_date', datetime.now().strftime('%Y-%m-%d')),
|
|
data['je_count'],
|
|
data['cheop_total'],
|
|
data['pouch_total'],
|
|
data.get('prescription_no'),
|
|
data.get('notes'),
|
|
data.get('created_by', 'system')
|
|
))
|
|
compound_id = cursor.lastrowid
|
|
|
|
total_cost = 0
|
|
|
|
# 조제 약재 처리
|
|
for ingredient in data['ingredients']:
|
|
herb_item_id = ingredient['herb_item_id']
|
|
total_grams = ingredient['total_grams']
|
|
origin_country = ingredient.get('origin_country') # 원산지 선택 정보
|
|
|
|
# 조제 약재 구성 기록
|
|
cursor.execute("""
|
|
INSERT INTO compound_ingredients (compound_id, herb_item_id,
|
|
grams_per_cheop, total_grams)
|
|
VALUES (?, ?, ?, ?)
|
|
""", (compound_id, herb_item_id,
|
|
ingredient['grams_per_cheop'], total_grams))
|
|
|
|
# 재고 차감 (FIFO 방식 - 원산지 지정 시 해당 원산지만)
|
|
remaining_qty = total_grams
|
|
|
|
# 원산지가 지정된 경우 해당 원산지만, 아니면 전체에서 FIFO
|
|
if origin_country and origin_country != 'auto':
|
|
cursor.execute("""
|
|
SELECT lot_id, quantity_onhand, unit_price_per_g
|
|
FROM inventory_lots
|
|
WHERE herb_item_id = ? AND is_depleted = 0 AND quantity_onhand > 0
|
|
AND origin_country = ?
|
|
ORDER BY unit_price_per_g, received_date, lot_id
|
|
""", (herb_item_id, origin_country))
|
|
else:
|
|
# 자동 선택: 가격이 저렴한 것부터
|
|
cursor.execute("""
|
|
SELECT lot_id, quantity_onhand, unit_price_per_g
|
|
FROM inventory_lots
|
|
WHERE herb_item_id = ? AND is_depleted = 0 AND quantity_onhand > 0
|
|
ORDER BY unit_price_per_g, received_date, lot_id
|
|
""", (herb_item_id,))
|
|
|
|
lots = cursor.fetchall()
|
|
|
|
for lot in lots:
|
|
if remaining_qty <= 0:
|
|
break
|
|
|
|
lot_id = lot[0]
|
|
available = lot[1]
|
|
unit_price = lot[2]
|
|
|
|
used = min(remaining_qty, available)
|
|
cost = used * unit_price
|
|
total_cost += cost
|
|
|
|
# 소비 내역 기록
|
|
cursor.execute("""
|
|
INSERT INTO compound_consumptions (compound_id, herb_item_id, lot_id,
|
|
quantity_used, unit_cost_per_g, cost_amount)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""", (compound_id, herb_item_id, lot_id, used, unit_price, cost))
|
|
|
|
# 로트 재고 감소
|
|
new_qty = available - used
|
|
cursor.execute("""
|
|
UPDATE inventory_lots
|
|
SET quantity_onhand = ?, is_depleted = ?
|
|
WHERE lot_id = ?
|
|
""", (new_qty, 1 if new_qty == 0 else 0, lot_id))
|
|
|
|
# 재고 원장 기록
|
|
cursor.execute("""
|
|
INSERT INTO stock_ledger (event_type, herb_item_id, lot_id,
|
|
quantity_delta, unit_cost_per_g,
|
|
reference_table, reference_id)
|
|
VALUES ('CONSUME', ?, ?, ?, ?, 'compounds', ?)
|
|
""", (herb_item_id, lot_id, -used, unit_price, compound_id))
|
|
|
|
remaining_qty -= used
|
|
|
|
if remaining_qty > 0:
|
|
raise Exception(f"재고 부족: {ingredient.get('herb_name', herb_item_id)}")
|
|
|
|
# 총 원가 업데이트
|
|
cursor.execute("""
|
|
UPDATE compounds SET cost_total = ? WHERE compound_id = ?
|
|
""", (total_cost, compound_id))
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': '조제가 완료되었습니다',
|
|
'compound_id': compound_id,
|
|
'total_cost': total_cost
|
|
})
|
|
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 재고 원장 API ====================
|
|
|
|
@app.route('/api/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
|
|
|
|
# 가용 로트 목록 (소진되지 않은 재고)
|
|
cursor.execute("""
|
|
SELECT
|
|
lot_id,
|
|
origin_country,
|
|
quantity_onhand,
|
|
unit_price_per_g,
|
|
received_date,
|
|
supplier_id
|
|
FROM inventory_lots
|
|
WHERE herb_item_id = ?
|
|
AND is_depleted = 0
|
|
AND quantity_onhand > 0
|
|
ORDER BY origin_country, unit_price_per_g, received_date
|
|
""", (herb_item_id,))
|
|
|
|
lots = []
|
|
for row in cursor.fetchall():
|
|
lots.append({
|
|
'lot_id': row[0],
|
|
'origin_country': row[1] or '미지정',
|
|
'quantity_onhand': row[2],
|
|
'unit_price_per_g': row[3],
|
|
'received_date': row[4],
|
|
'supplier_id': row[5]
|
|
})
|
|
|
|
# 원산지별 요약
|
|
origin_summary = {}
|
|
for lot in lots:
|
|
origin = lot['origin_country']
|
|
if origin not in origin_summary:
|
|
origin_summary[origin] = {
|
|
'origin_country': origin,
|
|
'total_quantity': 0,
|
|
'min_price': float('inf'),
|
|
'max_price': 0,
|
|
'lot_count': 0,
|
|
'lots': []
|
|
}
|
|
|
|
origin_summary[origin]['total_quantity'] += lot['quantity_onhand']
|
|
origin_summary[origin]['min_price'] = min(origin_summary[origin]['min_price'], lot['unit_price_per_g'])
|
|
origin_summary[origin]['max_price'] = max(origin_summary[origin]['max_price'], lot['unit_price_per_g'])
|
|
origin_summary[origin]['lot_count'] += 1
|
|
origin_summary[origin]['lots'].append(lot)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': {
|
|
'herb_name': herb[0],
|
|
'insurance_code': herb[1],
|
|
'origins': list(origin_summary.values()),
|
|
'total_quantity': sum(lot['quantity_onhand'] for lot in lots)
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# ==================== 재고 현황 API ====================
|
|
|
|
@app.route('/api/inventory/summary', methods=['GET'])
|
|
def get_inventory_summary():
|
|
"""재고 현황 요약 - 원산지별 구분 표시"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
SELECT
|
|
h.herb_item_id,
|
|
h.insurance_code,
|
|
h.herb_name,
|
|
COALESCE(SUM(il.quantity_onhand), 0) as total_quantity,
|
|
COUNT(DISTINCT il.lot_id) as lot_count,
|
|
COUNT(DISTINCT il.origin_country) as origin_count,
|
|
AVG(il.unit_price_per_g) as avg_price,
|
|
MIN(il.unit_price_per_g) as min_price,
|
|
MAX(il.unit_price_per_g) as max_price,
|
|
COALESCE(SUM(il.quantity_onhand * il.unit_price_per_g), 0) as total_value,
|
|
GROUP_CONCAT(DISTINCT et.tag_name) as efficacy_tags
|
|
FROM herb_items h
|
|
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id AND il.is_depleted = 0
|
|
LEFT JOIN herb_item_tags hit ON h.herb_item_id = hit.herb_item_id
|
|
LEFT JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
|
|
GROUP BY h.herb_item_id, h.insurance_code, h.herb_name
|
|
HAVING total_quantity > 0
|
|
ORDER BY h.herb_name
|
|
""")
|
|
inventory = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 태그를 리스트로 변환
|
|
for item in inventory:
|
|
if item['efficacy_tags']:
|
|
item['efficacy_tags'] = item['efficacy_tags'].split(',')
|
|
else:
|
|
item['efficacy_tags'] = []
|
|
|
|
# 전체 요약
|
|
total_value = sum(item['total_value'] for item in inventory)
|
|
total_items = len(inventory)
|
|
|
|
# 주성분코드 기준 보유 현황 추가
|
|
cursor.execute("""
|
|
SELECT COUNT(DISTINCT ingredient_code)
|
|
FROM herb_masters
|
|
""")
|
|
total_ingredient_codes = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
SELECT COUNT(DISTINCT h.ingredient_code)
|
|
FROM herb_items h
|
|
INNER JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id
|
|
WHERE il.quantity_onhand > 0 AND il.is_depleted = 0
|
|
AND h.ingredient_code IS NOT NULL
|
|
""")
|
|
owned_ingredient_codes = cursor.fetchone()[0]
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'data': inventory,
|
|
'summary': {
|
|
'total_items': total_items,
|
|
'total_value': total_value,
|
|
'total_ingredient_codes': total_ingredient_codes, # 전체 급여 약재 수
|
|
'owned_ingredient_codes': owned_ingredient_codes, # 보유 약재 수
|
|
'coverage_rate': round(owned_ingredient_codes * 100 / total_ingredient_codes, 1) if total_ingredient_codes > 0 else 0 # 보유율
|
|
}
|
|
})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
@app.route('/api/inventory/detail/<int:herb_item_id>', methods=['GET'])
|
|
def get_inventory_detail(herb_item_id):
|
|
"""약재별 재고 상세 - 원산지별로 구분"""
|
|
try:
|
|
with get_db() as conn:
|
|
cursor = conn.cursor()
|
|
|
|
# 약재 기본 정보
|
|
cursor.execute("""
|
|
SELECT herb_item_id, insurance_code, herb_name
|
|
FROM herb_items
|
|
WHERE herb_item_id = ?
|
|
""", (herb_item_id,))
|
|
herb = cursor.fetchone()
|
|
|
|
if not herb:
|
|
return jsonify({'success': False, 'error': '약재를 찾을 수 없습니다'}), 404
|
|
|
|
herb_data = dict(herb)
|
|
|
|
# 원산지별 재고 정보
|
|
cursor.execute("""
|
|
SELECT
|
|
il.lot_id,
|
|
il.origin_country,
|
|
il.quantity_onhand,
|
|
il.unit_price_per_g,
|
|
il.received_date,
|
|
il.supplier_id,
|
|
s.name as supplier_name,
|
|
il.quantity_onhand * il.unit_price_per_g as lot_value
|
|
FROM inventory_lots il
|
|
LEFT JOIN suppliers s ON il.supplier_id = s.supplier_id
|
|
WHERE il.herb_item_id = ? AND il.is_depleted = 0
|
|
ORDER BY il.origin_country, il.unit_price_per_g, il.received_date
|
|
""", (herb_item_id,))
|
|
|
|
lots = [dict(row) for row in cursor.fetchall()]
|
|
|
|
# 원산지별 그룹화
|
|
by_origin = {}
|
|
for lot in lots:
|
|
origin = lot['origin_country'] or '미지정'
|
|
if origin not in by_origin:
|
|
by_origin[origin] = {
|
|
'origin_country': origin,
|
|
'lots': [],
|
|
'total_quantity': 0,
|
|
'total_value': 0,
|
|
'min_price': float('inf'),
|
|
'max_price': 0,
|
|
'avg_price': 0
|
|
}
|
|
|
|
by_origin[origin]['lots'].append(lot)
|
|
by_origin[origin]['total_quantity'] += lot['quantity_onhand']
|
|
by_origin[origin]['total_value'] += lot['lot_value']
|
|
by_origin[origin]['min_price'] = min(by_origin[origin]['min_price'], lot['unit_price_per_g'])
|
|
by_origin[origin]['max_price'] = max(by_origin[origin]['max_price'], lot['unit_price_per_g'])
|
|
|
|
# 평균 단가 계산
|
|
for origin_data in by_origin.values():
|
|
if origin_data['total_quantity'] > 0:
|
|
origin_data['avg_price'] = origin_data['total_value'] / origin_data['total_quantity']
|
|
|
|
herb_data['origins'] = list(by_origin.values())
|
|
herb_data['total_origins'] = len(by_origin)
|
|
|
|
return jsonify({'success': True, 'data': herb_data})
|
|
except Exception as e:
|
|
return jsonify({'success': False, 'error': str(e)}), 500
|
|
|
|
# 서버 실행
|
|
# ==================== 재고 보정 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) |