Update app2.py

app2.py

@@ -1,202 +1,279 @@
 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 os
 import sys
-from typing import List, Dict, Any
-from datetime import datetime
-from pytz import timezone
-import pytz
 import requests
 from bs4 import BeautifulSoup
-import pandas as pd
-import numpy as np
-import matplotlib.pyplot as plt
-from sklearn.model_selection import train_test_split
-from sklearn.linear_model import LinearRegression
-from sklearn import metrics
-
-# Global variables for database session and engine
+from datetime import datetime
+from pytz import timezone
+from typing import List, Dict, Any
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.future import select
+from sqlalchemy.orm import sessionmaker
+from dotenv import load_dotenv
+from pydantic import BaseSettings
+
+# --- Configuration ---
+load_dotenv()  # Load environment variables from .env file
+
+class Settings(BaseSettings):
+    DATABASE_HOST: str = os.getenv("DATABASE_HOST", "localhost")
+    DATABASE_PORT: int = int(os.getenv("DATABASE_PORT", 3306))
+    DATABASE_USER: str = os.getenv("DATABASE_USER", "")
+    DATABASE_PASSWORD: str = os.getenv("DATABASE_PASSWORD", "")
+    DATABASE_NAME: str = os.getenv("DATABASE_NAME", "monitoring")
+    RSS_FEED_URL: str = os.getenv("RSS_FEED_URL", "")
+    STORAGE_LOCATION: str = os.getenv("STORAGE_LOCATION", "")
+    CHATBOT_API_KEY: str = os.getenv("CHATBOT_API_KEY", "")  # Add your chatbot API key
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+settings = Settings()
+
+# --- Database Model (Example) ---
+class Article:
+    def __init__(self, title, url, content, timestamp):
+        self.title = title
+        self.url = url
+        self.content = content
+        self.timestamp = timestamp
+
+# --- Global Variables ---
 db_session = None
 engine = None
 monitoring_task = None
 logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
 
-# Configure logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-
-# Function for dynamically setting the database connection
-async def set_db_connection(host: str, port: str, user: str, password: str, db_name: str):
+# --- Database Connection ---
+async def set_db_connection():
     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()
+        engine = create_async_engine(
+            f"mysql+aiomysql://{settings.DATABASE_USER}:{settings.DATABASE_PASSWORD}@{settings.DATABASE_HOST}:{settings.DATABASE_PORT}/{settings.DATABASE_NAME}",
+            echo=False,
+        )
+        async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+        db_session = async_session_maker()
+        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}"
 
-# Periodic feed updater with error handling and logging improvements
-async def periodic_update_with_error_handling():
+# --- Website Monitoring ---
+async def start_monitoring(urls: List[str], storage_loc: str, feed_enabled: bool):
     while True:
-        try:
-            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.
+        for url in urls:
+            try:
+                content = await scrape_website(url)
+                analysis = await analyze_website_content(content)
+                await store_data(url, content, analysis)
+                if feed_enabled:
+                    await update_feed_content()
+            except Exception as e:
+                logger.error(f"Error monitoring website {url}: {e}")
+        await asyncio.sleep(300)  # Check every 5 minutes
 
-# 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]:
+async def store_data(url: str, content: str, analysis: Dict[str, Any]):
     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': 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] if articles else []
-        }
-        return feed
+        async with db_session as session:
+            article = Article(
+                title=f"Change on {url}",
+                url=url,
+                content=content,
+                timestamp=datetime.now(timezone("UTC")),
+            )
+            session.add(article)
+            await session.commit()
     except Exception as e:
-        logger.error(f"Error fetching feed content: {e}")
-        return {}
+        logger.error(f"Error storing data: {e}")
+
+async def update_feed_content():
+    try:
+        # ... (Your RSS feed generation logic) ...
+        pass
+    except Exception as e:
+        logger.error(f"Error updating feed content: {e}")
 
-# Function to scrape website content
+# --- Website Scraping and Analysis ---
 async def scrape_website(url: str) -> str:
     try:
         response = requests.get(url)
-        soup = BeautifulSoup(response.text, 'html.parser')
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, "html.parser")
         return soup.get_text()
-    except Exception as e:
-        logger.error(f"Error scraping website: {e}")
+    except requests.exceptions.RequestException as e:
+        logger.error(f"Error scraping website {url}: {e}")
         return ""
 
-# Function to analyze website content
 async def analyze_website_content(content: str) -> Dict[str, Any]:
     try:
-        # Perform sentiment analysis using Natural Language Processing (NLP) techniques
-        # For simplicity, we'll use a basic sentiment analysis approach
+        # Perform sentiment analysis or other analysis
         sentiment = "Positive" if content.count("good") > content.count("bad") else "Negative"
-        return {'sentiment': sentiment}
+        return {"sentiment": sentiment}
     except Exception as e:
         logger.error(f"Error analyzing website content: {e}")
         return {}
 
-# Function to predict website traffic
+# --- Website Traffic Prediction ---
 async def predict_website_traffic(url: str) -> Dict[str, Any]:
     try:
-        # Use machine learning model to predict website traffic
-        # For simplicity, we'll use a basic linear regression model
-        X = pd.DataFrame({'url': [url]})
-        y = pd.DataFrame({'traffic': [100]})  # Replace with actual traffic data
-        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
-        model = LinearRegression()
-        model.fit(X_train, y_train)
-        y_pred = model.predict(X_test)
-        return {'traffic': y_pred}
+        # ... (Your machine learning model for traffic prediction) ...
+        return {"traffic": 100}  # Placeholder
     except Exception as e:
         logger.error(f"Error predicting website traffic: {e}")
         return {}
 
-# Function to update database status
-async def update_db_status():
+# --- Chatbot Integration ---
+async def chatbot_response(message: str, chat_history: List[Dict[str, str]]) -> List[Dict[str, str]]:
     try:
-        await db_session.execute("SELECT 1")
-        return "Connected"
-    except SQLAlchemyError:
-        return "Disconnected"
+        if not settings.CHATBOT_API_KEY:
+            raise ValueError("Chatbot API key is not set.")
+        # ... (Your code to call the chatbot API) ...
+        # Example using a hypothetical API:
+        response = requests.post(
+            "https://your-chatbot-api.com/chat",
+            json={"message": message, "api_key": settings.CHATBOT_API_KEY},
+        )
+        response.raise_for_status()
+        bot_response = response.json()["response"]
+        chat_history.append({"role": "user", "content": message})
+        chat_history.append({"role": "bot", "content": bot_response})
+        return chat_history, ""
+    except Exception as e:
+        logger.error(f"Error calling chatbot API: {e}")
+        chat_history.append({"role": "bot", "content": "Sorry, I'm having trouble responding right now."})
+        return chat_history, ""
 
-# Main application that runs Gradio UI and background tasks
+# --- Database Status ---
+async def update_db_status():
+    global db_session, engine
+    if db_session and engine:
+        try:
+            await db_session.execute(select(1))
+            return "Database connection is active."
+        except SQLAlchemyError as e:
+            return f"Database error: {e}"
+    else:
+        return "Database connection not established."
+
+# --- Gradio UI ---
 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")
-                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")
+                # ... (Database settings are now configured in .env file) ...
                 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")
+                # ... (RSS feed settings are now configured in .env file) ...
                 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")
+                # ... (Storage location is now configured in .env file) ...
                 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', stateful=True)  # Enable session state
+                chatbot_interface = gr.Chatbot(type="messages")
                 message_input = gr.Textbox(placeholder="Type your message here...")
                 send_button = gr.Button("Send")
                 scrape_button = gr.Button("Scrape Website")
                 analyze_button = gr.Button("Analyze Website Content")
                 predict_button = gr.Button("Predict Website Traffic")
                 scrape_output = gr.Textbox(label="Scraped Website Content", interactive=False)
-                analyze_output = gr.JSON(label="Website Content Analysis")  # Removed interactive=False
-                predict_output = gr.JSON(label="Website Traffic Prediction")  # Removed interactive=False
-        # 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)
-            monitoring_task = asyncio.create_task(start_monitoring(urls, storage_loc, feed_enabled))
-            return "Monitoring started."
-        async def on_stop_click():
-            global monitoring_task
-            if monitoring_task:
-                monitoring_task.cancel()
-                monitoring_task = None
-            return "Monitoring stopped."
-        async def on_view_feed_click(feed_url: str):
-            return await fetch_feed_content(feed_url)
-        async def on_scrape_click(url: str):
-            return await scrape_website(url)
-        async def on_analyze_click(content: str):
-            return await analyze_website_content(content)
-        async def on_predict_click(url: str):
-            return await predict_website_traffic(url)
-        stop_button.click(on_stop_click, outputs=[status_text])
-        view_button.click(on_view_feed_click, inputs=[feed_target_url], outputs=[feed_content])
-        scrape_button.click(on_scrape_click, inputs=[target_urls], outputs=[scrape_output])
-        analyze_button.click(on_analyze_click, inputs=[scrape_output], outputs=[analyze_output])
-        predict_button.click(on_predict_click, inputs=[target_urls], outputs=[predict_output])
-        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())
-        # Launch the Gradio demo with a custom theme
-        await demo.launch(theme="default", title="Website Monitor and Chatbot")
-
-# Function to handle chatbot responses with session state
-async def chatbot_response(message: str, chat_history: List[str]) -> List[str]:
-    chat_history = chat_history or []
-    response = f"Echo: {message}"
-    chat_history.append((message, response))
-    return chat_history, ""
-
-# Function to start monitoring
-async def start_monitoring(urls: List[str], storage_location: str, feed_enabled: bool):
-    # 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
-
-# Launch the app using asyncio
+                analyze_output = gr.JSON(label="Website Content Analysis", interactive=False)
+                predict_output = gr.JSON(label="Website Traffic Prediction", interactive=False)
+
+        # --- Button Actions ---
+        start_button.click(
+            fn=on_start_click,
+            inputs=[target_urls, feed_rss_checkbox],  # Removed database settings
+            outputs=[status_text],
+        )
+
+        stop_button.click(fn=on_stop_click, outputs=[status_text])
+
+        view_button.click(
+            fn=on_view_feed_click,
+            inputs=[],  # Removed feed_url input (now from settings)
+            outputs=[feed_content],
+        )
+
+        scrape_button.click(
+            fn=on_scrape_click,
+            inputs=[target_urls],
+            outputs=[scrape_output],
+        )
+
+        analyze_button.click(
+            fn=on_analyze_click,
+            inputs=[scrape_output],
+            outputs=[analyze_output],
+        )
+
+        predict_button.click(
+            fn=on_predict_click,
+            inputs=[target_urls],
+            outputs=[predict_output],
+        )
+
+        send_button.click(
+            fn=chatbot_response,
+            inputs=[message_input, chatbot_interface],
+            outputs=[chatbot_interface, message_input],
+        )
+
+        # --- Periodic Updates ---
+        feed_updater = gr.Timer(interval=300)  # Update every 5 minutes
+        feed_updater.tick(fn=update_feed_content, outputs=[feed_content])
+
+        # --- Load Database Status ---
+        demo.load(fn=update_db_status, outputs=[db_status_textbox])
+
+        # --- Launch Gradio ---
+        await demo.launch()
+
+
+# --- Helper Functions ---
+async def on_start_click(target_urls_str: str, feed_enabled: bool):
+    global monitoring_task
+    urls = [url.strip() for url in target_urls_str.split(",")]
+    await set_db_connection()
+    monitoring_task = asyncio.create_task(start_monitoring(urls, settings.STORAGE_LOCATION, feed_enabled))
+    return "Monitoring started."
+
+async def on_stop_click():
+    global monitoring_task
+    if monitoring_task:
+        monitoring_task.cancel()
+        monitoring_task = None
+    return "Monitoring stopped."
+
+async def on_view_feed_click():
+    return await fetch_feed_content(settings.RSS_FEED_URL)
+
+async def on_scrape_click(url: str):
+    return await scrape_website(url)
+
+async def on_analyze_click(content: str):
+    return await analyze_website_content(content)
+
+async def on_predict_click(url: str):
+    return await predict_website_traffic(url)
+
+# --- Main Execution ---
 if __name__ == "__main__":
     asyncio.run(main())