backend/app/services/utility.py

# 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