thug0bin/kdrug-inventory-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
f3f1efd8c2
kdrug-inventory-system/app.py
시골약사 f3f1efd8c2 feat: 판매관리 시스템 Phase 1 및 마일리지 시스템 구현 
- 판매 관리 기능 추가
  - compounds 테이블에 판매 관련 컬럼 추가 (payment_method, discount_rate, delivery_method 등)
  - 판매 상태 관리 (조제완료→결제대기→결제완료→배송대기→배송완료)
  - 판매 처리 모달 UI 구현
  - 9가지 상태별 뱃지 표시

- 마일리지 시스템 구축
  - patients 테이블에 마일리지 컬럼 추가 (balance, earned, used)
  - mileage_transactions 테이블 생성 (거래 이력 관리)
  - 마일리지 사용/적립 기능 구현

- 복합 결제 기능
  - 할인율(%) / 할인액(원) 직접 입력 선택 가능
  - 마일리지 + 현금 + 카드 + 계좌이체 복합 결제
  - 결제 금액 자동 검증
  - 결제 방법 자동 분류 (복합결제 지원)

- API 엔드포인트 추가
  - POST /api/compounds/<id>/status (상태 업데이트)
  - PUT /api/compounds/<id>/price (가격 조정)
  - GET /api/sales/statistics (판매 통계)

- 데이터베이스 설정 통합
  - config.py 생성하여 DB 경로 중앙화

TODO: 처방별 기본가격 정책 시스템 (price_policies 테이블 활용)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-02-18 05:42:29 +00:00

3064 lines
124 KiB
Python
Raw Blame History

#!/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
from config import DATABASE_PATH, STATIC_PATH, TEMPLATES_PATH
# Flask 앱 초기화
app = Flask(__name__, static_folder='static', template_folder='templates')
app.config['SECRET_KEY'] = 'your-secret-key-change-in-production'
app.config['DATABASE'] = str(DATABASE_PATH) # config.py의 경로 사용
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
-- 간단한 JOIN: ingredient_code로 직접 연결
LEFT JOIN herb_products hp ON h.insurance_code = hp.product_code
LEFT JOIN herb_item_tags hit ON COALESCE(h.ingredient_code, hp.ingredient_code) = hit.ingredient_code
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,
-- 확장 정보
hme.herb_id,
hme.property,
hme.taste,
hme.meridian_tropism,
hme.main_effects,
-- 재고 정보
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 herb_master_extended hme ON m.ingredient_code = hme.ingredient_code
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
-- 간단한 JOIN: ingredient_code로 직접 연결
LEFT JOIN herb_item_tags hit ON m.ingredient_code = hit.ingredient_code
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, hme.herb_id, hme.property, hme.taste, hme.meridian_tropism, hme.main_effects, 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, efficacy
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, efficacy, 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('efficacy'),
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,
-- 해당 성분코드를 가진 제품들의 재고 합계
COALESCE((
SELECT SUM(il.quantity_onhand)
FROM herb_items hi
LEFT JOIN inventory_lots il ON hi.herb_item_id = il.herb_item_id
WHERE hi.ingredient_code = fi.ingredient_code
AND il.is_depleted = 0
), 0) as stock_quantity
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
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
@app.route('/api/formulas/<int:formula_id>', methods=['GET'])
def get_formula_detail(formula_id):
"""처방 상세 정보 조회"""
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, efficacy, created_at, updated_at
FROM formulas
WHERE formula_id = ? AND is_active = 1
""", (formula_id,))
formula = cursor.fetchone()
if formula:
return jsonify({'success': True, 'data': dict(formula)})
else:
return jsonify({'success': False, 'error': '처방을 찾을 수 없습니다'}), 404
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/formulas/<int:formula_id>', methods=['PUT'])
def update_formula(formula_id):
"""처방 수정"""
try:
data = request.json
with get_db() as conn:
cursor = conn.cursor()
# 처방 기본 정보 업데이트
cursor.execute("""
UPDATE formulas
SET formula_code = ?, formula_name = ?, formula_type = ?,
base_cheop = ?, base_pouches = ?, description = ?, efficacy = ?,
updated_at = CURRENT_TIMESTAMP
WHERE formula_id = ?
""", (
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('efficacy'),
formula_id
))
# 기존 구성 약재 삭제
cursor.execute("DELETE FROM formula_ingredients WHERE formula_id = ?", (formula_id,))
# 새로운 구성 약재 추가
if 'ingredients' in data:
for idx, ingredient in enumerate(data['ingredients']):
# ingredient_code 기반으로 저장
cursor.execute("""
INSERT INTO formula_ingredients (formula_id, ingredient_code,
grams_per_cheop, notes, sort_order)
VALUES (?, ?, ?, ?, ?)
""", (
formula_id,
ingredient.get('ingredient_code', ingredient.get('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>', methods=['DELETE'])
def delete_formula(formula_id):
"""처방 삭제 (소프트 삭제)"""
try:
with get_db() as conn:
cursor = conn.cursor()
# 조제에서 사용 중인지 확인
cursor.execute("""
SELECT COUNT(*) as count
FROM compounds
WHERE formula_id = ?
""", (formula_id,))
count = cursor.fetchone()['count']
if count > 0:
return jsonify({
'success': False,
'error': '이 처방은 조제 내역에서 사용 중이므로 삭제할 수 없습니다'
}), 400
# 소프트 삭제 (is_active를 0으로 설정)
cursor.execute("""
UPDATE formulas
SET is_active = 0, updated_at = CURRENT_TIMESTAMP
WHERE formula_id = ?
""", (formula_id,))
return jsonify({
'success': True,
'message': '처방이 삭제되었습니다'
})
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():
"""재고 현황 요약 - 원산지별 구분 표시
Query Parameters:
mode (str): 계산 모드
- 'all' (기본): 모든 LOT 포함
- 'receipt_only': 입고장과 연결된 LOT만
- 'verified': 검증된 LOT만 (is_verified=1)
"""
try:
# 계산 모드 파라미터 가져오기
mode = request.args.get('mode', 'all')
# 모드별 WHERE 조건 설정
where_conditions = ["il.is_depleted = 0"]
if mode == 'receipt_only':
# receipt_line_id > 0인 것만 (0은 입고장 없음 표시)
where_conditions.append("il.receipt_line_id > 0")
elif mode == 'verified':
# is_verified 컬럼이 없을 경우를 대비해 조건 추가
# 현재는 receipt_line_id > 0인 것을 검증된 것으로 간주
where_conditions.append("il.receipt_line_id > 0")
# 'all' 모드는 추가 조건 없음
where_clause = " AND ".join(where_conditions) if where_conditions else "1=1"
with get_db() as conn:
cursor = conn.cursor()
# 먼저 재고 정보를 정확하게 계산 (효능 태그 JOIN 없이)
cursor.execute(f"""
SELECT
h.herb_item_id,
h.insurance_code,
h.herb_name,
h.ingredient_code,
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
FROM herb_items h
LEFT JOIN inventory_lots il ON h.herb_item_id = il.herb_item_id AND {where_clause}
GROUP BY h.herb_item_id, h.insurance_code, h.herb_name, h.ingredient_code
HAVING total_quantity > 0
ORDER BY h.herb_name
""")
inventory = []
for row in cursor.fetchall():
item = dict(row)
# 효능 태그를 별도 쿼리로 가져오기
if item['ingredient_code']:
cursor.execute("""
SELECT GROUP_CONCAT(DISTINCT et.tag_name) as tags
FROM herb_item_tags hit
JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
WHERE hit.ingredient_code = ?
""", (item['ingredient_code'],))
tags_row = cursor.fetchone()
if tags_row and tags_row[0]:
item['efficacy_tags'] = tags_row[0].split(',')
else:
item['efficacy_tags'] = []
else:
# ingredient_code가 없는 경우 product_code로 시도
cursor.execute("""
SELECT hp.ingredient_code
FROM herb_products hp
WHERE hp.product_code = ?
""", (item['insurance_code'],))
prod_row = cursor.fetchone()
if prod_row and prod_row[0]:
cursor.execute("""
SELECT GROUP_CONCAT(DISTINCT et.tag_name) as tags
FROM herb_item_tags hit
JOIN herb_efficacy_tags et ON hit.tag_id = et.tag_id
WHERE hit.ingredient_code = ?
""", (prod_row[0],))
tags_row = cursor.fetchone()
if tags_row and tags_row[0]:
item['efficacy_tags'] = tags_row[0].split(',')
else:
item['efficacy_tags'] = []
else:
item['efficacy_tags'] = []
# ingredient_code 제거 (API 응답에 불필요)
if 'ingredient_code' in item:
del item['ingredient_code']
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]
# 계산 모드별 추가 정보 조회
mode_info = {
'mode': mode,
'mode_label': {
'all': '전체 재고',
'receipt_only': '입고장 기준',
'verified': '검증된 재고'
}.get(mode, mode)
}
# 입고장 없는 LOT 수 확인 (mode='all'일 때만)
# receipt_line_id = 0을 입고장 없음으로 처리
if mode == 'all':
cursor.execute("""
SELECT COUNT(*) as count,
COALESCE(SUM(quantity_onhand * unit_price_per_g), 0) as value
FROM inventory_lots
WHERE receipt_line_id = 0
AND is_depleted = 0
AND quantity_onhand > 0
""")
no_receipt = cursor.fetchone()
mode_info['no_receipt_lots'] = no_receipt[0]
mode_info['no_receipt_value'] = no_receipt[1]
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, # 보유율
'calculation_mode': mode_info # 계산 모드 정보 추가
}
})
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
# 효능 태그 조회 - ingredient_code 기반
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
JOIN herb_master_extended hme2 ON hit.ingredient_code = hme2.ingredient_code
WHERE hme2.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_ids 또는 tags 파라미터 받기
tag_ids = request.args.get('tag_ids', '')
tag_names = request.args.getlist('tags')
if not tag_ids and not tag_names:
return jsonify({'error': '검색할 태그를 지정해주세요'}), 400
with get_db() as conn:
cursor = conn.cursor()
if tag_ids:
# tag_ids로 검색
tag_id_list = [int(tid) for tid in tag_ids.split(',') if tid]
placeholders = ','.join('?' * len(tag_id_list))
cursor.execute(f"""
SELECT DISTINCT
hme.herb_id,
hme.ingredient_code,
COALESCE(hm.herb_name, hme.name_korean) as herb_name,
hme.name_hanja,
hme.main_effects,
hme.property,
hme.taste,
GROUP_CONCAT(het.tag_name) as tags
FROM herb_master_extended hme
LEFT JOIN herb_masters hm ON hme.ingredient_code = hm.ingredient_code
JOIN herb_item_tags hit ON hme.ingredient_code = hit.ingredient_code
JOIN herb_efficacy_tags het ON hit.tag_id = het.tag_id
WHERE hit.tag_id IN ({placeholders})
GROUP BY hme.herb_id
ORDER BY hm.herb_name
""", tag_id_list)
else:
# tag_names로 검색
placeholders = ','.join('?' * len(tag_names))
cursor.execute(f"""
SELECT DISTINCT
hme.herb_id,
hme.ingredient_code,
COALESCE(hm.herb_name, hme.name_korean) as herb_name,
hme.name_hanja,
hme.main_effects,
hme.property,
hme.taste,
GROUP_CONCAT(het.tag_name) as tags
FROM herb_master_extended hme
LEFT JOIN herb_masters hm ON hme.ingredient_code = hm.ingredient_code
JOIN herb_item_tags hit ON hme.ingredient_code = hit.ingredient_code
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 hm.herb_name
""", tag_names)
results = []
for row in cursor.fetchall():
results.append({
'herb_id': row['herb_id'],
'ingredient_code': row['ingredient_code'],
'herb_name': row['herb_name'],
'name_hanja': row['name_hanja'],
'main_effects': row['main_effects'],
'property': row['property'],
'taste': row['taste'],
'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
# ==================== 판매 관리 API ====================
@app.route('/api/compounds/<int:compound_id>/status', methods=['POST'])
def update_compound_status(compound_id):
"""조제 상태 업데이트"""
try:
data = request.json
new_status = data.get('status')
valid_statuses = ['PREPARED', 'PENDING_PAYMENT', 'PAID', 'PENDING_DELIVERY',
'DELIVERED', 'COMPLETED', 'OTC_CONVERTED', 'CANCELLED', 'REFUNDED']
if new_status not in valid_statuses:
return jsonify({'error': '유효하지 않은 상태입니다'}), 400
with get_db() as conn:
cursor = conn.cursor()
# 현재 상태 조회
cursor.execute("SELECT status FROM compounds WHERE compound_id = ?", (compound_id,))
current = cursor.fetchone()
if not current:
return jsonify({'error': '조제 정보를 찾을 수 없습니다'}), 404
old_status = current['status']
# 상태 업데이트
update_fields = ['status = ?']
update_values = [new_status]
# 결제 관련 정보
if new_status == 'PAID':
if 'payment_method' in data:
update_fields.append('payment_method = ?')
update_values.append(data['payment_method'])
if 'payment_date' in data:
update_fields.append('payment_date = ?')
update_values.append(data['payment_date'])
if 'actual_payment_amount' in data:
update_fields.append('actual_payment_amount = ?')
update_values.append(data['actual_payment_amount'])
# 배송 관련 정보
elif new_status in ['PENDING_DELIVERY', 'DELIVERED']:
if 'delivery_method' in data:
update_fields.append('delivery_method = ?')
update_values.append(data['delivery_method'])
if 'delivery_date' in data:
update_fields.append('delivery_date = ?')
update_values.append(data['delivery_date'])
if 'invoice_number' in data:
update_fields.append('invoice_number = ?')
update_values.append(data['invoice_number'])
update_values.append(compound_id)
cursor.execute(f"""
UPDATE compounds
SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP
WHERE compound_id = ?
""", update_values)
# 상태 변경 이력 저장
cursor.execute("""
INSERT INTO sales_status_history (compound_id, old_status, new_status, changed_by, change_reason)
VALUES (?, ?, ?, ?, ?)
""", (compound_id, old_status, new_status,
data.get('changed_by', 'system'), data.get('reason', '')))
# 판매 거래 기록 (결제 완료시)
if new_status == 'PAID' and 'actual_payment_amount' in data:
cursor.execute("""
INSERT INTO sales_transactions (compound_id, transaction_date, transaction_type,
amount, payment_method, payment_status, created_by)
VALUES (?, ?, 'SALE', ?, ?, 'COMPLETED', ?)
""", (compound_id, data.get('payment_date', datetime.now()),
data['actual_payment_amount'], data.get('payment_method'),
data.get('changed_by', 'system')))
conn.commit()
return jsonify({'success': True, 'message': '상태가 업데이트되었습니다'})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/compounds/<int:compound_id>/price', methods=['PUT'])
def update_compound_price(compound_id):
"""조제 가격 조정 (복합결제 지원)"""
try:
data = request.json
with get_db() as conn:
cursor = conn.cursor()
# 현재 정보 조회
cursor.execute("""
SELECT sell_price_total, cost_total, patient_id
FROM compounds WHERE compound_id = ?
""", (compound_id,))
current = cursor.fetchone()
if not current:
return jsonify({'error': '조제 정보를 찾을 수 없습니다'}), 404
update_fields = []
update_values = []
# 판매가격
if 'sell_price_total' in data:
update_fields.append('sell_price_total = ?')
update_values.append(data['sell_price_total'])
# 할인 정보
if 'discount_rate' in data:
update_fields.append('discount_rate = ?')
update_values.append(data['discount_rate'])
if 'discount_amount' in data:
update_fields.append('discount_amount = ?')
update_values.append(data['discount_amount'])
if 'discount_reason' in data:
update_fields.append('discount_reason = ?')
update_values.append(data['discount_reason'])
# 복합 결제 정보
if 'payment_cash' in data:
update_fields.append('payment_cash = ?')
update_values.append(data['payment_cash'])
if 'payment_card' in data:
update_fields.append('payment_card = ?')
update_values.append(data['payment_card'])
if 'payment_transfer' in data:
update_fields.append('payment_transfer = ?')
update_values.append(data['payment_transfer'])
# 마일리지 사용
if 'mileage_used' in data and data['mileage_used'] > 0:
update_fields.append('mileage_used = ?')
update_values.append(data['mileage_used'])
# 환자 마일리지 차감
if current['patient_id']:
cursor.execute("""
UPDATE patients
SET mileage_balance = mileage_balance - ?,
total_mileage_used = total_mileage_used + ?
WHERE patient_id = ?
""", (data['mileage_used'], data['mileage_used'], current['patient_id']))
# 마일리지 거래 이력 추가
cursor.execute("""
INSERT INTO mileage_transactions
(patient_id, compound_id, transaction_type, amount, balance_after, description, created_by)
SELECT patient_id, ?, 'USED', ?, mileage_balance, '한약 결제 사용', 'system'
FROM patients WHERE patient_id = ?
""", (compound_id, data['mileage_used'], current['patient_id']))
# 실제 결제 금액 계산
total_payment = (data.get('mileage_used', 0) +
data.get('payment_cash', 0) +
data.get('payment_card', 0) +
data.get('payment_transfer', 0))
if total_payment > 0:
update_fields.append('actual_payment_amount = ?')
update_values.append(total_payment)
update_values.append(compound_id)
if update_fields:
cursor.execute(f"""
UPDATE compounds
SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP
WHERE compound_id = ?
""", update_values)
conn.commit()
return jsonify({'success': True, 'message': '가격이 조정되었습니다'})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/sales/statistics', methods=['GET'])
def get_sales_statistics():
"""판매 통계 조회"""
try:
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
with get_db() as conn:
cursor = conn.cursor()
# 기간 내 매출 통계
query = """
SELECT
COUNT(*) as total_count,
SUM(COALESCE(actual_payment_amount, sell_price_total)) as total_sales,
AVG(COALESCE(actual_payment_amount, sell_price_total)) as avg_price,
COUNT(CASE WHEN status = 'COMPLETED' THEN 1 END) as completed_count,
COUNT(CASE WHEN status = 'CANCELLED' THEN 1 END) as cancelled_count,
COUNT(CASE WHEN status = 'REFUNDED' THEN 1 END) as refunded_count
FROM compounds
WHERE status IN ('PAID', 'PENDING_DELIVERY', 'DELIVERED', 'COMPLETED')
"""
params = []
if start_date:
query += " AND compound_date >= ?"
params.append(start_date)
if end_date:
query += " AND compound_date <= ?"
params.append(end_date)
cursor.execute(query, params)
stats = cursor.fetchone()
# 일별 매출 추이
daily_query = """
SELECT
DATE(compound_date) as sale_date,
COUNT(*) as count,
SUM(COALESCE(actual_payment_amount, sell_price_total)) as daily_total
FROM compounds
WHERE status IN ('PAID', 'PENDING_DELIVERY', 'DELIVERED', 'COMPLETED')
"""
if start_date:
daily_query += " AND compound_date >= ?"
if end_date:
daily_query += " AND compound_date <= ?"
daily_query += " GROUP BY DATE(compound_date) ORDER BY sale_date"
cursor.execute(daily_query, params)
daily_sales = []
for row in cursor.fetchall():
daily_sales.append({
'date': row['sale_date'],
'count': row['count'],
'total': row['daily_total']
})
# 처방별 매출
formula_query = """
SELECT
f.formula_name,
COUNT(*) as count,
SUM(COALESCE(c.actual_payment_amount, c.sell_price_total)) as total
FROM compounds c
LEFT JOIN formulas f ON c.formula_id = f.formula_id
WHERE c.status IN ('PAID', 'PENDING_DELIVERY', 'DELIVERED', 'COMPLETED')
"""
if start_date:
formula_query += " AND c.compound_date >= ?"
if end_date:
formula_query += " AND c.compound_date <= ?"
formula_query += " GROUP BY f.formula_name ORDER BY total DESC LIMIT 10"
cursor.execute(formula_query, params)
formula_sales = []
for row in cursor.fetchall():
formula_sales.append({
'formula_name': row['formula_name'] or '직접조제',
'count': row['count'],
'total': row['total']
})
return jsonify({
'summary': {
'total_count': stats['total_count'],
'total_sales': stats['total_sales'],
'avg_price': stats['avg_price'],
'completed_count': stats['completed_count'],
'cancelled_count': stats['cancelled_count'],
'refunded_count': stats['refunded_count']
},
'daily_sales': daily_sales,
'formula_sales': formula_sales
})
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/compounds/<int:compound_id>/convert-to-otc', methods=['POST'])
def convert_to_otc(compound_id):
"""OTC 전환"""
try:
data = request.json
with get_db() as conn:
cursor = conn.cursor()
# 조제 정보 업데이트
cursor.execute("""
UPDATE compounds
SET status = 'OTC_CONVERTED',
notes = COALESCE(notes || ' | ', '') || 'OTC 전환: ' || ?,
updated_at = CURRENT_TIMESTAMP
WHERE compound_id = ?
""", (data.get('reason', ''), compound_id))
# 상태 변경 이력
cursor.execute("""
INSERT INTO sales_status_history (compound_id, old_status, new_status,
changed_by, change_reason)
SELECT compound_id, status, 'OTC_CONVERTED', ?, ?
FROM compounds WHERE compound_id = ?
""", (data.get('changed_by', 'system'), data.get('reason', ''), compound_id))
conn.commit()
return jsonify({'success': True, 'message': 'OTC로 전환되었습니다'})
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)
Reference in New Issue View Git Blame Copy Permalink