Spaces:
Runtime error
Runtime error
File size: 6,857 Bytes
3f315aa |
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 |
import gradio as gr
import re
from neo4j import GraphDatabase
# Static Neo4j connection details
NEO4J_URI = "bolt://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "0507@#Ny"
# Function to query details from Neo4j
def query_details_from_neo4j(issue_key, driver):
details = {}
try:
with driver.session() as session:
result = session.run(
"""
MATCH (i:IssueKey {key: $issue_key})
OPTIONAL MATCH (i)-[:HAS_SUMMARY]->(s:Summary)
OPTIONAL MATCH (i)-[:HAS_CREATION_DATE]->(cd:CreationDate)
OPTIONAL MATCH (i)-[:HAS_ASSIGNEE]->(a:Assignee)
OPTIONAL MATCH (i)-[:HAS_REPORTER]->(r:Reporter)
OPTIONAL MATCH (i)-[:HAS_REQUIREMENT_ID]->(rid:RequirementId)
OPTIONAL MATCH (i)-[:HAS_LABELS]->(l:Labels)
OPTIONAL MATCH (i)-[:HAS_COMPONENTS]->(c:Components)
RETURN s.text AS summary, cd.value AS creation_date, a.name AS assignee,
r.name AS reporter, rid.value AS requirement_id, l.value AS labels,
c.value AS components
""",
issue_key=issue_key
)
record = result.single()
if record:
details["summary"] = record["summary"]
details["creation_date"] = record["creation_date"]
details["assignee"] = record["assignee"]
details["reporter"] = record["reporter"]
details["requirement_id"] = record["requirement_id"]
details["labels"] = record["labels"]
details["components"] = record["components"]
except Exception as e:
print(f"Error querying Neo4j: {e}")
return details
# Function to extract entities from user input
def extract_entities_from_user_input(user_input):
entities = {}
issue_key_match = re.search(r"[A-Z]+-\d+", user_input)
if issue_key_match:
entities["issue_key"] = issue_key_match.group(0)
requested_details = re.findall(r"(issue key|creation date|assignee|reporter|requirement id|labels|components|summary)", user_input, flags=re.IGNORECASE)
entities["requested_details"] = requested_details
return entities
# Function to handle the random response from the chatbot
def random_response(driver, message, history):
try:
# Process user input
entities = extract_entities_from_user_input(message)
# Check if issue key is provided
issue_key = entities.get("issue_key")
if issue_key:
details = query_details_from_neo4j(issue_key, driver)
if details:
requested_details = entities.get("requested_details", [])
if not requested_details:
bot_message = "Please specify the details you want (e.g., 'show me creation date and summary')"
else:
bot_message = ""
for detail in requested_details:
if detail.lower() == "issue key":
bot_message += f"Issue Key: {issue_key}\n"
elif detail.lower() == "creation date":
bot_message += f"Creation Date: {details['creation_date']}\n"
elif detail.lower() == "assignee":
bot_message += f"Assignee: {details['assignee']}\n"
elif detail.lower() == "reporter":
bot_message += f"Reporter: {details['reporter']}\n"
elif detail.lower() == "requirement id":
bot_message += f"Requirement ID: {details['requirement_id']}\n"
elif detail.lower() == "labels":
bot_message += f"Labels: {details['labels']}\n"
elif detail.lower() == "components":
bot_message += f"Components: {details['components']}\n"
elif detail.lower() == "summary":
bot_message += f"Summary: {details['summary']}\n"
else:
bot_message += f"Unknown detail: {detail}\n"
else:
bot_message = f"No details found for issue key {issue_key}"
else:
bot_message = "Issue key not found in the query"
history.append(bot_message)
return history, "\n".join(history)
except Exception as e:
return history, f"Error connecting to Neo4j: {e}"
# Function to handle the authentication process
def authenticate(user, password):
try:
# Neo4j driver initialization
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
with driver.session() as session:
session.run("RETURN 1")
return driver, True, "Authentication successful!"
except Exception as e:
return None, False, f"Authentication failed: {e}"
# Function to show the chatbot page upon successful authentication
def show_chatbot_page(auth_status, auth_message):
if auth_status:
return gr.update(visible=False), gr.update(visible=True), auth_message
else:
return gr.update(visible=True), gr.update(visible=False), auth_message
# Main Gradio app
with gr.Blocks() as demo:
auth_status = gr.State(False)
driver = gr.State(None)
history = gr.State([])
# Authentication Page
with gr.Column() as login_page:
user = gr.Textbox(label="Jira Username")
password = gr.Textbox(label="Jira Password", type="password")
login_button = gr.Button("Login")
auth_message = gr.Textbox(label="Authentication Status", interactive=False)
# Chatbot Interface
with gr.Column(visible=False) as chatbot_page:
message = gr.Textbox(label="Enter your query", lines=3)
output = gr.Textbox(label="Response", lines=15)
submit_button = gr.Button("Query")
clear_button = gr.Button("Clear")
undo_button = gr.Button("Undo")
def clear_chat(history):
history.clear()
return [], ""
def undo_last(history):
if history:
history.pop()
return history, "\n".join(history)
login_button.click(authenticate, inputs=[user, password], outputs=[driver, auth_status, auth_message]).then(
show_chatbot_page, inputs=[auth_status, auth_message], outputs=[login_page, chatbot_page, auth_message]
)
submit_button.click(random_response, inputs=[driver, message, history], outputs=[history, output])
clear_button.click(clear_chat, inputs=[history], outputs=[history, output])
undo_button.click(undo_last, inputs=[history], outputs=[history, output])
if __name__ == "__main__":
demo.launch(share=True)
|