Update app.py

app.py

@@ -1,356 +1,212 @@
-import datetime
 import os
+import asyncio
 import csv
-import time
-import hashlib
-import threading
-from pathlib import Path
 import logging
 from typing import List, Tuple
-
+from dotenv import load_dotenv
+import aiohttp
 import gradio as gr
-from selenium import webdriver
-from selenium.webdriver.chrome.service import Service
-from selenium.webdriver.chrome.options import Options
-from selenium.webdriver.common.by import By
-from selenium.common.exceptions import (
-    WebDriverException,
-    NoSuchElementException,
-    TimeoutException,
-    StaleElementReferenceException,
-)
-from webdriver_manager.chrome import ChromeDriverManager
+from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.exc import SQLAlchemyError
+from bs4 import BeautifulSoup
+import hashlib
+import datetime
+import feedparser
 from huggingface_hub import InferenceClient
-import mysql.connector
-import feedparser  # For parsing RSS feeds
-import sqlite3  # For simple local storage if needed
+import validators
 
-# Configure logging
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(filename)s - %(lineno)d - %(message)s')
+# Load environment variables
+load_dotenv()
 
-# Configuration (replace with your actual values or environment variables)
-DB_HOST = os.environ.get("DB_HOST", "your_host")
-DB_USER = os.environ.get("DB_USER", "your_user")
-DB_PASSWORD = os.environ.get("DB_PASSWORD", "your_password")
-DB_NAME = os.environ.get("DB_NAME", "your_database")
-HUGGINGFACE_API_KEY = os.environ.get("HUGGINGFACE_API_KEY")  # Add API key
-DEFAULT_MONITORING_INTERVAL = 300  # 5 minutes in seconds
-MAX_MONITORING_INTERVAL = 600  # 10 minutes in seconds
-CHANGE_FREQUENCY_THRESHOLD = 3  # Number of changes to trigger faster monitoring
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# Configuration
+DB_URL = os.getenv('DB_URL', 'sqlite:///monitoring.db')
+HUGGINGFACE_API_KEY = os.getenv('HUGGINGFACE_API_KEY')
+DEFAULT_MONITORING_INTERVAL = 300
+MAX_MONITORING_INTERVAL = 600
+CHANGE_FREQUENCY_THRESHOLD = 3
+
+# Database setup
+Base = declarative_base()
+
+class Article(Base):
+    __tablename__ = 'articles'
+    id = Column(Integer, primary_key=True)
+    url = Column(String(255), nullable=False)
+    title = Column(String(255))
+    content = Column(Text)
+    hash = Column(String(32))
+    timestamp = Column(DateTime, default=datetime.datetime.utcnow)
+
+engine = create_engine(DB_URL)
+Base.metadata.create_all(engine)
+Session = sessionmaker(bind=engine)
 
 # Global variables
-monitoring_thread = None
-stop_event = threading.Event()
-db_connection = None
-current_task = None
+monitoring_tasks = {}
+url_monitoring_intervals = {}
+change_counts = {}
 history = []
-url_monitoring_intervals = {}  # Store monitoring intervals for each URL
-change_counts = {}  # Track change frequency for each URL
-
-def get_db_connection():
-    global db_connection
-    if db_connection is None or not db_connection.is_connected():
-        try:
-            db_connection = mysql.connector.connect(
-                host=DB_HOST,
-                user=DB_USER,
-                password=DB_PASSWORD,
-            )  # Connect to the server first
-
-            # Select the database
-            if DB_NAME:
-                cursor = db_connection.cursor()
-                cursor.execute(f"USE {DB_NAME}") 
-                cursor.close()
-            else:
-                logging.error("Database name is not set.")
-                return None
 
-            return db_connection
-        except Exception as e:
-            logging.error(f"Error connecting to database: {e}")
-            return None
-    else:
-        return db_connection
+# Utility functions
+def sanitize_url(url: str) -> str:
+    return validators.url(url)
 
-# Function to create the articles table if it doesn't exist
-def create_articles_table():
-    conn = get_db_connection()
-    if conn:
-        cursor = conn.cursor()
-        cursor.execute("""
-            CREATE TABLE IF NOT EXISTS articles (
-                id INT AUTO_INCREMENT PRIMARY KEY,
-                url VARCHAR(255) NOT NULL,
-                title VARCHAR(255),
-                content TEXT,
-                hash VARCHAR(32),
-                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
-            )
-        """)
-        conn.commit()
-        cursor.close()
+async def fetch_url_content(url: str, session: aiohttp.ClientSession) -> Tuple[str, str]:
+    async with session.get(url) as response:
+        content = await response.text()
+        soup = BeautifulSoup(content, 'html.parser')
+        title = soup.title.string if soup.title else "No Title"
+        return title, content
 
-# Initialize the articles table
-create_articles_table()
-
-# Function to monitor URLs for changes
-def monitor_urls(target_urls: List[str], storage_location: str, feed_rss: bool, stop_event: threading.Event):
-    global history, url_monitoring_intervals, change_counts
-    previous_hashes = {url: "" for url in target_urls}
-    options = Options()
-    options.headless = True
-    options.add_argument("--disable-gpu")
-    options.add_argument("--no-sandbox")
-    options.add_argument("--disable-dev-shm-usage")
-    options.add_experimental_option("excludeSwitches", ["enable-logging"])  # Suppress unnecessary logs
-
-    driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
-    driver.implicitly_wait(10)  # Implicit wait for elements
+def calculate_hash(content: str) -> str:
+    return hashlib.md5(content.encode('utf-8')).hexdigest()
 
+async def save_to_database(url: str, title: str, content: str, hash: str):
+    session = Session()
     try:
-        while not stop_event.is_set():
-            for url in target_urls:
-                try:
-                    # Dynamic monitoring interval
-                    interval = url_monitoring_intervals.get(url, DEFAULT_MONITORING_INTERVAL)
-
-                    driver.get(url)
-                    time.sleep(2)  # Allow page to load
-
-                    # Check for changes
-                    try:
-                        current_content = driver.find_element(By.TAG_NAME, "body").get_attribute("innerHTML")
-                        current_hash = hashlib.md5(current_content.encode('utf-8')).hexdigest()
-                    except (NoSuchElementException, TimeoutException, StaleElementReferenceException) as e:
-                        logging.warning(f"Error getting content for {url}: {e}")
-                        continue
-
-                    if current_hash != previous_hashes[url]:
-                        previous_hashes[url] = current_hash
-                        timestamp = datetime.datetime.now()
-
-                        try:
-                            title_element = driver.find_element(By.TAG_NAME, "title")
-                            title = title_element.text
-                        except NoSuchElementException:
-                            title = "No Title"
-
-                        history.append(f"Change detected at {url} on {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
-
-                        if storage_location:
-                            save_to_storage(storage_location, url, title, current_content, timestamp)
-
-                        if feed_rss:
-                            save_to_database(url, title, current_content, current_hash)
-
-                        # Adjust monitoring interval based on change frequency
-                        change_counts[url] = change_counts.get(url, 0) + 1
-                        if change_counts[url] >= CHANGE_FREQUENCY_THRESHOLD:
-                            url_monitoring_intervals[url] = 60  # Check more frequently after multiple changes
-                        else:
-                            url_monitoring_intervals[url] = min(interval + 60, MAX_MONITORING_INTERVAL)  # Gradually increase interval
-
-                    else:
-                        # Increase interval if no changes detected
-                        change_counts[url] = 0  # Reset change count if no change
-                        url_monitoring_intervals[url] = min(interval + 60, MAX_MONITORING_INTERVAL)
-
-                except WebDriverException as e:
-                    logging.error(f"Error accessing {url}: {e}")
-
-                if stop_event.is_set():
-                    break  # Exit inner loop if stop event is set
-
-            if not stop_event.is_set():
-                time.sleep(interval)
-
-    except Exception as e:
-        logging.error(f"Unexpected error in monitoring thread: {e}")
+        article = Article(url=url, title=title, content=content, hash=hash)
+        session.add(article)
+        session.commit()
+    except SQLAlchemyError as e:
+        logger.error(f"Database error: {e}")
+        session.rollback()
     finally:
-        driver.quit()
-        logging.info("Monitoring thread has been stopped.")
+        session.close()
 
-# Function to save data to local storage (CSV)
-def save_to_storage(storage_location: str, url: str, title: str, content: str, timestamp: datetime.datetime):
+def save_to_csv(storage_location: str, url: str, title: str, content: str, timestamp: datetime.datetime):
     try:
         with open(storage_location, "a", newline='', encoding='utf-8') as csvfile:
             csv_writer = csv.writer(csvfile)
             csv_writer.writerow([timestamp.strftime("%Y-%m-%d %H:%M:%S"), url, title, content])
     except Exception as e:
-        logging.error(f"Error saving to storage: {e}")
+        logger.error(f"Error saving to CSV: {e}")
+
+async def monitor_url(url: str, interval: int, storage_location: str, feed_rss: bool):
+    previous_hash = ""
+    async with aiohttp.ClientSession() as session:
+        while True:
+            try:
+                title, content = await fetch_url_content(url, session)
+                current_hash = calculate_hash(content)
+
+                if current_hash != previous_hash:
+                    previous_hash = current_hash
+                    timestamp = datetime.datetime.now()
+                    
+                    if feed_rss:
+                        await save_to_database(url, title, content, current_hash)
+                    
+                    if storage_location:
+                        save_to_csv(storage_location, url, title, content, timestamp)
+                    
+                    history.append(f"Change detected at {url} on {timestamp.strftime('%Y-%m-%d %H:%M:%S')}")
+                    logger.info(f"Change detected at {url}")
+
+                    change_counts[url] = change_counts.get(url, 0) + 1
+                    if change_counts[url] >= CHANGE_FREQUENCY_THRESHOLD:
+                        interval = max(60, interval // 2)
+                else:
+                    change_counts[url] = 0
+                    interval = min(interval * 2, MAX_MONITORING_INTERVAL)
+
+                url_monitoring_intervals[url] = interval
+            except Exception as e:
+                logger.error(f"Error monitoring {url}: {e}")
+                history.append(f"Error monitoring {url}: {e}")
+
+            await asyncio.sleep(interval)
+
+async def start_monitoring(urls: List[str], storage_location: str, feed_rss: bool):
+    for url in urls:
+        if url not in monitoring_tasks:
+            sanitized_url = sanitize_url(url)
+            if sanitized_url:
+                task = asyncio.create_task(monitor_url(sanitized_url, DEFAULT_MONITORING_INTERVAL, storage_location, feed_rss))
+                monitoring_tasks[sanitized_url] = task
+            else:
+                logger.warning(f"Invalid URL: {url}")
+                history.append(f"Invalid URL: {url}")
 
-# Function to save data to the database
-def save_to_database(url: str, title: str, content: str, hash: str):
-    conn = get_db_connection()
-    if conn:
-        cursor = conn.cursor()
-        try:
-            sql = "INSERT INTO articles (url, title, content, hash) VALUES (%s, %s, %s, %s)"
-            val = (url, title, content, hash)
-            cursor.execute(sql, val)
-            conn.commit()
-        except Exception as e:
-            logging.error(f"Error saving to database: {e}")
-        finally:
-            cursor.close()
+def stop_monitoring(url: str):
+    if url in monitoring_tasks:
+        monitoring_tasks[url].cancel()
+        del monitoring_tasks[url]
 
-# Function to generate RSS feed from the database
 def generate_rss_feed():
-    conn = get_db_connection()
-    if conn:
-        cursor = conn.cursor()
-        try:
-            cursor.execute("SELECT * FROM articles ORDER BY timestamp DESC")
-            articles = cursor.fetchall()
-
-            feed = feedparser.FeedParserDict()
-            feed['title'] = 'Website Changes Feed'
-            feed['link'] = 'http://yourwebsite.com/feed'  # Replace with your actual feed URL
-            feed['description'] = 'Feed of changes detected on monitored websites.'
-            feed['entries'] = []
-
-            for article in articles:
-                entry = feedparser.FeedParserDict()
-                entry['title'] = article[2]  # Title
-                entry['link'] = article[1]   # URL
-                entry['description'] = article[3]  # Content
-                entry['published'] = article[5]  # Timestamp
-                feed['entries'].append(entry)
-
-            return feedparser.FeedGenerator().feed_from_dictionary(feed).writeString('utf-8')
-        except Exception as e:
-            logging.error(f"Error generating RSS feed: {e}")
-        finally:
-            cursor.close()
-    return None
-
-# Function to start monitoring
-def start_monitoring(target_urls: List[str], storage_location: str, feed_rss: bool):
-    global monitoring_thread, stop_event, current_task, history, change_counts
-    if monitoring_thread and monitoring_thread.is_alive():
-        return "Monitoring is already running.", history
-
-    stop_event.clear()
-    current_task = f"Monitoring URLs: {', '.join(target_urls)}"
-    history.append(f"Task started: {current_task}")
-    change_counts = {url: 0 for url in target_urls}  # Reset change counts
-    monitoring_thread = threading.Thread(
-        target=monitor_urls,
-        args=(target_urls, storage_location, feed_rss, stop_event),
-        daemon=True
-    )
-    monitoring_thread.start()
-    return "Monitoring started.", history
-
-# Function to stop monitoring
-def stop_monitoring():
-    global monitoring_thread, stop_event, current_task, history
-    if monitoring_thread and monitoring_thread.is_alive():
-        stop_event.set()
-        monitoring_thread.join()
-        current_task = None
-        history.append("Monitoring stopped by user.")
-        return "Monitoring stopped.", history
-    else:
-        return "No monitoring task is currently running.", history
+    session = Session()
+    try:
+        articles = session.query(Article).order_by(Article.timestamp.desc()).limit(20).all()
+        feed = feedparser.FeedParserDict()
+        feed['title'] = 'Website Changes Feed'
+        feed['link'] = 'http://yourwebsite.com/feed'
+        feed['description'] = 'Feed of changes detected on monitored websites.'
+        feed['entries'] = [
+            {'title': article.title, 'link': article.url, 'description': article.content, 'published': article.timestamp}
+            for article in articles
+        ]
+        return feedparser.FeedGenerator().feed_from_dictionary(feed).writeString('utf-8')
+    except SQLAlchemyError as e:
+        logger.error(f"Database error: {e}")
+        return None
+    finally:
+        session.close()
 
-# Function to handle chatbot responses
-def chatbot_response(message: str, history: List[Tuple[str, str]]):
+async def chatbot_response(message: str, history: List[Tuple[str, str]]):
     try:
         client = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1", token=HUGGINGFACE_API_KEY)
-        response = client.inference(message)
+        response = await client.inference(message)
         history.append((message, response))
         return history, history
     except Exception as e:
-        logging.error(f"Error getting chatbot response: {e}")
+        logger.error(f"Chatbot error: {e}")
         history.append((message, "Error: Could not get a response from the chatbot."))
         return history, history
 
-# --- Gradio Interface ---
+# Gradio interface
 with gr.Blocks() as demo:
     gr.Markdown("# Website Monitor and Chatbot")
 
-    # Configuration Tab
     with gr.Tab("Configuration"):
-        with gr.Row():
-            target_urls = gr.Textbox(
-                label="Target URLs (comma-separated)",
-                placeholder="https://example.com, https://another-site.com"
-            )
-        with gr.Row():
-            storage_location = gr.Textbox(
-                label="Storage Location (CSV file path)",
-                placeholder="/path/to/your/file.csv",
-                visible=False  # You can enable this if you want CSV storage
-            )
-        with gr.Row():
-            feed_rss_checkbox = gr.Checkbox(label="Enable RSS Feed")
-        with gr.Row():
-            start_button = gr.Button("Start Monitoring")
-            stop_button = gr.Button("Stop Monitoring")
-        with gr.Row():
-            status_text = gr.Textbox(label="Status", interactive=False)
-        with gr.Row():
-            history_text = gr.Textbox(
-                label="History", lines=10, interactive=False
-            )
+        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")
+        status_text = gr.Textbox(label="Status", interactive=False)
+        history_text = gr.Textbox(label="History", lines=10, interactive=False)
 
-    # User-End View Tab
     with gr.Tab("User-End View"):
-        with gr.Row():
-            feed_content = gr.JSON(label="RSS Feed Content")
+        feed_content = gr.JSON(label="RSS Feed Content")
 
-    # Chatbot Tab
     with gr.Tab("Chatbot"):
-        chatbot_interface = gr.Chatbot(type='messages') 
-        with gr.Row():
-            message_input = gr.Textbox(placeholder="Type your message here...")
-            send_button = gr.Button("Send")
-
-    # --- Event Handlers ---
-
-    # Start monitoring button click
-    def on_start_click(target_urls_str: str, storage_loc: str, feed_enabled: bool):
-        global history, url_monitoring_intervals
-        try:
-            target_urls = [url.strip() for url in target_urls_str.split(",")]
-            if not all(target_urls):
-                return "Please enter valid URLs.", history
-
-            # Reset monitoring intervals when starting
-            url_monitoring_intervals = {url: DEFAULT_MONITORING_INTERVAL for url in target_urls}
-
-            status, history = start_monitoring(target_urls, storage_loc if storage_loc else None, feed_enabled)
-            return status, history
-        except Exception as e:
-            return f"Error starting monitoring: {e}", history
+        chatbot_interface = gr.Chatbot()
+        message_input = gr.Textbox(placeholder="Type your message here...")
+        send_button = gr.Button("Send")
 
-    start_button.click(
-        on_start_click,
-        inputs=[target_urls, storage_location, feed_rss_checkbox],
-        outputs=[status_text, history_text]
-    )
+    async def on_start_click(target_urls_str: str, storage_loc: str, feed_enabled: bool):
+        urls = [url.strip() for url in target_urls_str.split(",")]
+        await start_monitoring(urls, storage_loc if storage_loc else None, feed_enabled)
+        return "Monitoring started for valid URLs."
 
-    # Stop monitoring button click
-    stop_button.click(
-        stop_monitoring,
-        outputs=[status_text, history_text]
-    )
+    async def on_stop_click():
+        for url in list(monitoring_tasks.keys()):
+            stop_monitoring(url)
+        return "Monitoring stopped for all URLs."
 
-    # Send message to chatbot button click
-    send_button.click(
-        chatbot_response,
-        inputs=[message_input, chatbot_interface],
-        outputs=[chatbot_interface, chatbot_interface]
-    )
+    start_button.click(on_start_click, inputs=[target_urls, storage_location, feed_rss_checkbox], outputs=[status_text])
+    stop_button.click(on_stop_click, outputs=[status_text])
+    send_button.click(chatbot_response, inputs=[message_input, chatbot_interface], outputs=[chatbot_interface, chatbot_interface])
 
-    # Update RSS feed content periodically
-    def update_feed_content():
+    async def update_feed_content():
         return generate_rss_feed()
 
-    # Use gr.Timer to periodically update the RSS feed content
-    gr.Timer(update_feed_content, every=5).start()
+    feed_updater = gr.Timer(update_feed_content, every=5)
 
 if __name__ == "__main__":
     demo.launch()