240905 회의내용

# 필요한 라이브러리 import import speech_recognition as sr from playsound import playsound import os # Recognizer 객체 생성 r = sr.Recognizer() # 마이크를 오디오 소스로 사용 mic = sr.Microphone() # 저장할 오디오 파일의 이름 folder_name = "./data" file_name = os.path.join(folder_name, "recorded_audio.wav") if not os.path.exists(folder_name): os.makedirs(folder_name) # 음성 녹음 함수 def record_audio(): with mic as source: print("녹음 시작...") # audio_data = r.listen(source, phrase_time_limit=5) audio_data = r.listen(source) print("녹음 완료!") with open(file_name, "wb") as f: f.write(audio_data.get_wav_data()) print("녹음 파일이 저장되었습니다.") # 오디오 파일 재생 함수 def play_audio(): playsound(file_name) """ if __name__ == "__main__": # 음성 녹음 audio_data = record_audio() # 오디오 파일로 저장 with open(file_name, "wb") as f: f.write(audio_data.get_wav_data()) # 오디오 파일 재생 play_audio() """

import requests import json

class ClovaSpeechClient: # Clova Speech invoke URL invoke_url = 'https://clovaspeech-gw.ncloud.com/external/v1/8983/54d5b161d3da62f82f3592398295d44524110cfef1b81e5ca3dae43e79f634d0' secret = 'b256e2de4b3a4898b3f020894236934f' # 외부 URL을 text로 전환 def req_url(self,url, completion, callback=None, userdata =None, forbiddens=None,boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None ): request_body = { 'url': url, 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'Content-Type': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } return requests.post(headers=headers, url=self.invoke_url + '/recognizer/url', data=json.dumps(request_body).encode('UTF-8')) # 클라두으 스토리지에 저장되어 있는 파일을 text로 전환 def req_object_storage(self, data_key, completion, callback=None, userdata=None, forbiddens=None, boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None): request_body = { 'dataKey': data_key, 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'Content-Type': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } return requests.post(headers=headers, url=self.invoke_url + '/recognizer/object-storage', data=json.dumps(request_body).encode('UTF-8')) # 로컬 저장소에 있는 파일을 text로 전환 def req_upload(self, file, completion, callback=None, userdata=None, forbiddens=None, boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None): request_body = { 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } print(json.dumps(request_body, ensure_ascii=False).encode('UTF-8')) files = { 'media': open(file, 'rb'), 'params': (None, json.dumps(request_body, ensure_ascii=False).encode('UTF-8'), 'application/json') } response = requests.post(headers=headers, url=self.invoke_url + '/recognizer/upload', files=files) return response

if __name__ == '__main__': # res = ClovaSpeechClient().req_url(url='http://example.com/media.mp3', completion='sync') # res = ClovaSpeechClient().req_object_storage(data_key='data/media.mp3', completion='sync') """ file에 인식하고자 하는 파일의 경로를 입력하기 """ res = ClovaSpeechClient().req_upload(file='./data/recorded_audio.wav, completion='sync') print(res.text)

import json import requests import pandas as pd import mysql.connector import time

class CompletionExecutor: def __init__(self, host, api_key, api_key_primary_val, request_id): self._host = host self._api_key = api_key self._api_key_primary_val = api_key_primary_val self._request_id = request_id # completion_request : {"role" : "user","content" : f"{question}"} # POST 요청을 보내고, 실시간으로 응답을 처리하여 대화 내용을 생성함 def execute(self, completion_request): headers = { 'X-NCP-CLOVASTUDIO-API-KEY': self._api_key, 'X-NCP-APIGW-API-KEY': self._api_key_primary_val, 'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id, 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'text/event-stream' } content = "" with requests.post(self._host + '/testapp/v1/tasks/4ce94y7k/chat-completions', headers=headers, json=completion_request, stream=True) as r: for line in r.iter_lines(): if line: decoded_line = line.decode("utf-8") if decoded_line.startswith("data:"): json_data = decoded_line[5:] # 'data:' 이후의 JSON 데이터 추출 try: parsed_data = json.loads(json_data) if 'message' in parsed_data and 'content' in parsed_data['message']: if content == parsed_data['message']['content']: pass else: content += parsed_data['message']['content'] except json.JSONDecodeError: # JSON 형식이 맞지 않을 경우 무시 continue return content

def get_user_info(user_id): connection = mysql.connector.connect( host = 'localhost', user = 'hanshin', password = 'hanshin2024', database = 'user_data') cur = connection.cursor() cur.execute(""" SELECT name, gender, date_of_birth, contact_number, address, photo, disability_types, disability_grade, special_notes FROM users WHERE user_id = %s """, (user_id,)) user_info = cur.fetchone() connection.close() if user_info: return { 'name': user_info[0], 'gender': user_info[1], 'date_of_birth': user_info[2], 'contact_number': user_info[3], 'address': user_info[4], 'photo': user_info[5], # BLOB 타입이므로 적절한 처리가 필요할 수 있음 'disability_type': user_info[6], 'disability_grade': user_info[7], 'special_notes': user_info[8] } else: return None

if __name__ == '__main__': completion_executor = CompletionExecutor( host='https://clovastudio.stream.ntruss.com', api_key='NTA0MjU2MWZlZTcxNDJiY5OgY4I/+eAv267dQbt+9aDUE0J6DJ/sL2uIey9+AkMG', api_key_primary_val='dn7QeKh9ZQNxR782O4jPEUU02ae6esNz3DhLUsF2', request_id='fc9998a2-5db4-4661-bbfe-887a78a6be7a' ) path = r"C:\Users\sook3\OneDrive\바탕 화면\something\학습\DEL 과제\국립재활연구원_2408\chatbot 만들기\data\datasets.csv" df = pd.read_csv(path) questions_lst = df['Text'] # preset_text = [{"role":"system","content":"당신은 인지에 장애가 있는 장애인들이 일상생활에서 겪는 어려움을 해결해주는 챗봇입니다\n 당신은 인지에 장애가 있는 장애인들이 일상생활에서 겪는 어려움을 해결해주는 챗봇입니다. \n ### 답변 생성 방법 ### \n - 대화체로 생성한다.\n - 답변은 인지 장애인들이 겪는 어려움과 관련 있게 생성한다.\n - 욕설, 차별적 요소, 부적절한 언어들은 제외한다.\n - 해당 다변은 반드시 리스트나 배열에 사용할 수 있도록 다음과 같은 형식으로 출력한다.\n"}] for idx,question in enumerate(questions_lst): user_info = get_user_info(user_id = 1) if not user_info: print("User info not found for user_id ") continue preset_text = [{"role":"system", "content" : f"""당신은 {user_info['name']}님을 도와주는 챗봇입니다. {user_info['name']}님은 {user_info['disability_type']}을 가지고 있습니다. 다음 대화에서는 이 정보를 바탕으로 대화하세요.\n - 선호도: {user_info['preferences']} \n ### 답변 생성 방법 ### \n - 대화체로 생성한다.\n - 답변은 사용자의 disability_type과 관련된 어려움에 관련 있게 생성한다.\n - 욕설, 차별적 요소, 부적절한 언어들은 제외한다.\n - 해당 답변은 반드시 리스트나 배열에 사용할 수 있도록 출력한다.\n - 답변에 사용자의 이름을 불러주는 역할을 추가한다.\n - 답변에 사용자의 선호도를 반영하여 적절한 답변을 생성한다."""} ] preset_text.append({"role" : "user","content" : f"{question}"}) print(preset_text) request_data = { 'messages': preset_text, 'topP': 0.8, 'topK': 0, 'maxTokens': 256, 'temperature': 0.5, 'repeatPenalty': 5.0, 'stopBefore': [], 'includeAiFilters': True, 'seed': 0 } content = completion_executor.execute(request_data) print(f"답변 생성 결과: {content}") print("----------------------------------") df.loc[idx,'Completion'] = content time.sleep(1) del preset_text[-1] #print(df) df.to_csv(path, index =False, encoding="utf-8-sig")

import record_play import CSR if __name__ == "__main__": # 음성 녹음 audio_data = record_play.record_audio() # 오디오 파일로 저장 with open(record_play.file_name, "wb") as f: f.write(audio_data.get_wav_data()) # 오디오 파일 재생 record_play.play_audio() res = CSR.ClovaSpeechClient().req_upload(file='./data/recorded_audio.wav', completion='sync') print(res.text)