Spaces:
Sleeping
Sleeping
File size: 6,778 Bytes
d779a9b 69e9d44 859f8b3 69e9d44 d779a9b 69e9d44 859f8b3 d779a9b 859f8b3 d779a9b 859f8b3 d779a9b | 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 | # import dropbox
import openai
from config.settings import settings
from langchain_openai import ChatOpenAI
import sqlparse
import re
import os
class UtilityClass:
"""
Classifies user input as 'sql' (SQL/data context) or 'chat' (normal conversation) using an LLM.
"""
@staticmethod
def classify_user_input(text: str) -> str:
prompt = (
"Classify the following message as 'sql' if it is a database/data/SQL question, "
"or 'chat' if it is normal conversation.\n"
f"Message: {text}\n"
"Respond with only 'sql' or 'chat'."
)
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
api_key=settings.OPENAI_API_KEY
)
label = response.choices[0].message.content.strip().lower()
return label if label in ("sql", "chat") else "chat"
@staticmethod
def chat_response(text: str) -> str:
chat_prompt = f"You are a helpful assistant. Respond conversationally to: {text}"
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": chat_prompt}],
api_key=settings.OPENAI_API_KEY
)
return response.choices[0].message.content.strip()
# Helper to prepare JSON with heading from LLM and records
@staticmethod
def prepare_llm_heading_and_records(user_question, rows):
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, api_key=settings.OPENAI_API_KEY)
record_count = len(rows)
column_names = list(rows[0].keys()) if rows and isinstance(rows[0], dict) else []
heading_prompt = (
"Given the user's question and the following column names and record count, generate:\n"
"1. A short, clear, and specific heading for the results (do NOT use generic phrases like 'Here are the top X results I found').\n"
"2. A 1-2 sentence summary or description of what the results represent, using the question and columns for context.\n"
f"User question: {user_question}\nColumns: {column_names}\nRecord count: {record_count}\n"
"Return your answer as JSON with keys 'heading' and 'summary'. Do not mention the actual data."
)
heading = llm.invoke([{"role": "user", "content": heading_prompt}]).content.strip()
return {"heading": heading, "records": rows}
@staticmethod
def is_valid_sql_query(text: str):
if not text or not text.strip():
return False
# Remove markdown formatting like ```sql ... ```
text = re.sub(r"```sql|```", "", text, flags=re.IGNORECASE).strip()
# Collapse multiple spaces & line breaks into a single space
normalized = re.sub(r"\s+", " ", text).strip().upper()
# Common SQL query structure patterns (heuristics)
sql_patterns = [
r"^SELECT\s+.+\s+FROM\s+.+", # SELECT ... FROM ...
r"^INSERT\s+INTO\s+.+\s+VALUES\s*\(", # INSERT INTO ... VALUES (...)
r"^UPDATE\s+.+\s+SET\s+.+", # UPDATE ... SET ...
r"^CREATE\s+(TABLE|DATABASE)\s+.+", # CREATE TABLE/DATABASE ...
r"^WITH\s+.+\s+AS\s*\(.+\)" # WITH ... AS (...)
]
for pattern in sql_patterns:
# Format the SQL for readability
formatted_sql = sqlparse.format(text, reindent=True, keyword_case='upper')
return formatted_sql
return False
@staticmethod
def get_health_status() -> dict:
"""
Get the health status of the application and its dependencies.
Returns:
dict: Health status information including API status, database connectivity, and timestamp
"""
import time
from datetime import datetime
from app.db.db_connector import get_connection
health_data = {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"version": "1.0.0",
"checks": {}
}
# Check database connectivity
try:
with get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT 1")
result = cursor.fetchone()
if result:
health_data["checks"]["database"] = {
"status": "healthy",
"message": "Database connection successful"
}
else:
health_data["checks"]["database"] = {
"status": "unhealthy",
"message": "Database query failed"
}
health_data["status"] = "unhealthy"
except Exception as e:
health_data["checks"]["database"] = {
"status": "unhealthy",
"message": f"Database connection failed: {str(e)}"
}
health_data["status"] = "unhealthy"
# Check OpenAI API (if configured)
try:
if hasattr(settings, 'OPENAI_API_KEY') and settings.OPENAI_API_KEY:
health_data["checks"]["openai_api"] = {
"status": "configured",
"message": "OpenAI API key is configured"
}
else:
health_data["checks"]["openai_api"] = {
"status": "warning",
"message": "OpenAI API key not configured"
}
except Exception as e:
health_data["checks"]["openai_api"] = {
"status": "error",
"message": f"OpenAI API check failed: {str(e)}"
}
return health_data
# @staticmethod
# def download_sqlite_db_from_dropbox(dropbox_token: str, dropbox_db_path: str, local_db_path: str = None) -> str:
# """
# Downloads a SQLite DB file from Dropbox and returns the local file path.
# :param dropbox_token: Dropbox API access token.
# :param dropbox_db_path: Path to the DB file in Dropbox (e.g., '/folder/mydb.db').
# :param local_db_path: Local path to save the file. If None, saves to './downloaded_db.db'.
# :return: Local file path of the downloaded DB.
# """
# if local_db_path is None:
# local_db_path = os.path.join(os.getcwd(), "downloaded_db.db")
# dbx = dropbox.Dropbox(dropbox_token)
# with open(local_db_path, 'wb') as f:
# metadata, res = dbx.files_download(path=dropbox_db_path)
# f.write(res.content)
# return local_db_path
|