app2.py

import asyncio
import gradio as gr
import logging
import os
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()

class Settings(BaseSettings):
    # Database settings
    DB_HOST: str
    DB_USER: str
    DB_PASSWORD: str
    DB_NAME: str
    DB_TYPE: str = "mysql"  # Default value
    DB_PORT: int = 3306    # Default value
    
    # API and storage settings
    CHATBOT_API_KEY: str
    RSS_FEED_URL: str
    STORAGE_LOCATION: str = "data"  # Default value
    
    class Config:
        env_file = ".env"
        case_sensitive = False

# --- Database Model ---
class Article:
    def __init__(self, title, url, content, timestamp):
        self.title = title
        self.url = url
        self.content = content
        self.timestamp = timestamp

# --- Global Variables ---
settings = Settings()
db_session = None
engine = None
monitoring_task = None
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# --- Database Connection ---
def get_db_url() -> str:
    if settings.DB_TYPE == "mysql":
        return f"mysql+aiomysql://{settings.DB_USER}:{settings.DB_PASSWORD}@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
    elif settings.DB_TYPE == "postgresql":
        return f"postgresql+asyncpg://{settings.DB_USER}:{settings.DB_PASSWORD}@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
    else:
        return "sqlite+aiosqlite:///default.db"

async def set_db_connection():
    global db_session, engine
    try:
        db_url = get_db_url()
        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}"

# --- 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:
        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:
        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 {}

# --- 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.")
        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():
    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}"
    return "Database connection not established."

# --- Gradio UI ---
async def main():
    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")
                scrape_output = gr.Textbox(label="Scraped Website Content", interactive=False)
                analyze_output = gr.JSON(label="Website Content Analysis")

        # --- Button Actions ---
        async def on_start_click(target_urls_str: str, feed_enabled: bool):
            urls = [url.strip() for url in target_urls_str.split(",")]
            await set_db_connection()
            global monitoring_task
            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."

        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=update_feed_content, outputs=[feed_content])
        scrape_button.click(fn=scrape_website, inputs=[target_urls], outputs=[scrape_output])
        analyze_button.click(fn=analyze_website_content, inputs=[scrape_output], outputs=[analyze_output])
        send_button.click(
            fn=chatbot_response,
            inputs=[message_input, chatbot_interface],
            outputs=[chatbot_interface, message_input],
        )

        # --- Periodic Updates ---
        demo.load(fn=update_db_status, outputs=[db_status_textbox])

        # --- Launch Gradio ---
        await demo.launch()

if __name__ == "__main__":
    asyncio.run(main())