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 # Correct async query API from sqlalchemy.orm import sessionmaker import logging import os import sys current_dir = os.path.dirname(os.path.realpath(__file__)) parent_dir = os.path.abspath(os.path.join(current_dir, "..")) sys.path.insert(0, parent_dir) logger = logging.getLogger(__name__) # This will constantly check the database status and update the textbox async def update_db_status(db_status_textbox): while True: try: await db_session.execute("SELECT 1") # Simple query to check connection db_status_textbox.update("Connected") except SQLAlchemyError: db_status_textbox.update("Disconnected") await asyncio.sleep(60) # Wait for 60 seconds before checking again # Fetch the latest 20 articles and return them in a feed format async def update_feed_content(): try: result = await db_session.execute( select(Article).order_by(Article.timestamp.desc()).limit(20) ) articles = result.scalars().all() # Fetch latest articles feed = { 'title': 'Website Changes Feed', 'link': 'http://yourwebsite.com/feed', 'description': 'Feed of changes detected on monitored websites.', 'items': [{ 'title': article.title, 'link': article.url, 'description': article.content, 'pubDate': str(article.timestamp) } for article in articles] } return feed except SQLAlchemyError as e: logger.error(f"Database error: {e}") return None # Periodic feed updater with error handling async def periodic_update_with_error_handling(): while True: try: await asyncio.sleep(300) # Wait for 5 minutes await update_feed_content() # Update the feed content except Exception as e: logger.error(f"Error in periodic update: {e}") # 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() 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}" # Main application that runs Gradio UI and background tasks async def main(): global db_session engine = None demo = gr.Blocks() # Define the Gradio interface with 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) gr.Markdown("## RSS Feed Reader Settings") feed_target_url = gr.Textbox(label="RSS Feed Target URL", placeholder="http://yourwebsite.com/feed") view_button = gr.Button("View Feed") target_urls = gr.Textbox( label="Target URLs (comma-separated)", placeholder="https://example.com, https://another-site.com" ) storage_location = gr.Textbox( label="Storage Location (CSV file path)", placeholder="/path/to/your/file.csv" ) feed_rss_checkbox = gr.Checkbox(label="Enable RSS Feed") start_button = gr.Button("Start Monitoring") stop_button = gr.Button("Stop Monitoring") with gr.Column(): feed_content = gr.JSON(label="RSS Feed Content") chatbot_interface = gr.Chatbot(type='messages') message_input = gr.Textbox(placeholder="Type your message here...") send_button = gr.Button("Send") # Define button actions async def on_start_click(target_urls_str: str, storage_loc: str, feed_enabled: bool, host: str, port: str, user: str, password: str, db_name: str): urls = [url.strip() for url in target_urls_str.split(",")] await set_db_connection(host, port, user, password, db_name) # Connect to the DB asyncio.create_task(start_monitoring(urls, storage_loc, feed_enabled)) # Start monitoring return "Monitoring started." async def on_view_feed_click(feed_url: str): # Logic to fetch and view RSS feed data based on URL return await fetch_feed_content(feed_url) stop_button.click(on_stop_click, outputs=[status_text]) view_button.click(on_view_feed_click, inputs=[feed_target_url], outputs=[feed_content]) send_button.click( chatbot_response, inputs=[message_input, chatbot_interface], outputs=[chatbot_interface, message_input] ) # Set up the timer for periodic updates feed_updater = gr.Timer(interval=300) feed_updater.tick(fn=update_feed_content, outputs=feed_content) # Load and check database status when the UI is loaded demo.load(update_db_status, outputs=db_status_textbox) asyncio.create_task(periodic_update_with_error_handling()) # Run periodic updates in the background # Launch the Gradio demo await demo.launch() async def fetch_feed_content(feed_url: str): # Logic to fetch RSS feed content from the provided URL # You would replace this with actual RSS fetching and parsing logic return { 'title': 'Sample Feed', 'link': feed_url, 'items': [ {'title': 'Sample Item 1', 'link': feed_url + '/1', 'description': 'This is a sample item.', 'pubDate': '2024-01-01'}, {'title': 'Sample Item 2', 'link': feed_url + '/2', 'description': 'This is another sample item.', 'pubDate': '2024-01-02'} ] } async def start_monitoring(urls, storage_location, feed_enabled): # Logic to start monitoring URLs and optionally save to CSV or enable RSS print(f"Starting monitoring for {urls}, saving to {storage_location}, RSS enabled: {feed_enabled}") return def stop_monitoring(url): # Logic to stop monitoring a specific URL print(f"Stopping monitoring for {url}") return async def chatbot_response(message, chat_interface): # Example chatbot logic to respond to a user message response = f"Echo: {message}" chat_interface.append((message, response)) return chat_interface, "" # Launch the app using asyncio if __name__ == "__main__": asyncio.run(main())