app.py

import asyncio
import gradio as gr
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import sessionmaker
import logging
import threading
import time

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Global variables
db_session = None
engine = None

# Function for dynamically setting the database connection
async def set_db_connection(host, port, user, password, db_name):
    global db_session, engine
    try:
        engine = create_async_engine(
            f"mysql+aiomysql://{user}:{password}@{host}:{port}/{db_name}",
            echo=False
        )
        Session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
        db_session = Session()
        logger.info("Database connection established.")
        return "Database connection established."
    except Exception as e:
        logger.error(f"Failed to establish database connection: {e}")
        return f"Failed to connect to database: {e}"

# Function to update database status
def update_db_status():
    global db_session
    try:
        if db_session:
            asyncio.run(db_session.execute(select(1)))
            return "Connected"
        else:
            return "Not connected"
    except SQLAlchemyError:
        return "Disconnected"

# Background task to update status
def background_update(db_status_textbox):
    while True:
        status = update_db_status()
        db_status_textbox.value = status
        logger.info(f"Database status updated: {status}")
        time.sleep(60)

# Main application that runs Gradio UI and background tasks
def main():
    with gr.Blocks() as demo:
        gr.Markdown("# Website Monitor and Chatbot")

        with gr.Row():
            with gr.Column():
                gr.Markdown("## Database Settings")
                db_host = gr.Textbox(label="Database Host", placeholder="localhost", value="localhost")
                db_port = gr.Textbox(label="Database Port", placeholder="3306", value="3306")
                db_user = gr.Textbox(label="Database User", placeholder="username", value="")
                db_pass = gr.Textbox(label="Database Password", placeholder="password", type="password", value="")
                db_name = gr.Textbox(label="Database Name", placeholder="database_name", value="monitoring")

                db_status_textbox = gr.Textbox(label="Database Status", interactive=False)
                status_text = gr.Textbox(label="Status", interactive=False)

                connect_button = gr.Button("Connect to Database")

        # Connect button click event
        connect_button.click(
            set_db_connection,
            inputs=[db_host, db_port, db_user, db_pass, db_name],
            outputs=[status_text]
        )

        # Start background task to update status
        threading.Thread(target=background_update, args=(db_status_textbox,), daemon=True).start()

    # Launch the Gradio interface with a timeout
    logger.info("Launching Gradio interface...")
    demo.launch(prevent_thread_lock=True)
    logger.info("Gradio interface launched successfully.")

if __name__ == "__main__":
    main()
    logger.info("Main function completed. App is running.")