Update app2.py

app2.py

@@ -2,91 +2,64 @@ 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.future import select
 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__)
+from typing import List, Dict, Any
 
 # Global variables for database session and engine
 db_session = None
 engine = None
-monitoring_task = None  # Global variable to track the monitoring task
+monitoring_task = None
+logger = logging.getLogger(__name__)
+
+# Function for dynamically setting the database connection
+async def set_db_connection(host: str, port: str, user: str, password: str, db_name: str):
+    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}"
 
-# This will constantly check the database status and update the textbox
-async def update_db_status(db_status_textbox):
+# Periodic feed updater with error handling and logging improvements
+async def periodic_update_with_error_handling():
     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
+            await asyncio.sleep(300)  # Wait for 5 minutes before updating the feed content.
+            await update_feed_content()  # Update the feed content.
+        except Exception as e:  # Catch all exceptions for logging purposes.
+            logger.error(f"Error in periodic update: {e}")  # Improved logging message format.
 
-# Fetch the latest 20 articles and return them in a feed format
-async def update_feed_content():
+# Function to fetch RSS feed content from the provided URL with error handling and logging improvements.
+async def fetch_feed_content(feed_url: str) -> Dict[str, Any]:
     try:
-        result = await db_session.execute(
-            select(Article).order_by(Article.timestamp.desc()).limit(20)
-        )
+        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',
+            'link': feed_url,
             '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]
+            'items': [{'title': article.title, 'link': article.url, 'description': article.content, 'pubDate': str(article.timestamp)} for article in articles] if articles else []
         }
         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}"
+        logger.error(f"Error fetching feed content: {e}")
+        return {}
 
 # Main application that runs Gradio UI and background tasks
 async def main():
     global db_session, monitoring_task
     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")
@@ -95,95 +68,48 @@ async def main():
                 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"
-                )
+                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):
+        # 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):
             global monitoring_task
             urls = [url.strip() for url in target_urls_str.split(",")]
-            await set_db_connection(host, port, user, password, db_name)  # Connect to the DB
-            monitoring_task = asyncio.create_task(start_monitoring(urls, storage_loc, feed_enabled))  # Start monitoring
+            await set_db_connection(host, port, user, password, db_name)
+            monitoring_task = asyncio.create_task(start_monitoring(urls, storage_loc, feed_enabled))
             return "Monitoring started."
-
-        async def on_stop_click():  # Define the on_stop_click function
+        async def on_stop_click():
             global monitoring_task
             if monitoring_task:
-                monitoring_task.cancel()  # Cancel the monitoring task
+                monitoring_task.cancel()
                 monitoring_task = None
-                return "Monitoring stopped."
-            else:
-                return "Monitoring is not running."
-
+            return "Monitoring stopped."
         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]
-        )
-
+        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
-
+        asyncio.create_task(periodic_update_with_error_handling())
         # 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
-        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
-
-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
+# Launch the app using asyncio
 if __name__ == "__main__":
     asyncio.run(main())