""" 간호 인수인계 교육 플랫폼 - Gradio 멀티페이지 애플리케이션 (채팅 기반) """ import os import gradio as gr from services.gemini_service import GeminiEvaluator from services.supabase_service import SupabaseService from config.database import supabase_client # 데이터베이스 초기화 try: print("🔗 데이터베이스 연결 시도 중...") # Supabase 서비스 초기화 if supabase_client: try: supabase_service = SupabaseService() print("✅ Supabase 클라이언트 초기화 완료") # 시나리오 데이터 로드 (테이블이 없는 경우 무시) try: print("📥 시나리오 데이터 로드 중...") supabase_service.load_scenarios_from_json("data/scenarios.json") print("✅ 시나리오 데이터 로드 완료") except Exception as load_error: print(f"⚠️ 시나리오 데이터 로드 실패: {load_error}") print(" (이미 데이터가 있거나 테이블이 없을 수 있습니다)") except Exception as service_error: print(f"⚠️ Supabase 서비스 초기화 실패: {service_error}") supabase_service = None else: print("⚠️ Supabase 클라이언트가 초기화되지 않았습니다.") print(" 환경 변수 SUPABASE_URL과 SUPABASE_ANON_KEY를 확인하세요.") supabase_service = None except Exception as e: print(f"⚠️ 데이터베이스 초기화 중 오류: {e}") import traceback traceback.print_exc() supabase_service = None # Gemini 평가기 초기화 evaluator = GeminiEvaluator() # 초기 학생 ID 생성 (TSID 기반) from tsidpy import TSID tsid = TSID.create() initial_student_id = f"student_{tsid.to_string()}" # ===== EMR HTML 생성 함수들 (기존 로직 재사용) ===== def create_emr_html(patient, scenario): """실제 병원 EMR과 유사한 HTML 생성""" vitals = scenario.vitals or {} labs = scenario.labs or {} orders = scenario.orders or [] nursing_notes = scenario.nursing_notes or [] # EMR 헤더 emr_html = f"""
환자명 / 등록번호
{patient.name} / {patient.id}
나이/성별 / 병실
{patient.age}세 / {patient.gender} / {patient.room_number}
입원일 / 재원일수
{patient.admission_date} / D+{scenario.hospital_day}
진단명
{patient.diagnosis}
⚠️ 알레르기: {patient.allergies}   |   📋 과거력: {patient.comorbidities}   |   👨‍⚕️ 담당의: {patient.attending_physician}
📊 활력징후 ({vitals.get('time', 'N/A')})
{create_vital_card('혈압', 'BP', vitals.get('BP', 'N/A'), '100-140/60-90', vitals.get('BP', ''))} {create_vital_card('맥박', 'HR', vitals.get('HR', 'N/A'), '60-100', vitals.get('HR', ''))} {create_vital_card('호흡', 'RR', vitals.get('RR', 'N/A'), '12-20', vitals.get('RR', ''))} {create_vital_card('체온', 'BT', vitals.get('BT', 'N/A'), '36.0-37.5', vitals.get('BT', ''))} {create_vital_card('산소포화도', 'SpO2', vitals.get('SpO2', 'N/A'), '≥95%', vitals.get('SpO2', ''))} {create_vital_card('통증', 'Pain', vitals.get('Pain', 'N/A'), '0-3', vitals.get('Pain', ''))}
🧪 검사 결과 ({labs.get('time', 'N/A')})
{create_lab_results_table(labs)}
💊 의사 처방
{create_orders_list(orders)}
📝 간호 기록
{create_nursing_notes_timeline(nursing_notes)}
{create_surgery_info_panel(scenario) if scenario.surgery_info else ''} {create_discharge_education_panel(scenario) if scenario.discharge_education else ''}
""" return emr_html def create_vital_card(label, code, value, normal_range, raw_value): """활력징후 카드 생성""" is_abnormal = check_vital_abnormal(code, raw_value) bg_color = "#ffe6e6" if is_abnormal else "#e8f5e9" border_color = "#f44336" if is_abnormal else "#4caf50" icon = "⚠️" if is_abnormal else "✓" return f"""
{label}
{icon} {value}
정상: {normal_range}
""" def check_vital_abnormal(code, value_str): """활력징후 이상 여부 체크""" if not value_str or value_str == "N/A": return False import re numbers = re.findall(r"\d+\.?\d*", value_str) if not numbers: return False try: if code == "BP": bp_values = re.findall(r"(\d+)/(\d+)", value_str) if bp_values: systolic = int(bp_values[0][0]) diastolic = int(bp_values[0][1]) return systolic < 90 or systolic > 140 or diastolic < 60 or diastolic > 90 elif code == "HR": hr = int(numbers[0]) return hr < 60 or hr > 100 elif code == "RR": rr = int(numbers[0]) return rr < 12 or rr > 20 elif code == "BT": bt = float(numbers[0]) return bt < 36.0 or bt > 37.5 elif code == "SpO2": spo2 = int(numbers[0]) return spo2 < 95 elif code == "Pain": pain = int(numbers[0]) return pain > 3 except (ValueError, IndexError): pass return False def create_lab_results_table(labs): """검사 결과 테이블 생성""" if not labs: return "
검사 결과 없음
" html = '' html += '' html += '' html += '' html += '' html += '' html += "" # CBC 결과 if "CBC" in labs: cbc = labs["CBC"] html += create_lab_row("WBC", cbc.get("WBC", "N/A"), "4.0-10.0 ×10³/µL", check_lab_abnormal("WBC", cbc.get("WBC", ""))) html += create_lab_row("Hemoglobin", cbc.get("Hb", "N/A"), "13.0-17.0 g/dL (M)", check_lab_abnormal("Hb", cbc.get("Hb", ""))) html += create_lab_row("Platelet", cbc.get("Plt", "N/A"), "150-400 ×10³/µL", False) html += create_lab_row("Neutrophil", cbc.get("Neutrophil", "N/A"), "40-70%", False) # CRP if "CRP" in labs: html += create_lab_row("CRP", labs["CRP"], "<0.5 mg/dL", check_lab_abnormal("CRP", labs["CRP"])) # 전해질 if "BMP" in labs: bmp = labs["BMP"] html += create_lab_row("Sodium", f"{bmp.get('Na', 'N/A')} mmol/L", "136-145 mmol/L", False) html += create_lab_row("Potassium", f"{bmp.get('K', 'N/A')} mmol/L", "3.5-5.0 mmol/L", False) html += create_lab_row("Creatinine", f"{bmp.get('Cr', 'N/A')} mg/dL", "0.7-1.3 mg/dL", False) html += "
검사항목결과참고범위상태
" return html def create_lab_row(name, value, ref_range, is_abnormal): """검사 결과 행 생성""" status_icon = "🔴" if is_abnormal else "🟢" status_text = "비정상" if is_abnormal else "정상" row_bg = "#fff3f3" if is_abnormal else "white" return f""" {name} {value} {ref_range} {status_icon} {status_text} """ def check_lab_abnormal(lab_name, value_str): """검사 결과 이상 여부 체크""" if "↑" in value_str or "↓" in value_str or "상승" in value_str or "감소" in value_str: return True return False def create_orders_list(orders): """의사 처방 목록 생성""" if not orders: return "
처방 없음
" html = '
' for i, order in enumerate(orders, 1): if any(keyword in order for keyword in ["IV", "injection", "정주", "주사"]): icon = "💉" elif any(keyword in order for keyword in ["PO", "oral", "경구"]): icon = "💊" elif "NPO" in order or "금식" in order: icon = "🚫" elif any(keyword in order for keyword in ["수술", "surgery", "operation"]): icon = "🔪" else: icon = "📋" html += f'
{icon} {i}. {order}
' html += "
" return html def create_nursing_notes_timeline(notes): """간호 기록 타임라인 생성""" if not notes: return "
기록 없음
" html = '
' for note in notes: time = note.get("time", "N/A") content = note.get("note", "") html += f"""
🕐 {time}
{content}
""" html += "
" return html def create_surgery_info_panel(scenario): """수술 정보 패널 생성""" if not scenario.surgery_info: return "" surgery = scenario.surgery_info return f"""
🏥 수술 정보
술식: {surgery.get('procedure', 'N/A')}
수술시간: {surgery.get('time', 'N/A')}
출혈량: {surgery.get('EBL', 'N/A')}
수액량: {surgery.get('fluids', 'N/A')}
소견: {surgery.get('findings', 'N/A')}
합병증: {surgery.get('complications', '없음')}
""" def create_discharge_education_panel(scenario): """퇴원 교육 패널 생성""" if not scenario.discharge_education: return "" html = """
📚 퇴원 교육
""" for edu in scenario.discharge_education: html += f'
✓ {edu}
' html += "
" return html # ===== 핵심 로직 함수들 ===== def load_scenario_data(scenario_id: str): """시나리오 선택 시 EMR 데이터 로드""" if not scenario_id: return None, None, "" if not supabase_service: return None, None, "Supabase 서비스가 초기화되지 않았습니다." scenario = supabase_service.get_scenario(scenario_id) patient = supabase_service.get_patient(scenario['patient_id']) if scenario else None if not scenario or not patient: return None, None, "시나리오를 찾을 수 없습니다." # Supabase 데이터를 객체처럼 사용할 수 있도록 래핑 class Scenario: def __init__(self, data): self.id = data['id'] self.patient_id = data['patient_id'] self.title = data['title'] self.hospital_day = data.get('hospital_day') self.vitals = data.get('vitals') self.labs = data.get('labs') self.orders = data.get('orders') self.nursing_notes = data.get('nursing_notes') self.surgery_info = data.get('surgery_info') self.discharge_education = data.get('discharge_education') class Patient: def __init__(self, data): self.id = data['id'] self.name = data['name'] self.age = data['age'] self.gender = data['gender'] self.diagnosis = data['diagnosis'] self.admission_date = data['admission_date'] self.allergies = data['allergies'] self.comorbidities = data.get('comorbidities', '') self.attending_physician = data.get('attending_physician', '') self.room_number = data.get('room_number', '') scenario_obj = Scenario(scenario) patient_obj = Patient(patient) emr_display = create_emr_html(patient_obj, scenario_obj) return scenario_obj, patient_obj, emr_display def handle_chat_message(message, chat_history, session_id, student_id, scenario_id): """채팅 메시지 처리""" if not message or not message.strip(): return chat_history, "" # 세션 ID 확인 print(f"💬 채팅 메시지 처리: session_id={session_id}, student_id={student_id}, scenario_id={scenario_id}") # 시나리오가 선택되지 않은 경우 if not scenario_id or scenario_id == "": warning_msg = """ ⚠️ **시나리오를 먼저 선택해주세요!** 1. **EMR 정보 조회** 탭으로 이동하세요 2. 시나리오 드롭다운에서 시나리오를 선택하세요 3. EMR 정보를 확인한 후 다시 이 페이지로 돌아와 채팅을 시작하세요 현재 선택된 시나리오: 없음 """ return chat_history + [[message, warning_msg]], "" # Supabase 서비스 연결 supabase_service = SupabaseService() # 시나리오 및 환자 정보 조회 scenario = supabase_service.get_scenario(scenario_id) patient = supabase_service.get_patient(scenario['patient_id']) if scenario else None if not scenario or not patient: error_msg = """ ❌ **시나리오 정보를 찾을 수 없습니다.** EMR 정보 조회 탭에서 시나리오를 다시 선택해주세요. """ return chat_history + [[message, error_msg]], "" # 사용자 메시지 저장 (session_id가 있는 경우에만) if session_id: try: supabase_service.save_chat_message(session_id, student_id, scenario_id, "user", message) print(f"✅ 사용자 메시지 저장 완료: session_id={session_id}") except Exception as e: print(f"⚠️ 사용자 메시지 저장 실패: {e}") else: print(f"⚠️ session_id가 비어있어 메시지 저장 실패") # 시나리오 및 환자 데이터 준비 scenario_data = { "title": scenario['title'], "handoff_situation": scenario.get('handoff_situation'), "hospital_day": scenario.get('hospital_day'), "vitals": scenario.get('vitals'), "labs": scenario.get('labs'), "orders": scenario.get('orders'), "nursing_notes": scenario.get('nursing_notes'), } patient_data = { "name": patient['name'], "age": patient['age'], "gender": patient['gender'], "diagnosis": patient['diagnosis'], "admission_date": str(patient['admission_date']), "allergies": patient['allergies'], } # 이전 채팅 기록 포맷팅 (Gemini용) history_for_gemini = [] for user_msg, bot_msg in chat_history: if user_msg: history_for_gemini.append(("user", user_msg)) if bot_msg: history_for_gemini.append(("assistant", bot_msg)) # Gemini AI 피드백 생성 ai_response = evaluator.chat_feedback(message, history_for_gemini, scenario_data, patient_data) # AI 응답 저장 (session_id가 있는 경우에만) if session_id: try: supabase_service.save_chat_message(session_id, student_id, scenario_id, "assistant", ai_response) print(f"✅ AI 응답 저장 완료: session_id={session_id}") except Exception as e: print(f"⚠️ AI 응답 저장 실패: {e}") else: print(f"⚠️ session_id가 비어있어 AI 응답 저장 실패") # 채팅 히스토리 업데이트 updated_history = chat_history + [[message, ai_response]] return updated_history, "" def finalize_chat_session(chat_history, session_id, student_id, scenario_id): """ 채팅 종료 시 전체 대화를 분석하여 피드백 제공 - 점수 없이 질적 피드백만 제공 - 전체 대화 내용을 Gemini에게 전달 """ if not chat_history: return chat_history + [[None, "대화가 없습니다. 먼저 대화를 시작해주세요."]], "" # Supabase 서비스 연결 supabase_service = SupabaseService() # 시나리오 및 환자 정보 조회 scenario = supabase_service.get_scenario(scenario_id) patient = supabase_service.get_patient(scenario['patient_id']) if scenario else None if not scenario or not patient: return chat_history + [[None, "시나리오 정보를 찾을 수 없습니다."]], "" # 시나리오 및 환자 데이터 준비 scenario_data = { "title": scenario['title'], "handoff_situation": scenario.get('handoff_situation'), "hospital_day": scenario.get('hospital_day'), "vitals": scenario.get('vitals'), "labs": scenario.get('labs'), "orders": scenario.get('orders'), "nursing_notes": scenario.get('nursing_notes'), } patient_data = { "name": patient['name'], "age": patient['age'], "gender": patient['gender'], "diagnosis": patient['diagnosis'], "admission_date": str(patient['admission_date']), "allergies": patient['allergies'], } # 전체 대화를 Gemini에게 전달하여 평가 evaluation = evaluator.evaluate_conversation( chat_history=chat_history, scenario_data=scenario_data, patient_data=patient_data ) # 데이터베이스에 저장 (점수 없이) try: record = supabase_service.create_handoff_record( student_id=student_id, scenario_id=scenario_id, sbar_data={}, # 대화 기반 평가이므로 SBAR 데이터는 비움 evaluation=evaluation, session_id=session_id ) print(f"✅ 인수인계 기록 저장 완료 (ID: {record.get('id')})") # 세션 상태를 'completed'로 변경 supabase_service.update_session_status(session_id, 'completed') except Exception as e: print(f"⚠️ 데이터베이스 저장 오류: {e}") # 피드백 메시지 포맷팅 (점수 없이) strengths = evaluation.get('strengths', []) improvements = evaluation.get('improvements', []) detailed_feedback = evaluation.get('detailed_feedback', '') missing_info = evaluation.get('missing_critical_info', []) safety_concerns = evaluation.get('safety_concerns', []) result_message = f""" 🎓 **학습 종합 피드백** {'='*50} ### 💪 잘한 점 {chr(10).join([f"- {s}" for s in strengths]) if strengths else "- 없음"} ### 📈 개선할 점 {chr(10).join([f"- {i}" for i in improvements]) if improvements else "- 없음"} ### 🔍 종합 의견 {detailed_feedback if detailed_feedback else "종합 피드백이 없습니다."} """ if missing_info: result_message += f""" ### ⚠️ 누락된 중요 정보 {chr(10).join([f"- {m}" for m in missing_info])} """ if safety_concerns: result_message += f""" ### 🏥 환자 안전 관련 주의사항 {chr(10).join([f"- {s}" for s in safety_concerns])} """ result_message += f""" {'='*50} 수고하셨습니다! 다른 시나리오로 연습을 계속하시려면 EMR 페이지로 돌아가 새로운 시나리오를 선택하세요. """ return chat_history + [[None, result_message]], "" def get_student_history(student_id: str): """학생 인수인계 이력 조회 (관리자용)""" if not student_id: return "학생 ID를 입력해주세요." supabase_service = SupabaseService() stats = supabase_service.get_student_statistics(student_id) records = supabase_service.get_records_by_student(student_id) if stats["total_submissions"] == 0: return f"학생 ID '{student_id}'의 제출 기록이 없습니다." history = f""" # 학생 인수인계 이력 **학생 ID**: {student_id} ## 통계 - **총 제출 횟수**: {stats['total_submissions']}회 - **평균 점수**: {stats['average_score']:.1f}점 - **최고 점수**: {stats['highest_score']:.1f}점 - **최저 점수**: {stats['lowest_score']:.1f}점 ## 최근 제출 기록 """ for i, record in enumerate(records[:10], 1): submitted_time = record.get('submitted_at', 'N/A') if submitted_time and isinstance(submitted_time, str): # ISO 형식에서 시간 추출 try: from datetime import datetime dt = datetime.fromisoformat(submitted_time.replace('Z', '+00:00')) submitted_time = dt.strftime("%Y-%m-%d %H:%M") except: pass history += f""" ### {i}. {record.get('scenario_id', 'N/A')} - **제출 시간**: {submitted_time} - **점수**: {record.get('total_score', 0):.1f}점 - **강점**: {', '.join(record.get('strengths', [])[:2]) if record.get('strengths') else 'N/A'} --- """ return history # ===== Gradio UI 구성 ===== with gr.Blocks(title="간호 인수인계 교육 플랫폼", theme=gr.themes.Soft()) as app: # 전역 상태 변수 session_id_state = gr.State(value="") student_id_state = gr.State(value="") scenario_id_state = gr.State(value="S001_D0_ER_WARD") # 기본값 설정 user_role_state = gr.State(value="student") gr.Markdown(""" # 🏥 간호 인수인계 교육 플랫폼 (채팅 기반) SBAR 형식으로 환자 인수인계를 연습하고 AI와 대화하며 피드백을 받아보세요. """) # 초기 설정 섹션 with gr.Row(): with gr.Column(): student_id_input = gr.Textbox( label="👤 학생 ID (자동 생성됨)", value=initial_student_id, interactive=False, info="자동으로 생성된 고유 ID입니다. '새 ID 생성' 버튼으로 새 ID를 받을 수 있습니다." ) with gr.Row(): generate_id_btn = gr.Button("🔄 새 ID 생성", variant="secondary") # 탭 구성 with gr.Tabs() as tabs: # 페이지 1: EMR 정보 조회 with gr.Tab("📊 EMR 정보 조회", id=0) as tab_emr: gr.Markdown("## 환자 전자의무기록 (EMR)") # 사용 방법과 시나리오 선택을 한 행에 2열로 배치 with gr.Row(): with gr.Column(scale=2): gr.Markdown(""" **📌 사용 방법:** 1. 시나리오를 선택하세요 2. EMR 정보를 충분히 확인하세요 3. 준비가 되면 "답안 작성하러 가기" 버튼을 클릭하세요 """) with gr.Column(scale=3): scenario_dropdown = gr.Dropdown( choices=[ ("Day 0 - 응급실→병동 인계 (수술 전)", "S001_D0_ER_WARD"), ("Day 0 - 병동→수술실 인계", "S001_D0_WARD_OR"), ("Day 0 - 회복실→병동 전동 (수술 후)", "S001_D0_RECOVERY"), ("Day 1 - 아침 인계 (POD1)", "S001_D1_MORNING"), ("Day 2 - 퇴원 전 인계 (POD2)", "S001_D2_DISCHARGE"), ], label="📋 시나리오 선택", value="S001_D0_ER_WARD", info="시나리오를 선택하면 EMR 정보가 표시됩니다" ) emr_display = gr.HTML( value="
⬆️ 위에서 시나리오를 선택하세요.
" ) go_to_chat_btn = gr.Button("💬 답안 작성하러 가기", variant="primary", size="lg") gr.Markdown(""" > 💡 **Tip**: EMR 페이지에서 언제든지 "답안 작성" 탭으로 이동할 수 있습니다. """) # 페이지 2: 채팅 기반 답안 작성 with gr.Tab("💬 답안 작성 (채팅)", id=1) as tab_chat: gr.Markdown("## AI 교수와 함께하는 인수인계 연습") gr.Markdown(""" **📌 시작하기:** 1. 기본 시나리오는 **"Day 0 - 응급실→병동 인계"**입니다 (상단에서 확인 가능) 2. EMR 정보를 확인하려면 **"EMR 정보 조회"** 탭을 클릭하세요 3. 준비가 되면 AI 교수와 자유롭게 대화하며 인수인계 내용을 작성하세요 """) with gr.Row(): current_scenario_display = gr.Markdown("**현재 시나리오**: 로딩 중...") back_to_emr_btn = gr.Button("📊 EMR 다시 보기", size="sm", variant="secondary") chatbot = gr.Chatbot( label="AI 교수와 대화", height=500, show_label=True ) with gr.Row(): msg_input = gr.Textbox( label="메시지 입력", placeholder="자유롭게 대화하며 인수인계 내용을 작성하거나, AI 교수에게 질문하세요...", lines=3, scale=4 ) with gr.Row(): send_btn = gr.Button("전송", variant="primary") finalize_btn = gr.Button("채팅 종료 및 평가받기", variant="stop") clear_btn = gr.Button("대화 초기화") gr.Markdown(""" **사용 방법:** 1. 자유롭게 대화하며 인수인계 내용을 작성하세요 2. AI 교수가 즉시 피드백을 제공합니다 3. 피드백을 바탕으로 수정하고 더 나은 내용을 작성할 수 있습니다 4. 준비가 되면 **채팅 종료 및 평가받기** 버튼을 눌러 종합 피드백을 받으세요 **💡 팁:** - 자연스러운 대화로 배우는 것이 핵심입니다 - SBAR 형식을 강제하지 않습니다. 대화를 통해 자동으로 평가됩니다 - 언제든지 AI 교수에게 질문하세요 """) # 페이지 3: 관리자 이력 조회 with gr.Tab("📚 학생 이력 조회 (관리자)", id=2, visible=False) as tab_admin: gr.Markdown("## 학생 인수인계 이력 조회") with gr.Row(): history_student_id = gr.Textbox(label="학생 ID", placeholder="student001") history_btn = gr.Button("이력 조회") history_display = gr.Markdown(value="") # ===== 이벤트 핸들러 ===== def generate_student_id(): """TSID 기반 학생 ID 생성""" from tsidpy import TSID tsid = TSID.create() student_id = f"student_{tsid.to_string()}" return student_id def update_student_id(): """새 학생 ID 생성""" new_id = generate_student_id() return new_id # 새 ID 생성 버튼 (student_id_state도 함께 업데이트) def update_student_id_with_state(): """새 학생 ID 생성 및 상태 업데이트""" new_id = generate_student_id() is_admin = (new_id == "admin") role = "admin" if is_admin else "student" tab_visibility = gr.update(visible=is_admin) return new_id, new_id, role, tab_visibility generate_id_btn.click( fn=update_student_id_with_state, outputs=[student_id_input, student_id_state, user_role_state, tab_admin] ) def select_scenario_and_create_session(scenario_id, student_id): """시나리오 선택 및 세션 생성""" scenario, patient, emr_html = load_scenario_data(scenario_id) if scenario: # 새 채팅 세션 생성 try: supabase_service = SupabaseService() new_session_id = supabase_service.create_chat_session(student_id, scenario_id) print(f"📝 시나리오 선택 시 세션 생성: session_id={new_session_id}") scenario_info = f"**현재 시나리오**: {scenario.title}" return emr_html, new_session_id, scenario_id, scenario_info except Exception as e: print(f"⚠️ 세션 생성 실패: {e}") import traceback traceback.print_exc() return emr_html, "", scenario_id, "**현재 시나리오**: 선택되지 않음" return emr_html, "", scenario_id, "**현재 시나리오**: 선택되지 않음" scenario_dropdown.change( fn=select_scenario_and_create_session, inputs=[scenario_dropdown, student_id_state], outputs=[emr_display, session_id_state, scenario_id_state, current_scenario_display] ) # 채팅 메시지 전송 (개선된 버전) send_btn.click( fn=handle_chat_message, inputs=[msg_input, chatbot, session_id_state, student_id_state, scenario_id_state], outputs=[chatbot, msg_input], show_progress="minimal" ) msg_input.submit( fn=handle_chat_message, inputs=[msg_input, chatbot, session_id_state, student_id_state, scenario_id_state], outputs=[chatbot, msg_input], show_progress="minimal" ) # 채팅 종료 및 평가 finalize_btn.click( fn=finalize_chat_session, inputs=[chatbot, session_id_state, student_id_state, scenario_id_state], outputs=[chatbot, msg_input] ) # 대화 초기화 def clear_chat(student_id, scenario_id): """대화 초기화 및 새 세션 생성""" try: supabase_service = SupabaseService() new_session_id = supabase_service.create_chat_session(student_id, scenario_id) print(f"🔄 대화 초기화 - 새 세션 생성: {new_session_id}") return [], new_session_id except Exception as e: print(f"⚠️ 대화 초기화 중 세션 생성 실패: {e}") return [], "" clear_btn.click( fn=clear_chat, inputs=[student_id_state, scenario_id_state], outputs=[chatbot, session_id_state] ) # 이력 조회 history_btn.click( fn=get_student_history, inputs=[history_student_id], outputs=[history_display] ) # 버튼 핸들러 추가 def navigate_to_chat(session_id, scenario_id): """채팅 페이지로 이동""" return gr.update(selected=1) # 채팅 탭으로 전환 def navigate_to_emr(): """EMR 페이지로 이동""" return gr.update(selected=0) # EMR 탭으로 전환 # 답안 작성하기 버튼 클릭 시 채팅 페이지로 이동 go_to_chat_btn.click( fn=navigate_to_chat, inputs=[session_id_state, scenario_id_state], outputs=[tabs] ) # EMR 다시 보기 버튼 클릭 시 EMR 페이지로 이동 back_to_emr_btn.click( fn=navigate_to_emr, outputs=[tabs] ) # 초기 로드 함수 def initialize_app(): """앱 초기 로드 시 실행""" default_scenario = "S001_D0_ER_WARD" # student_id_state를 초기 ID로 설정 (모듈 레벨 변수 사용) current_student_id = initial_student_id initial_session_id = "" # Supabase 서비스가 있으면 시나리오 데이터 로드 if supabase_service: scenario, patient, emr_html = load_scenario_data(default_scenario) if scenario: # 기본 시나리오 정보 표시 scenario_info = f"**현재 시나리오**: {scenario.title}" # 초기 세션 생성 try: initial_session_id = supabase_service.create_chat_session(current_student_id, default_scenario) print(f"✅ 초기 세션 생성 완료: {initial_session_id}") except Exception as e: print(f"⚠️ 초기 세션 생성 실패: {e}") import traceback traceback.print_exc() return emr_html, default_scenario, scenario_info, current_student_id, "student", initial_session_id else: return emr_html, default_scenario, "**현재 시나리오**: 선택되지 않음", current_student_id, "student", "" else: return "", default_scenario, "**현재 시나리오**: 선택되지 않음", current_student_id, "student", "" # 초기 로드 app.load( fn=initialize_app, outputs=[emr_display, scenario_id_state, current_scenario_display, student_id_state, user_role_state, session_id_state] ) if __name__ == "__main__": # Hugging Face Space 환경 확인 is_hf_space = bool(os.getenv("SPACE_ID") or os.getenv("HF_SPACE_ID")) if is_hf_space: # Hugging Face Space에서 실행 print("🚀 Hugging Face Space에서 실행 중...") # Queue 설정으로 API 호출 안정성 향상 app.queue() # Space는 자동으로 호스팅하므로 최소 설정 app.launch(server_name="0.0.0.0", server_port=7860) else: # 로컬/직접 실행 print("🏠 로컬 환경에서 실행 중...") app.launch(server_name="0.0.0.0", server_port=7860, share=False)