Spaces:
Sleeping
Sleeping
File size: 6,236 Bytes
50a9c82 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 |
import gradio as gr
import mysql.connector
import os
# Use a pipeline as a high-level helper
from transformers import pipeline
classifier_model = pipeline(
"zero-shot-classification", model="MoritzLaurer/deberta-v3-large-zeroshot-v1"
)
# get db info from env vars
db_host = os.environ.get("DB_HOST")
db_user = os.environ.get("DB_USER")
db_pass = os.environ.get("DB_PASS")
db_name = os.environ.get("DB_NAME")
db_connection = mysql.connector.connect(
host=db_host,
user=db_user,
password=db_pass,
database=db_name,
)
db_cursor = db_connection.cursor()
def get_potential_labels():
# get potential labels from db
potential_labels = db_cursor.execute(
"SELECT message_category_name FROM message_categorys"
)
potential_labels = db_cursor.fetchall()
potential_labels = [label[0] for label in potential_labels]
return potential_labels
# make gloabl variable for potential labels
global potential_labels
potential_labels = get_potential_labels()
# Function to handle the classification
def classify_email(constituent_email):
potential_labels = get_potential_labels()
model_out = classifier_model(constituent_email, potential_labels, multi_label=True)
top_labels = [
label
for label, score in zip(model_out["labels"], model_out["scores"])
if score > 0.95
]
if top_labels == []:
# Find the index of the highest score
max_score_index = model_out["scores"].index(max(model_out["scores"]))
# Return the label with the highest score
return model_out["labels"][max_score_index]
return ", ".join(top_labels)
# Function to handle saving data
def save_data(orig_user_email, constituent_email, labels, user_response):
# save the data to the database
# orig_user_email should have volley 0
# constituent_email should have volley 1
# user_response should have volley 2
# app_id, org_id, and person_id should be 0
# subject should be "Email Classification and Response Tracking"
# body should be the original email
try:
db_cursor.execute(
"INSERT INTO messages (app_id, org_id, person_id, communication_method_id, status_id, subject, body, send_date, message_type, volley) VALUES (0, 0, 0, 1, 1, 'Email Classification and Response Tracking', %s, NOW(), 'Email Classification and Response Tracking', 0)",
(orig_user_email,),
)
db_cursor.execute(
"INSERT INTO messages (app_id, org_id, person_id, communication_method_id, status_id, subject, body, send_date, message_type, volley) VALUES (0, 0, 0, 1, 1, 'Email Classification and Response Tracking', %s, NOW(), 'Email Classification and Response Tracking', 1)",
(constituent_email,),
)
db_cursor.execute(
"INSERT INTO messages (app_id, org_id, person_id, communication_method_id, status_id, subject, body, send_date, message_type, volley) VALUES (0, 0, 0, 1, 1, 'Email Classification and Response Tracking', %s, NOW(), 'Email Classification and Response Tracking', 2)",
(user_response,),
)
# insert a row into the message_categorys_associations table for each valid label in labels with the message_id of the constituent_email
labels = labels.split(", ")
for label in labels:
label_exists = db_cursor.execute(
"SELECT * FROM radmap_frog12.message_categorys WHERE message_category_name = %s",
(label,),
)
label_exists = db_cursor.fetchall()
if label_exists:
db_cursor.execute(
"INSERT INTO message_category_associations (message_id, message_category_id) VALUES ((SELECT id FROM messages WHERE body = %s), (SELECT id FROM message_categorys WHERE message_category_name = %s))",
(constituent_email, label),
)
db_connection.commit()
return "Data successfully saved to database"
except Exception as e:
print(e)
db_connection.rollback()
return "Error saving data to database"
# read auth from env vars
auth_username = os.environ.get("AUTH_USERNAME")
auth_password = os.environ.get("AUTH_PASSWORD")
# Define your username and password pairs
auth = [(auth_username, auth_password)]
# Start building the Gradio interface
# Start building the Gradio interface with two columns
with gr.Blocks() as app:
with gr.Row():
gr.Markdown("## Email Classification and Response Tracking")
with gr.Row():
with gr.Column():
email_labels_input = gr.Markdown(
"## Valid Email Labels\n ### " + ", ".join(potential_labels),
)
original_email_input = gr.TextArea(
placeholder="Enter the original email sent by you",
label="Your Original Email",
)
spacer1 = gr.Label(visible=False)
constituent_response_input = gr.TextArea(
placeholder="Enter the constituent's response",
label="Constituent's Response",
lines=15,
)
classify_button = gr.Button("Classify Email")
with gr.Column():
classification_output = gr.TextArea(
label="Current Email Labels",
lines=1,
interactive=True,
)
spacer2 = gr.Label(visible=False)
user_response_input = gr.TextArea(
placeholder="Enter your response to the constituent",
label="Your Response",
lines=25,
)
save_button = gr.Button("Save Data")
save_output = gr.Label(label="Backend Response")
# Define button actions
classify_button.click(
fn=classify_email,
inputs=constituent_response_input,
outputs=classification_output,
)
save_button.click(
fn=save_data,
inputs=[
original_email_input,
constituent_response_input,
classification_output,
user_response_input,
],
outputs=save_output,
)
# Launch the app
app.launch(auth=auth, debug=True)
|