import frappe
import requests
from frappe import _
import json
import base64
import openai
from dotenv import load_dotenv
import os
load_dotenv()
erpnext_apikey = os.getenv(‘ERPNEXT_API’)
erpnext_secretkey = os.getenv(‘ERPNEXT_SECRET’)
openai.api_key = os.getenv(‘OPENAI_KEY’)
openai.organization = os.getenv(‘OPENAI_ORG’)
@frappe.whitelist(allow_guest=False)
def erpnextquery(doctype,query, username):
try:
# Fetch detailed data from ERPNext
detailed_data = fetch_data(doctype)
# Process ERPNext response based on user input
response = get_erp_response(detailed_data, query, username)
# Generate DALL-E image with OpenAI API
dalle_response = openai.Image.create(
prompt=f"Create a clear data visualization with the help of charts of {query} with the following data: {response}",
n=1,
size="1024x1024"
)
image_url = dalle_response['data'][0]['url']
image_data = requests.get(image_url).content
image_base64 = base64.b64encode(image_data).decode('utf-8')
# Return response and image data
return {'result': response, 'image': image_base64}
except Exception as e:
frappe.log_error(f"Error in erpnextquery: {str(e)}", "ERPNext Query API")
return {'error': f"An error occurred: {str(e)}"}, 500
def fetch_data(docvalue):
try:
# Fetch the list of documents from the specified Doctype using Frappe’s ORM
data = frappe.get_all(docvalue, fields=[“name”])
# Fetch details for each document
detailed_data = []
for item in data:
# Use frappe.get_doc to retrieve the full document by name
doc = frappe.get_doc(docvalue, item["name"])
detailed_data.append(doc.as_dict())
return detailed_data
except frappe.DoesNotExistError:
frappe.log_error(f"Document {docvalue} does not exist.", "Fetch Data")
return []
except Exception as e:
frappe.log_error(f"Error in fetch_data: {str(e)}", "Fetch Data")
return []
def get_erp_response(df1, usermessage, username):
try:
# Load chat messages from ERPNext
messages = loaderp_messages(df1, username)
# Post the user's message to ChatGPT
post_url = frappe.new_doc('Chatgpt')
post_url.role('user')
post_url.content(usermessage)
post_url.user(username)
post_url.docstatus(1)
post_url.insert(
ignore_permissions=True,
ignore_links=True,
ignore_if_duplicate=True,
ignore_mandatory=True
)
post_url.save(
ignore_permissions=True,
ignore_version=True
)
messages = loaderp_messages(df1, username)
gpt_response = openai.ChatCompletion.create(
model="gpt-4",
messages=messages
)
# Parse the GPT response
parsed_gpt_response = gpt_response['choices'][0]['message']['content']
print(parsed_gpt_response)
# Save the response back to ERPNext
saveerp_messages(parsed_gpt_response, username)
return parsed_gpt_response
except Exception as e:
frappe.log_error(f"Error in get_erp_response: {str(e)}", "Get ERP Response")
return "An error occurred."
def loaderp_messages(df1, username):
try:
data = frappe.get_all(“Chatgpt”,filters=[[“Chatgpt”, “user”, “=”, username]])
# If no messages exist, create a system message
if not data.get("data"):
post_url = frappe.new_doc('Chatgpt')
post_url.role('system')
post_url.content(f"I am an expert in Data Analysis and Business Strategy, specializing in deriving actionable insights from business data. I excel in simplifying complex information and extracting valuable insights from data: {df1}")
post_url.user(username)
post_url.docstatus(1)
post_url.insert(
ignore_permissions=True,
ignore_links=True,
ignore_if_duplicate=True,
ignore_mandatory=True
)
post_url.save(
ignore_permissions=True,
ignore_version=True
)
messages = [{"role": item.get("role"), "content": item.get("content")} for item in data.get("data", [])]
return messages
except requests.exceptions.RequestException as e:
frappe.log_error(f"Error in loaderp_messages: {str(e)}", "Load ERP Messages")
return []
def saveerp_messages(gpt_response, username):
try:
# Save GPT response to ERPNext
post_url = frappe.new_doc(‘Chatgpt’)
post_url.role(‘assistant’)
post_url.content(gpt_response)
post_url.user(username)
post_url.docstatus(1)
post_url.insert(
ignore_permissions=True,
ignore_links=True,
ignore_if_duplicate=True,
ignore_mandatory=True
)
post_url.save(
ignore_permissions=True,
ignore_version=True
)
except requests.exceptions.RequestException as e:
frappe.log_error(f"Error in saveerp_messages: {str(e)}", “Save ERP Messages”)
always gets “result”:“An error occurred.” this in output