Ela_Back/tools_cherif/mysy_openai_file.py

795 lines
30 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
import pymongo
import zeep
from pymongo import MongoClient
import json
from bson import ObjectId
import re
from datetime import datetime
import prj_common as mycommon
import secrets
import inspect
import sys, os
from flask import send_file
import csv
import pandas as pd
from pymongo import ReturnDocument
from math import isnan
import GlobalVariable as MYSY_GV
import ela_index_bdd_classes as eibdd
import email_mgt as email
from validate_email import validate_email
import jinja2
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
"""
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)
status, retval, collection_instruction_id = mysy_openai_voice_to_text_traitement_1(requestion_response)
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 liste ")
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(voice_text):
try:
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'] = "partner_owner_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 liste ")
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 liste ")
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 en cours et effectue le traitement
"""
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 liste ")
voice_erreur_text = "Erreur technique 2 : Instruction invalide "
#return False, " Impossible de récupérer les informations"
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é. "
"""
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
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
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 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 liste ")
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' "