Ela_Back/tools_cherif/mysy_openai_file.py

889 lines
35 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

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 dune 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' "