import ast 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 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" """ Test openai voice to text mysy """ def mysy_openai_voice_to_text(diction): try: print(" ### diction = ", diction) file_name = "temp_direct/mysy_test_voice_mp3.mp3" if( "voice_file" in diction.keys() and diction['voice_file'] ): file_name = diction['voice_file'] print(" ### Traitement du fichier audio : ", file_name) """ 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 = mysy_openai_voice_to_text_traitement_1(requestion_response) 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_voice_to_text" """ 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, "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 : mysy_openai_text_to_voice" """ Cette fonction permet d'envoyer un fichier audio sur le front """ def send_mysy_openai_to_voice_file(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 out_file_name = "temp_direct/mysy_text_to_voice_89.mp3" if os.path.exists(out_file_name): path = str(out_file_name) print("path == ", path) return send_file(path, as_attachment=True) 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_retval_id = mysy_openai_insert_voice_to_collection(new_diction) print(" || == > Texte Final = ", voice_text_work, " ## ID message = ", local_retval_id) print("\n") return local_status, local_retval_message, local_retval_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"