240915 연구노트

import mysql.connector from record_play import record_audio, play_audio from CSR import audio_to_text from Chatbot_240913V import execute_chatbot, emergency_execute_chatbot from TTS_240914V import text_to_speech import re def connect_to_db(): try: connection = mysql.connector.connect( host='localhost', user='hanshin', password='hanshin2024', database='user_data' ) return connection except mysql.connector.Error as err: print(f"Error : {err}") return None def get_user_data(user_id): connection = connect_to_db() if connection is None: return {'error' : 'User_DB connection 실패'} try: cur = connection.cursor(dictionary =True) # query 설정 query = """ SELECT name, gender, date_of_birth, contact_number, address, disability_type, disability_grade, emergency_contact, special_notes FROM users WHERE user_id = %s """ # query 실행 cur.execute(query, (user_id,)) # user_data 내용 변수에 저장 user_data = cur.fetchone() # 디버깅 출력을 추가 print("get_user_data 반환 값:", user_data) if user_data: return user_data else: return {'error' : 'User not found'} except mysql.connector.Error as err: return {'error' : f"User Data 조회 중 에러 : {err}"} finally: if connection.is_connected(): connection.close() def get_situation_data(event_id): connection = connect_to_db() if connection is None: return {'error' : 'Situation_DB connection 실패'} try : cur = connection.cursor(dictionary = True) query = """ SELECT event_id, user_id, event_type, event_description, is_emergency, harm_severity, event_timestamp FROM situations WHERE event_id = %s """ cur.execute(query, (event_id,)) situation_data = cur.fetchone() if situation_data: return situation_data else: return {'error' : 'Event not found'} except mysql.connector.Error as err: return {'error' : f"Situation Data 조회 중 에러 : {err}"} finally: if connection.is_connected(): connection.close() def manage_dialogue_session(user_id, event_id): user_data = get_user_data(user_id) situation_data = get_situation_data(event_id) if not user_data: return {'error' : 'User not found'} if not situation_data: return {'error' : 'Event not found'} # situation_data 딕셔너리에서 is_emergency 값을 가져오기 # 만약 is_emergency 값이 없으면 default값으로 False를 사용함 is_emergency = situation_data.get('is_emergency',False) print(f"is_emergency : {is_emergency}") if is_emergency: response = generate_emergency_chat(user_data, situation_data) else: response = generate_usually_chat(user_data, situation_data) # 상황이 위급한 상황인 경우 : situation 정보를 바탕으로 바로 chatbot을 시작해서 상황을 해결하기 위한 대화를 생성 def generate_emergency_chat(user_data, situation_data): try: chatbot_result = emergency_execute_chatbot(user_data=user_data, situation_data=situation_data) print(f"chatbot_result : {chatbot_result}") cleaned_result = clean_chatbot_result(chatbot_result) print(f"cleand_result : {cleaned_result}") if chatbot_result: text_to_speech(text= cleaned_result, user_name=user_data['name']) else: print("chatbot 실행 실패") return None except Exception as e: print(f"Error : {e}") return None # 상황이 위급하지 않은 상황인 경우 : recording부터 시작해서 사용자의 대화를 받아와서 Chatbot 시작 def generate_usually_chat(user_data, situation_data): try: # record_audio() # play_audio() # Clova Speech API로 오디오 파일을 텍스트로 변환 text_result = audio_to_text() # debug 목적 print(f"""user_data : {user_data} \n situation_data : {situation_data} \n text : {text_result}""") if text_result: chatbot_result = execute_chatbot(user_data=user_data, situation_data=situation_data, text=text_result) print(f"chatbot_result : {chatbot_result}") else: print("오디오 텍스트 변환 실패") return None if chatbot_result: text_to_speech(text = chatbot_result, user_name= user_data['name']) else: print("chatbot 실행 실패") return None except Exception as e: print(f"Error : {e}") return None def clean_chatbot_result(result): # 정규 표현식을 사용하여 대괄호와 그 안의 내용 제거 cleaned_result = re.sub(r'\[.*?\]', '', result) return cleaned_result.strip() # 필요 시 앞뒤 공백도 제거 if __name__ == '__main__': user_id = 1 event_id = 4 # is_emergency = False manage_dialogue_session(user_id, event_id) pass