Ela_Back/tools_cherif/mysy_openai_file.py

516 lines
19 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 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 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"
"""
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"