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