import ast import base64 from bson import ObjectId import re from datetime import datetime from pymongo import ReturnDocument import prj_common as mycommon import secrets import inspect import sys, os import GlobalVariable as MYSY_GV import Ela_Spacy as ela_Spacy import class_mgt as class_mgt from openai import OpenAI #openai.api_key = MYSY_GV.OPENAI_KEY client = OpenAI( # Defaults to os.environ.get("OPENAI_API_KEY") api_key=MYSY_GV.OPENAI_KEY, ) GPT_MODEL = "gpt-4-1106-preview" #"gpt-3.5-turbo-1106" def mysy_openai(diction): try: requestion_response = {} requestion_response['question'] = "Ecris moi une conclusion d'un article qui parle de l'utilisation de l'API ChatGPT" prompt = f"Améliorer : 'Notre objectif est également de développer des solutions de proximité qui permettront à chacun de nos clients d’être unique, de se distinguer. Cette approche personnalisée leur permettra de disposer d’une solution adaptée à leur besoin.'" messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create( model=GPT_MODEL, messages=messages, temperature=0 ) response_message = response.choices[0].message.content #json_formatted_str = json.dumps(response, indent=2) refusal_status = response.choices[0].message.refusal print(" #refusal_status = ", refusal_status) print(" ### AFFICHAGE GPT RESPONSE ") print(response.choices[0]) print(response_message) requestion_response['reponse'] = str(response.choices[0].message.content) """ prompt = f"Ecris moi une conclusion d'un article qui parle de l'utilisation de l'API ChatGPT" completion = client.chat.completions.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) print(completion['choices'][0]['message']['content']) """ return True,requestion_response except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai" """ Cette fonction prends l'_id d'une instruction (collection : voice_instruction) et retourne le texte de confirmation (format text) """ def mysy_openai_instruction_collection_confirmation_texte(diction): try: """ Recuperer les elements de l'instruction depuis la collection : voice_instruction la clé est _id fourni en argument à la fonction """ voice_instruction_count = MYSY_GV.dbname['voice_instruction'].count_documents({'_id':ObjectId(diction['_id']), 'valide':'1', 'locked':'0'}) if( voice_instruction_count != 1 ): mycommon.myprint(str( inspect.stack()[0][3]) + " L'identifiant de l'instruction est invalide ") return False, " L'identifiant de l'instruction est invalide" voice_instruction_data = MYSY_GV.dbname['voice_instruction'].find_one({'_id': ObjectId(diction['_id']), 'valide': '1', 'locked': '0'}) action_a_realiser = "" if( "related_object" in voice_instruction_data.keys() and voice_instruction_data['related_object'] == "myclass"): # L'instruction concerne les formations action_data = MYSY_GV.mysy_voice_instruction_collection_action['myclass'] for val in action_data: for key in val: if( str(key) in voice_instruction_data['instruction'] and action_a_realiser == ""): action_a_realiser = val[key] elif ("related_object" in voice_instruction_data.keys() and voice_instruction_data[ 'related_object'] == "session_formation"): # L'instruction concerne les session de formation action_data = MYSY_GV.mysy_voice_instruction_collection_action['session_formation'] for val in action_data: for key in val: if (str(key) in voice_instruction_data['instruction'] and action_a_realiser == ""): action_a_realiser = val[key] """ Convertion des actions à realiser pour le texte audio """ if( str(action_a_realiser) == "create"): action_a_realiser = " Création " outpufile_name = str(MYSY_GV.SYSTEM_VOICE_MESSAGES_LOCATION)+"mysy_create_class_cherif_voice.mp3" elif( str(action_a_realiser) == "update"): action_a_realiser = " Mise à jour " outpufile_name = str(MYSY_GV.SYSTEM_VOICE_MESSAGES_LOCATION) + "mysy_update_class_cherif_voice.mp3" """ Convertion des actions à realiser pour le texte audio "" " type_objet = "" if (str(voice_instruction_data[ 'related_object']) == "myclass"): type_objet = " Une Formation " elif (str(voice_instruction_data[ 'related_object']) == "session_formation"): action_a_realiser = " Une Session de formation " print(" #### INSTRUCTION SYNTHESE : Object = ", voice_instruction_data[ 'related_object'], " ### ACTION = ", action_a_realiser) if(action_a_realiser and type_objet ): text_a_retourner = " Confirmez : l'action de "+str(action_a_realiser)+". Pour "+str(type_objet) else: text_a_retourner = " Instruction inconnue" print(" ### text_a_retourner = ", text_a_retourner) local_status, outpufile_name = mysy_openai_text_to_voice({'text':str(text_a_retourner)}) """ return True, outpufile_name except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_voice_to_text" """ Cette fonction prend un fichier audio et le converti en fichier texte. cette fonctio prend comme argument : - voice_text - token """ def mysy_openai_voice_to_text(diction): try: file_name = "temp_direct/mysy_test_voice_mp3.mp3" if( "voice_file" in diction.keys() and diction['voice_file'] ): file_name = diction['voice_file'] """ audio_file = open(file_name, "rb") transcription = client.audio.transcriptions.create( model="whisper-1", file=audio_file, #response_format = "text" ) requestion_response = transcription.text """ requestion_response = "OK, Missy, crée la formation avec le titre Je vais à la plage, stop. Description, comment aller à la plage, stop." print(" ### AFFICHAGE GPT RESPONSE - mysy_openai_voice_to_text") print(requestion_response) new_diction = {} new_diction['voice_text'] = str(requestion_response) new_diction['token'] = str(diction['token']) status, retval, collection_instruction_id = mysy_openai_voice_to_text_traitement_1(new_diction) return True, requestion_response, collection_instruction_id except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_voice_to_text", False """ Cette fonction permet de convertir text en voix """ def mysy_openai_text_to_voice(diction): try: """ Verification des input acceptés """ field_list = ['text', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas") return False, " Les informations fournies sont incorrectes", False """ Verification des champs obligatoires """ field_list_obligatoire = ['text', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes", False print(" ### diction = ", diction) now = datetime.now() # getting the timestamp ts = str(datetime.timestamp(now)).replace(".", "").replace(",", "") out_file_name = str(MYSY_GV.TEMPORARY_DIRECTORY_V2)+"mysy_text_to_voice_" + str(ts[-4:])+".mp3" print(" ### Traitement du fichier audio : ", out_file_name) response = client.audio.speech.create( model="tts-1-hd", voice="onyx", input=str(diction["text"]), ) response.stream_to_file(out_file_name) return True, str(out_file_name) except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_text_to_voice" """ Cette fonction permet d'envoyer un fichier audio sur le front """ def mysy_openai_get_voice_file(diction): try: """ Verification des input acceptés """ field_list = ['text', 'audio_file_name'] RetObject = [] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas") return False, " Les informations fournies sont incorrectes", False out_file_name = "temp_direct/mysy_text_to_voice_89.mp3" if( "audio_file_name" in diction.keys() and diction['audio_file_name']): out_file_name = diction['audio_file_name'] print(" ### mysy_openai_get_voice_file fichier a traiter = ", out_file_name); with open(out_file_name, 'rb') as binary_file: binary_file_data = binary_file.read() base64_encoded_data = base64.b64encode(binary_file_data) base64_output = base64_encoded_data.decode('utf-8') retval = {} retval['document_mp3'] = base64_output RetObject.append(mycommon.JSONEncoder().encode(retval)) return True, RetObject except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : send_mysy_openai_to_voice_file" def test_mysy_openai_voice_to_text(diction): try: working_text = ["Ok mysy, créer une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, création une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, crée une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, ajoute une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, ajout une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, nouvelle une formation .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, met à jour formation code CCCCCC .... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, mettre à jour formation code CCCCCC.... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, update formation code CCCCCC.... stop . programme yyyyy stop objectifyyyyy stop ", "Ok mysy, modifier formation code CCCCCC.... stop . Titre xxxxxx stop Descriptionxxxxx stop ", "Ok mysy, modifie formation code CCCCCC.... stop . Titre xxxxxx stop Descriptionxxxxx stop "] for text in working_text : status, retval, message_id = mysy_openai_voice_to_text_traitement_1(text) return True, "Ok" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : test_mysy_openai_voice_to_text" """ Cette fonction prend une phrase et fait un decoupage comme suit : 1 - enleve le mot "ok Missy", "ok mysy" 2 - met la pharase d'instruction dans une tableau """ tab_order_mysy = ['ok, missi', 'ok missi', 'ok; missi','ok, missy', 'ok mysy'] def mysy_openai_voice_to_text_traitement_1(diction): try: voice_text = diction['voice_text'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data({'token': diction['token']}) if (local_status is not True): return local_status, my_partner print(" ### Texte initial = ", str(voice_text).strip().lower()) voice_text_work = str(voice_text).strip().lower() for val in tab_order_mysy: voice_text_work = str(voice_text_work).strip().lower().replace(str(val), '[INSTRUCTION]') voice_text_work = str(voice_text_work).strip().lower().replace("stop", '[stop]') new_diction = {} new_diction['text'] = str(voice_text_work) new_diction['partner_owner_recid'] = str(my_partner['recid']) new_diction['related_object'] = "myclass" new_diction['related_collection_recid'] = "dqsqddd0ed" local_status, local_retval_message, local_collection_instruction_id = mysy_openai_insert_voice_to_collection(new_diction) print(" || == > Texte Final = ", voice_text_work, " ## ID message = ", local_collection_instruction_id) print("\n") return local_status, local_retval_message, local_collection_instruction_id except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_voice_to_text_traitement_1", False """ Cette fonction prend une phrase du type : [instruction], modifie formation code cccccc.... [stop] . titre: xxxxxx [stop] description:xxxxx [stop] [fin] et insert les données dans une collection (voice_instruction) comme suit : [ { "_id": "02sdsddfdfd52", "related_collection": "myclass", "related_collection_recid": "", "instruction": "créer une formation ....[stop] Titre xxxxxx [stop] Description xxxxx[stop]", "action": "create", "partner_owner_recid": "ddddd", "valide": "1", "locked": "0", "data": [ { "field": "Titre", "value": "xxxxx" }, { "field": "Description", "value": "xxxx" } ] }, { "_id": "34sdrrdfdfd53", "related_collection": "myclass", "related_collection_recid": "ikddd933ob", "instruction": "update formation ....[stop]. programme: yyyyy[stop] objectif:yyyyy[stop]", "action": "update", "partner_owner_recid": "ddddd", "valide": "1", "locked": "0", "data": [ { "field": "objectif", "value": "yyyyy" }, ] } ] """ def mysy_openai_insert_voice_to_collection(diction): try: """ Verification des input acceptés """ field_list = ['text', "partner_owner_recid", "related_object", "related_collection_recid"] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas") return False, " Les informations fournies sont incorrectes", False """ Verification des champs obligatoires """ field_list_obligatoire = ['text', "partner_owner_recid", "related_object", "related_collection_recid"] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes", False voice_text_work = str(diction['text']).strip().lower().replace("[instruction] ", '') voice_text_work = str(voice_text_work).strip().lower().replace("[instruction], ", '') local_data = {} local_data['text'] = str(voice_text_work) local_data['related_collection'] = "myclass" local_status, local_retval, local_retval_action = mysy_openai_transform_instruction_text_to_json_tab(local_data) new_data = {} new_data['partner_owner_recid'] = diction['partner_owner_recid'] new_data['related_object'] = diction['related_object'] new_data['related_collection_recid'] = diction['related_collection_recid'] new_data['instruction'] = str(voice_text_work) new_data['valide'] = '1' new_data['locked'] = '0' new_data['date_create'] = str(datetime.now()) new_data['date_update'] = str(datetime.now()) if( local_status): new_data['action'] = str(local_retval_action) new_data['data'] = local_retval else: new_data['action'] = "" new_data['data'] = [] inserted_id = "" inserted_id = MYSY_GV.dbname['voice_instruction'].insert_one(new_data).inserted_id if (not inserted_id): mycommon.myprint( " Impossible de créer le message vocal dans la collection ") return False, " Impossible de créer le message vocal dans la collection (2) " return True, "Le message a été correctement enregistré en base", str(inserted_id) except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de créer le message vocal dans la collection ", False """ Cette fonction prends une phrase d'instruction du type : "créer une formation .... [stop] . titre: xxxxxx [stop] description:xxxxx [stop]" et la transforme en un tableau de json comme suit : [ { "field": "Titre", "value": "xxxxx" }, { "field": "Description", "value": "xxxx" } ] """ def mysy_openai_transform_instruction_text_to_json_tab(diction): try: """ Verification des input acceptés """ field_list = ['text', "related_collection",] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'existe pas") return False, " Les informations fournies sont incorrectes", False """ Verification des champs obligatoires """ field_list_obligatoire = ['text', "related_collection",] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes", False local_related_collection = str(diction['related_collection']).strip().lower() if( local_related_collection not in MYSY_GV.mysy_voice_instruction_related_collection): mycommon.myprint( str(inspect.stack()[0][3]) + " La collection "+str(diction['related_collection']).strip().lower()+" n'est pas" " autorisé pour les actions voices") return False, "La collection "+str(diction['related_collection']).strip().lower()+" n'est pas autorisé pour les actions voices", False """ Recuperer la liste des mots clés de cette collection qui est stocké ici : mysy_voice_instruction_collection_field """ collection_tab_champs_cle = [] if (local_related_collection in MYSY_GV.mysy_voice_instruction_collection_field.keys()): collection_tab_champs_cle = MYSY_GV.mysy_voice_instruction_collection_field[str(local_related_collection)] collection_tab_action_autorise = [] if (local_related_collection in MYSY_GV.mysy_voice_instruction_collection_action.keys()): collection_tab_action_autorise = MYSY_GV.mysy_voice_instruction_collection_action[str(local_related_collection)] #print(' ### Collection concernée = ', local_related_collection) #print(' ### Collection champs = ', collection_tab_champs_cle) #print(' ### Collection action autorise = ', collection_tab_action_autorise) new_tab = str(diction['text']).strip().lower().split("[stop]") #print(' ### new_tab = ', new_tab) tab_text = [] # Recuperation des champs/valeur for val in new_tab: for champ in collection_tab_champs_cle: champ = str(champ).strip().lower() if (champ in val): print(str(champ) + " = " + str(val).replace(str(champ), "")) my_value = str(val).replace(str(champ), "") node = {} node['field'] = str(champ) node['value'] = str(my_value) tab_text.append(node) # Recuperation de l'action action = "" for val in new_tab: for champ in collection_tab_action_autorise : for my_cle in champ.keys(): cle_action = my_cle if( cle_action in val ): #print(" OK POUR l'action ", cle_action, " ## la valeur = ", champ[str(cle_action)]) action = champ[str(cle_action)] return True, tab_text, action except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_assistant", False def mysy_openai_assistant(diction): try: requestion_response = {} my_assistant = client.beta.assistants.create( instructions="You are a personal math tutor. When asked a question, write and run Python code to answer the question.", name="Math Tutor", tools=[{"type": "code_interpreter"}], model="gpt-4o", ) print(my_assistant) return True, my_assistant except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_assistant" """ Cette fonction prend l'_id d'une instruction, un fichier audi. Si l'audit confirme l'instruction, alors l'instruction est traitée. si l'audit annule l'instruction, alors on annule l'instruction. """ def mysy_openai_confirme_voice_instruction(diction): try: voice_erreur_text = "" # Dictionnaire des champs utilisables field_list = ['token', 'voice_file', 'instruction_id'] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint( str(inspect.stack()[0][3]) + " - Le champ '" + val + "' n'existe pas, requete annulée") voice_erreur_text = "Erreur technique 1 : Instruction invalide " #return False, " Impossible de récupérer les informations" ''' Une fois qu'on a controlé que toutes les clés mise dans l'API sont correcte. etape precedente, On controle que les champs obligatoires sont presents dans la liste ''' field_list_obligatoire = ['token', 'voice_file', 'instruction_id'] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - : La valeur '" + val + "' n'est pas presente dans la liste des arguments ") voice_erreur_text = "Erreur technique 2 : Instruction invalide " #return False, " Impossible de récupérer les informations" local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): voice_erreur_text = "Erreur technique 4 : Instruction invalide " return False, voice_erreur_text, False file_name = "" if( "voice_file" in diction.keys() and diction['voice_file'] ): file_name = diction['voice_file'] else: mycommon.myprint( str(inspect.stack()[0][3]) + " Aucune instruction à traiter. Processus annulé. ") voice_erreur_text = "Erreur technique 3 : Instruction invalide " return False, " Aucune instruction à traiter. Processus annulé. ", False """ Verifier si l'instruction a été confirmée par le fichier audio """ audio_file = open(file_name, "rb") transcription = client.audio.transcriptions.create( model="whisper-1", file=audio_file, # response_format = "text" ) requestion_response = transcription.text print(" ### mysy_openai_confirme_voice_instruction message de confirmtion = ", requestion_response) status, tab_tokens = ela_Spacy.Ela_Tokenize(requestion_response) if (status is False): voice_erreur_text = " Je n'ai pas compris votre message." repeat = "1" return True, voice_erreur_text, repeat local_status, tab_retval = ela_Spacy.Ela_remove_stop_words(tab_tokens) print(" ### mysy_openai_confirme_voice_instruction tab_retval= ", tab_retval) confirmation_annulation_action = "" confirmation_annulation_action_value = "" # Verifier d'abord si c'est une annulation for voice_message_token in tab_retval: for confirmation_mot in MYSY_GV.VOICE_INSTRUCTION_ANNULATION : cmp_levenshtein = int(mycommon.levenshtein(voice_message_token, str(confirmation_mot))) if( cmp_levenshtein < 2 ): print(" ### Le message orale est 11 : ", str(confirmation_mot)) confirmation_annulation_action = str(confirmation_mot) confirmation_annulation_action_value = "0" continue # Puis verifier si c'est un message de confirmation pour exécuter l'instruction for voice_message_token in tab_retval: for confirmation_mot in MYSY_GV.VOICE_INSTRUCTION_CONFIRMATION: cmp_levenshtein = int(mycommon.levenshtein(voice_message_token, str(confirmation_mot))) if (cmp_levenshtein < 2): print(" ### Le message orale est 22 : ", str(confirmation_mot)) confirmation_annulation_action = str(confirmation_mot) confirmation_annulation_action_value = "1" continue print(" ### ACTION A REALISER EST : ", confirmation_annulation_action) if( confirmation_annulation_action_value == "0" ): # Il faut annuler l'instruction after_intruction_text = " L'instruction a été annulée" new_data = {} new_data['valide'] = '0' new_data['locked'] = '0' new_data['update_by'] = str(my_partner['_id']) new_data['date_update'] = str(datetime.now()) result = MYSY_GV.dbname['voice_instruction'].find_one_and_update( {'_id': ObjectId(str(diction['instruction_id'])), 'valide': '1', 'locked': '0'}, {"$set": new_data}, upsert=False, return_document=ReturnDocument.AFTER ) local_status, outpufile_name = mysy_openai_text_to_voice({'text': str(after_intruction_text)}) print(" le fichier audio de confirmation est : ", outpufile_name) return True, outpufile_name, "0" elif( confirmation_annulation_action_value == ""): # Impossible de comprendre le message after_intruction_text = " Je n'ai pas compris l'instruction. merci de répéter " local_status, outpufile_name = mysy_openai_text_to_voice({'text': str(after_intruction_text)}) print(" le fichier audio de confirmation est : ", outpufile_name) repeat = "1" return True, outpufile_name, repeat """ Verifier la validité de l'instruction_id """ is_valide_instruction_id_count = MYSY_GV.dbname['voice_instruction'].count_documents({'_id':ObjectId(str(diction['instruction_id'])), 'valide':'1', 'locked':'0'}) if( is_valide_instruction_id_count != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'instruction est invalide ") voice_erreur_text = "L'identifiant de l'instruction est invalide " if( voice_erreur_text and len(str(voice_erreur_text)) > 5): return False, voice_erreur_text, "0" audio_file = open(file_name, "rb") transcription = client.audio.transcriptions.create( model="whisper-1", file=audio_file, #response_format = "text" ) requestion_response = transcription.text """ requestion_response = "OK, Missy, crée la formation avec le titre Je vais à la plage, stop. Description, comment aller à la plage, stop." print(" ### AFFICHAGE GPT RESPONSE - mysy_openai_voice_to_text") """ print(" ### mysy_openai_confirme_voice_instruction = ", requestion_response) new_diction = {} new_diction['token'] = diction['token'] new_diction['instruction_id'] = diction['instruction_id'] print(" GRRR new_diction = ", new_diction) local_status, local_retval = voice_instruction_make_instruction(new_diction) if( local_status is False ): after_intruction_text = " Erreur - " + str(local_retval) else: after_intruction_text = " L'opération a été réalisée avec succès " print(" ### mysy_openai_confirme_voice_instruction text_a_retourner = ", after_intruction_text) local_status, outpufile_name = mysy_openai_text_to_voice({'text': str(after_intruction_text)}) print(" le fichier audio de confirmation est : ", outpufile_name) return True, outpufile_name, "0" except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - ERRORRRR AT Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de traiter : mysy_openai_voice_to_text", "0" """ Cette fonction prends une voice_instruction_id, puis effectue l'operation concernée """ def voice_instruction_make_instruction(diction): try: diction = mycommon.strip_dictionary(diction) print(" voice_instruction_make_instruction diction == ", diction) """ Verification des input acceptés """ field_list = ['token', 'instruction_id', ] incom_keys = diction.keys() for val in incom_keys: if val not in field_list and val.startswith('my_') is False: mycommon.myprint(str( inspect.stack()[0][3]) + " Le champ '" + val + "' n'est pas autorisé") return False, " Les informations fournies sont incorrectes" """ Verification des champs obligatoires """ field_list_obligatoire = ['token', 'instruction_id', ] for val in field_list_obligatoire: if val not in diction: mycommon.myprint( str(inspect.stack()[0][3]) + " - La valeur '" + val + "' n'est pas presente dans la liste des arguments ") return False, " Les informations fournies sont incorrectes" """ Verification de l'identité et autorisation de l'entité qui appelle cette API """ token = "" if ("token" in diction.keys()): if diction['token']: token = diction['token'] local_status, my_partner = mycommon.Check_Connexion_And_Return_Partner_Data(diction) if (local_status is not True): return local_status, my_partner """ Verifier la validité de l'instruction_id """ is_valide_instruction_id_count = MYSY_GV.dbname['voice_instruction'].count_documents( {'_id': ObjectId(str(diction['instruction_id'])), 'valide': '1', 'locked': '0'}) if (is_valide_instruction_id_count != 1): mycommon.myprint( str(inspect.stack()[0][3]) + " L'identifiant de l'instruction est invalide ") voice_erreur_text = "L'identifiant de l'instruction est invalide " instruction_id_data = MYSY_GV.dbname['voice_instruction'].find_one( {'_id': ObjectId(str(diction['instruction_id'])), 'valide': '1', 'locked': '0'}) """ Traitement pour les formations """ if( instruction_id_data['related_object'] == "myclass"): new_class_data = {} for mydata in instruction_id_data['data']: new_class_data[str(mydata['field']).strip().lower()] = mydata['value'] """ Normalisation de certains clés. par exemple 'titre' devient 'title' ==> a faire """ print(" ### new_class_data = ", new_class_data) if( instruction_id_data['action'] == "create"): local_status, local_retval = class_mgt.add_class(new_class_data) if (local_status is False): return local_status, local_retval return True, " OK " except Exception as e: exc_type, exc_obj, exc_tb = sys.exc_info() mycommon.myprint(str(inspect.stack()[0][3]) + " -" + str(e) + " - Line : " + str(exc_tb.tb_lineno)) return False, " Impossible de réaliser : 'voice_instruction_make_instruction' "