app2.py

import asyncio
import gradio as gr
import logging
import os
import sys
import requests
from bs4 import BeautifulSoup
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_settings import BaseSettings

# --- Configuration ---
load_dotenv()  # Load environment variables from .env file

class Settings(BaseSettings):
    # Define the database fields explicitly
    db_host: str
    db_user: str
    db_password: str
    db_name: str
    database_type: str
    database_port: int
    chatbot_api_key: str
    rss_feed_url: str
    storage_location: str
    
    class Config:
        # Optional: if you want to allow extra fields
        extra = "allow"
        
settings = Settings()

# --- Database Connection ---
def get_db_url(settings: Settings) -> str:
    if settings.database_type == "mysql":
        return f"mysql+aiomysql://{settings.db_user}:{settings.db_password}@{settings.db_host}:{settings.database_port}/{settings.db_name}"
    elif settings.database_type == "postgresql":
        return f"postgresql+asyncpg://{settings.db_user}:{settings.db_password}@{settings.db_host}:{settings.database_port}/{settings.db_name}"
    else:
        return "sqlite+aiosqlite:///default.db"

async def set_db_connection(
    db_type: str = None,
    db_host: str = None,
    db_port: int = None,
    db_user: str = None,
    db_password: str = None,
    db_name: str = None
):
    global db_session, engine, settings
    try:
        # Update settings if new values provided
        if db_type:
            settings.database_type = db_type
        if db_host:
            settings.db_host = db_host
        if db_port:
            settings.database_port = db_port
        if db_user:
            settings.db_user = db_user
        if db_password:
            settings.db_password = db_password
        if db_name:
            settings.db_name = db_name

        # Close existing connection if any
        if db_session:
            await db_session.close()
        if engine:
            await engine.dispose()

        # Create new connection
        db_url = get_db_url(settings)
        engine = create_async_engine(db_url, 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}"

# --- 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)

# --- Database Connection ---
async def set_db_connection():
    global db_session, engine
    try:
        engine = create_async_engine(
            f"mysql+aiomysql://{settings.db_user}:{settings.db_password}@{settings.db_host}:{settings.database_port}/{settings.db_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}"

# --- Website Monitoring ---
async def start_monitoring(urls: List[str], storage_loc: str, feed_enabled: bool):
    while True:
        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

async def store_data(url: str, content: str, analysis: Dict[str, Any]):
    try:
        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 storing data: {e}")

async def update_feed_content():
    try:
        # Fetch RSS feed content from database
        async with db_session as session:
            articles = await session.execute(select(Article))
            feed_content = []
            for article in articles.scalars():
                feed_content.append({
                    "title": article.title,
                    "url": article.url,
                    "content": article.content,
                    "timestamp": article.timestamp,
                })
            return feed_content
    except Exception as e:
        logger.error(f"Error updating feed content: {e}")
        return []

# --- Website Scraping and Analysis ---
async def scrape_website(url: str) -> str:
    try:
        response = requests.get(url)
        response.raise_for_status()
        soup = BeautifulSoup(response.text, "html.parser")
        return soup.get_text()
    except requests.exceptions.RequestException as e:
        logger.error(f"Error scraping website {url}: {e}")
        return ""

async def analyze_website_content(content: str) -> Dict[str, Any]:
    try:
        # Perform sentiment analysis or other analysis
        sentiment = "Positive" if content.count("good") > content.count("bad") else "Negative"
        return {"sentiment": sentiment}
    except Exception as e:
        logger.error(f"Error analyzing website content: {e}")
        return {}

# --- Website Traffic Prediction ---
async def predict_website_traffic(url: str) -> Dict[str, Any]:
    try:
        # ... (Your machine learning model for traffic prediction) ...
        return {"traffic": 100}  # Placeholder
    except Exception as e:
        logger.error(f"Error predicting website traffic: {e}")
        return {}

# --- Chatbot Integration ---
async def chatbot_response(message: str, chat_history: List[Dict[str, str]]) -> List[Dict[str, str]]:
    try:
        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, ""

# --- 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
    demo = gr.Blocks()

    with demo:
        gr.Markdown("# Website Monitor and Chatbot")
        with gr.Row():
            with gr.Column():
                gr.Markdown("## Database Settings")
                db_status_textbox = gr.Textbox(label="Database Status", interactive=False)
                status_text = gr.Textbox(label="Status", interactive=False)

                gr.Markdown("## RSS Feed Reader Settings")
                view_button = gr.Button("View Feed")
                target_urls = gr.Textbox(label="Target URLs (comma-separated)", placeholder="https://example.com, https://another-site.com")
                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")
                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")
                predict_output = gr.JSON(label="Website Traffic Prediction")

        # --- Button Actions ---
        start_button.click(
            fn=on_start_click,
            inputs=[target_urls, feed_rss_checkbox],
            outputs=[status_text],
        )

        stop_button.click(fn=on_stop_click, outputs=[status_text])

        view_button.click(
            fn=on_view_feed_click,
            inputs=[],
            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 ---
        async def update_feed_periodically(feed_content):
            while True:
                await update_feed_content()
                await asyncio.sleep(300)  # Check every 5 minutes

        feed_updater = asyncio.create_task(update_feed_periodically(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 update_feed_content()

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())