Skip to content

Commit

Permalink
refactor: encapsulated methods to a single doctype - AutoInscribe Int…
Browse files Browse the repository at this point in the history
…egration
  • Loading branch information
deepak-redsoftware committed Jan 5, 2024
1 parent f878852 commit 47b8c75
Show file tree
Hide file tree
Showing 8 changed files with 195 additions and 107 deletions.
Empty file.
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
// Copyright (c) 2024, RedSoft Solutions Pvt. Ltd. and contributors
// For license information, please see license.txt

// frappe.ui.form.on("AutoInscribe Integration", {
// refresh(frm) {

// },
// });
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
{
"actions": [],
"allow_rename": 1,
"creation": "2024-01-05 10:45:27.907271",
"doctype": "DocType",
"engine": "InnoDB",
"field_order": [
"section_break_ozge"
],
"fields": [
{
"fieldname": "section_break_ozge",
"fieldtype": "Section Break"
}
],
"index_web_pages_for_search": 1,
"issingle": 1,
"links": [],
"modified": "2024-01-05 10:45:39.174023",
"modified_by": "Administrator",
"module": "AutoInscribe",
"name": "AutoInscribe Integration",
"owner": "Administrator",
"permissions": [
{
"create": 1,
"delete": 1,
"email": 1,
"print": 1,
"read": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"sort_field": "modified",
"sort_order": "DESC",
"states": [],
"track_changes": 1
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
# Copyright (c) 2024, RedSoft Solutions Pvt. Ltd. and contributors
# For license information, please see license.txt

import frappe
from frappe.model.document import Document
import base64
import requests
from openai import OpenAI
from google.cloud import vision
from google.oauth2 import service_account

# autoinscribe_settings = frappe.get_doc("AutoInscribe Settings").as_dict()

class AutoInscribeIntegration(Document):
'''Encapsulates a set of methods used to make external calls to OpenAI API & Google Vision API'''

def get_openai_gpt_key(self):
'''Returns decrypted OpenAI API key from AutoInscribe Settings'''

return frappe.get_single("AutoInscribe Settings").get_password('openai_gpt_key')

def get_vision_client_email(self):
'''Returns Vision Client Email from AutoInscribe Settings'''

return frappe.get_single("AutoInscribe Settings").as_dict()["vision_client_email"]

def get_vision_project_id(self):
'''Returns Vision Project ID from AutoInscribe Settings'''

return frappe.get_single("AutoInscribe Settings").as_dict()["vision_project_id"]

def get_vision_token_uri(self):
'''Returns Vision Token URI from AutoInscribe Settings'''

return frappe.get_single("AutoInscribe Settings").as_dict()["vision_token_uri"]

def ask_gpt(self, prompt):
'''Returns response from OpenAI API given a prompt'''

try:
gpt_client = OpenAI(api_key=self.get_openai_gpt_key())
chat_completion = gpt_client.chat.completions.create(
messages=[
{
"role": "user",
"content": prompt,
}
],
model="gpt-3.5-turbo-1106",
)
return chat_completion.choices[0].message.content.strip()
except Exception as e:
frappe.throw("Please enter a valid OpenAI API key in AutoInscribe Settings")

def extract_text_from_img(self, img_url):
'''Extracts and returns first_name, middle_name, last_name, gender, salutation, designation contact_numbers, email_ids, company_name, website, address, mobile_number, phone_number, city, state and country from an image given the image URL'''

try:
credentials = service_account.Credentials.from_service_account_info({
"type": "service_account",
"project_id": self.get_vision_project_id(),
"private_key": frappe.conf["vision_private_key"],
"client_email": self.get_vision_client_email(),
"token_uri": self.get_vision_token_uri(),
})
response = requests.get(img_url)
# Encode the image content to base64
base64_img = base64.b64encode(response.content).decode('utf-8')
client = vision.ImageAnnotatorClient(credentials=credentials)
img_data = base64.b64decode(base64_img)
# Create an image object
image = vision.Image(content=img_data)
# Perform OCR on the image
response = client.text_detection(image=image)
texts = response.text_annotations
except Exception as e:
frappe.throw("Please check your AutoInscribe Settings and try again")
# Extracting detected text
if texts:
detected_text = texts[0].description
prompt = f"From the following text, identify the first_name, middle_name, last_name, gender, salutation, designation contact_numbers, email_ids, company_name, website, address, mobile_number, phone_number, city, state, country: {detected_text}. Output must be a string containing one key-value pair per line and for absence of values use 'NULL' for value as placeholder. contact_numbers and email_ids must be comma-separated if there are multiple. Guess the salutation and gender. gender can be Male, Female, Transgender or Other. phone_number must be the telephone number whereas mobile_number must be the mobile number. country must have the value as full country name, e.g, US becomes United States, UK becomes United Kingdom."
reply = self.ask_gpt(prompt)
return reply
else:
return "No text detected"

def create_address(self, address):
'''Given an address string, extract city, state, postal_code, country and create an address if country exists & return the inserted doc. Return None otherwise.'''

prompt = f"From the following address text, identify city, state, country and postal_code: {address}. Output must be a string containing one key-value pair per line and for absence of values use 'NULL'. country must have the value as full country name, e.g, US becomes United States, UK becomes United Kingdom"
reply = self.ask_gpt(prompt)
addr_lines = reply.strip().splitlines()
city = addr_lines[0].split(':')[1].strip()
state = addr_lines[1].split(':')[1].strip()
postal_code = addr_lines[3].split(':')[1].strip()
country = addr_lines[2].split(':')[1].strip()
country_exists = frappe.db.exists("Country", {"country_name": country})
if country_exists:
doc = frappe.get_doc({
"doctype": "Address",
"address_title": address,
"address_type": "Office",
"address_line1": address,
"city": city if city != "NULL" else None,
"state": state if state != "NULL" else None,
"country": country,
"pincode": postal_code if postal_code != "NULL" else None,
})
doc.insert()
return doc
else:
return None
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
# Copyright (c) 2024, RedSoft Solutions Pvt. Ltd. and Contributors
# See license.txt

# import frappe
from frappe.tests.utils import FrappeTestCase


class TestAutoInscribeIntegration(FrappeTestCase):
pass
21 changes: 21 additions & 0 deletions autoinscribe/fixtures/server_script.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
[
{
"allow_guest": 0,
"api_method": null,
"cron_format": null,
"disabled": 0,
"docstatus": 0,
"doctype": "Server Script",
"doctype_event": "Before Save",
"enable_rate_limit": 0,
"event_frequency": "All",
"modified": "2024-01-05 14:32:56.543165",
"module": "AutoInscribe",
"name": "AutoInscribe Contact Error",
"rate_limit_count": 5,
"rate_limit_seconds": 86400,
"reference_doctype": "Contact",
"script": "if doc.custom_upload_image and not doc.first_name:\n frappe.msgprint(\"Please check your AutoInscribe Settings and try again\", title=\"Invalid AutoInscribe Settings\", indicator=\"red\")\n",
"script_type": "DocType Event"
}
]
1 change: 1 addition & 0 deletions autoinscribe/hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,7 @@
fixtures = [
{
"doctype": "Client Script",
"doctype": "Server Script",
},
{
"dt": "Custom Field",
Expand Down
111 changes: 4 additions & 107 deletions autoinscribe/inscribe/api.py
Original file line number Diff line number Diff line change
@@ -1,118 +1,15 @@
'''A set of methods used to make API calls to OpenAI & Google Vision API'''
'''Responsible for calling methods in doctype class "AutoInscribe Integration"'''

import frappe
import base64
import requests
from openai import OpenAI
from google.cloud import vision
from google.oauth2 import service_account

autoinscribe_settings = frappe.get_doc("AutoInscribe Settings").as_dict()

def get_openai_gpt_key():
'''Returns decrypted OpenAI API key from AutoInscribe Settings'''
return frappe.get_single("AutoInscribe Settings").get_password('openai_gpt_key')


def get_vision_client_email():
'''Returns Vision Client Email from AutoInscribe Settings'''
return autoinscribe_settings["vision_client_email"]


def get_vision_project_id():
'''Returns Vision Project ID from AutoInscribe Settings'''
return autoinscribe_settings["vision_project_id"]


def get_vision_token_uri():
'''Returns Vision Token URI from AutoInscribe Settings'''
return autoinscribe_settings["vision_token_uri"]


credentials = service_account.Credentials.from_service_account_info({
"type": "service_account",
"project_id": get_vision_project_id(),
"private_key": frappe.conf["vision_private_key"],
"client_email": get_vision_client_email(),
"token_uri": get_vision_token_uri(),
})


@frappe.whitelist()
def ask_gpt(prompt):
'''Returns response from OpenAI API given a prompt'''
try:
gpt_client = OpenAI(api_key=get_openai_gpt_key())
chat_completion = gpt_client.chat.completions.create(
messages=[
{
"role": "user",
"content": prompt,
}
],
model="gpt-3.5-turbo-1106",
)
return chat_completion.choices[0].message.content.strip()
except Exception as e:
frappe.throw("Please enter a valid OpenAI API key in AutoInscribe Settings")


@frappe.whitelist()
def get_gpt_task(**args):
'''Returns task subject & description as a response from OpenAI API given task details'''
task_details = args.get('task_details')
prompt = f"You are an assistant that is responsible for generating the Subject and Details of the following task: {task_details}. Subject must be a string that gives an overview of the task. Details must be a formatted HTML body that describes that task."
reply = ask_gpt(prompt)
return reply

doc = frappe.get_single("AutoInscribe Integration")

@frappe.whitelist()
def extract_text_from_img(img_url):
'''Extracts and returns first_name, middle_name, last_name, gender, salutation, designation contact_numbers, email_ids, company_name, website, address, mobile_number, phone_number, city, state and country from an image given the image URL'''
response = requests.get(img_url)
# Encode the image content to base64
base64_img = base64.b64encode(response.content).decode('utf-8')
client = vision.ImageAnnotatorClient(credentials=credentials)
img_data = base64.b64decode(base64_img)
# Create an image object
image = vision.Image(content=img_data)
# Perform OCR on the image
response = client.text_detection(image=image)
texts = response.text_annotations
# Extracting detected text
if texts:
detected_text = texts[0].description
prompt = f"From the following text, identify the first_name, middle_name, last_name, gender, salutation, designation contact_numbers, email_ids, company_name, website, address, mobile_number, phone_number, city, state, country: {detected_text}. Output must be a string containing one key-value pair per line and for absence of values use 'NULL' for value as placeholder. contact_numbers and email_ids must be comma-separated if there are multiple. Guess the salutation and gender. gender can be Male, Female, Transgender or Other. phone_number must be the telephone number whereas mobile_number must be the mobile number. country must have the value as full country name, e.g, US becomes United States, UK becomes United Kingdom."
reply = ask_gpt(prompt)
return reply
else:
return "No text detected"
return doc.extract_text_from_img(img_url)


@frappe.whitelist()
def create_address(address):
'''Given an address string, extract city, state, postal_code, country and create an address if country exists & return the inserted doc. Return False otherwise.'''
prompt = f"From the following address text, identify city, state, country and postal_code: {address}. Output must be a string containing one key-value pair per line and for absence of values use 'NULL'. country must have the value as full country name, e.g, US becomes United States, UK becomes United Kingdom"
reply = ask_gpt(prompt)
addr_lines = reply.strip().splitlines()
city = addr_lines[0].split(':')[1].strip()
state = addr_lines[1].split(':')[1].strip()
postal_code = addr_lines[3].split(':')[1].strip()
country = addr_lines[2].split(':')[1].strip()
country_exists = frappe.db.exists("Country", {"country_name": country})
if country_exists:
doc = frappe.get_doc({
"doctype": "Address",
"address_title": address,
"address_type": "Office",
"address_line1": address,
"city": city if city != "NULL" else None,
"state": state if state != "NULL" else None,
"country": country,
"pincode": postal_code if postal_code != "NULL" else None,
})
doc.insert()
return doc
else:
return False
return doc.create_address(address)

0 comments on commit 47b8c75

Please sign in to comment.