NLPulse.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyNK828JLowutnpc+6nlx8EF",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/abhy-kumar/NLPulse/blob/main/NLPulse.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install feedparser TextBlob\n",
        "!pip install nltk transformers torch\n",
        "#Cell 1"
      ],
      "metadata": {
        "id": "0RN8JfIeH5pl"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Cell 2\n",
        "\n",
        "import nltk\n",
        "import torch\n",
        "import requests\n",
        "import feedparser\n",
        "import sqlite3\n",
        "import pandas as pd\n",
        "import plotly.graph_objects as go\n",
        "import numpy as np\n",
        "from plotly.subplots import make_subplots\n",
        "from datetime import datetime, timedelta\n",
        "from nltk.sentiment import SentimentIntensityAnalyzer\n",
        "from transformers import (\n",
        "    AutoTokenizer,\n",
        "    AutoModelForSequenceClassification,\n",
        "    pipeline\n",
        ")\n",
        "from concurrent.futures import ThreadPoolExecutor\n",
        "from functools import lru_cache\n",
        "import logging\n",
        "from typing import List, Tuple, Set\n",
        "from wordcloud import WordCloud\n",
        "import seaborn as sns\n",
        "from collections import Counter\n",
        "import matplotlib.pyplot as plt\n",
        "from nltk.tokenize import word_tokenize\n",
        "from nltk.corpus import stopwords\n",
        "import warnings\n",
        "from difflib import SequenceMatcher\n",
        "\n",
        "warnings.filterwarnings('ignore')\n",
        "\n",
        "# Configure logging\n",
        "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n",
        "\n",
        "# Constants\n",
        "DATABASE_NAME = 'news_sentiment.db'\n",
        "NEWS_FEED_URL = \"https://www.thehindu.com/feeder/default.rss\"\n",
        "DATE_FORMAT = \"%Y-%m-%d\"\n",
        "DATETIME_FORMAT = \"%a, %d %b %Y %H:%M:%S %z\"\n",
        "\n",
        "class DatabaseManager:\n",
        "    \"\"\"\n",
        "    Manages the SQLite database operations for storing and retrieving sentiment scores.\n",
        "    It handles the creation of the necessary tables, insertion of new data,\n",
        "    and various queries to retrieve statistics and specific datasets.\n",
        "    \"\"\"\n",
        "    def __init__(self, db_name: str):\n",
        "        self.db_name = db_name\n",
        "        self._setup_database()\n",
        "        self.connection = None\n",
        "        self._setup_connection()\n",
        "\n",
        "    def _setup_database(self):\n",
        "        \"\"\"\n",
        "        Initializes the database by creating the sentiment_scores table if it does not exist.\n",
        "        Also creates indexes on date, title, and score columns to optimize query performance.\n",
        "        \"\"\"\n",
        "        with sqlite3.connect(self.db_name) as conn:\n",
        "            conn.execute('''\n",
        "                CREATE TABLE IF NOT EXISTS sentiment_scores (\n",
        "                    date TEXT,\n",
        "                    time TEXT,\n",
        "                    title TEXT,\n",
        "                    summary TEXT,\n",
        "                    score REAL\n",
        "                )\n",
        "            ''')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_date ON sentiment_scores(date)')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_title ON sentiment_scores(title)')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_score ON sentiment_scores(score)')\n",
        "\n",
        "    def _setup_connection(self):\n",
        "        \"\"\"\n",
        "        Establishes a persistent connection to the SQLite database with optimized parameters\n",
        "        for performance. Configures the database to use Write-Ahead Logging and sets cache sizes.\n",
        "        \"\"\"\n",
        "        if not self.connection:\n",
        "            self.connection = sqlite3.connect(self.db_name, check_same_thread=False)\n",
        "            self.connection.execute('PRAGMA journal_mode=WAL')  # Write-Ahead Logging\n",
        "            self.connection.execute('PRAGMA synchronous=NORMAL')\n",
        "            self.connection.execute('PRAGMA cache_size=-2000')  # 2MB cache\n",
        "            self.connection.execute('PRAGMA temp_store=MEMORY')\n",
        "\n",
        "    def store_score(self, date: str, time: str, title: str, summary: str, score: float):\n",
        "        \"\"\"\n",
        "        Inserts a new sentiment score entry into the sentiment_scores table.\n",
        "\n",
        "        Parameters:\n",
        "            date (str): The date of the news article.\n",
        "            time (str): The time of publication.\n",
        "            title (str): The headline of the news article.\n",
        "            summary (str): The summary of the news article.\n",
        "            score (float): The sentiment score associated with the article.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            self.connection.execute(\n",
        "                \"INSERT INTO sentiment_scores VALUES (?, ?, ?, ?, ?)\",\n",
        "                (date, time, title, summary, score)\n",
        "            )\n",
        "            self.connection.commit()\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error storing score: {e}\")\n",
        "\n",
        "    @lru_cache(maxsize=128)\n",
        "    def get_daily_average(self, date: str) -> float:\n",
        "        \"\"\"\n",
        "        Retrieves the average sentiment score for a specific date.\n",
        "\n",
        "        Parameters:\n",
        "            date (str): The date for which to calculate the average sentiment.\n",
        "\n",
        "        Returns:\n",
        "            float: The average sentiment score, or 0.0 if no entries are found.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            result = self.connection.execute(\n",
        "                \"SELECT AVG(score) FROM sentiment_scores WHERE date = ?\",\n",
        "                (date,)\n",
        "            ).fetchone()\n",
        "            return result[0] if result[0] is not None else 0.0\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error getting daily average: {e}\")\n",
        "            return 0.0\n",
        "\n",
        "    def get_headlines_with_scores(self, limit: int = 100, min_summary_length: int = 20) -> pd.DataFrame:\n",
        "        \"\"\"\n",
        "        Retrieves recent headlines along with their sentiment scores,\n",
        "        excluding entries with summaries shorter than the specified length.\n",
        "\n",
        "        Parameters:\n",
        "            limit (int): The maximum number of headlines to retrieve.\n",
        "            min_summary_length (int): The minimum length of the summary to include an entry.\n",
        "\n",
        "        Returns:\n",
        "            pd.DataFrame: A DataFrame containing the selected headlines and their scores.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            query = \"\"\"\n",
        "                SELECT date, time, title, score\n",
        "                FROM sentiment_scores\n",
        "                WHERE LENGTH(summary) >= ?\n",
        "                ORDER BY date DESC, time DESC\n",
        "                LIMIT ?\n",
        "            \"\"\"\n",
        "            return pd.read_sql_query(query, self.connection, params=(min_summary_length, limit))\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error retrieving headlines: {e}\")\n",
        "            return pd.DataFrame()\n",
        "\n",
        "    def get_extreme_sentiment_headlines(self) -> Tuple[pd.DataFrame, pd.DataFrame]:\n",
        "        \"\"\"\n",
        "        Retrieves the top five most positive and top five most negative headlines based on sentiment scores.\n",
        "\n",
        "        Returns:\n",
        "            Tuple[pd.DataFrame, pd.DataFrame]: Two DataFrames containing the most positive and most negative headlines.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            most_positive = pd.read_sql_query(\"\"\"\n",
        "                SELECT date, time, title, score\n",
        "                FROM sentiment_scores\n",
        "                ORDER BY score DESC\n",
        "                LIMIT 5\n",
        "            \"\"\", self.connection)\n",
        "\n",
        "            most_negative = pd.read_sql_query(\"\"\"\n",
        "                SELECT date, time, title, score\n",
        "                FROM sentiment_scores\n",
        "                ORDER BY score ASC\n",
        "                LIMIT 5\n",
        "            \"\"\", self.connection)\n",
        "\n",
        "            return most_positive, most_negative\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error retrieving extreme headlines: {e}\")\n",
        "            return pd.DataFrame(), pd.DataFrame()\n",
        "\n",
        "    def get_headline_stats(self) -> dict:\n",
        "        \"\"\"\n",
        "        Gathers statistical information about the headlines stored in the database,\n",
        "        including average sentiment, total number of headlines, and the date range covered.\n",
        "\n",
        "        Returns:\n",
        "            dict: A dictionary containing the statistical metrics.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            stats = {}\n",
        "\n",
        "            avg_query = \"SELECT AVG(score) FROM sentiment_scores\"\n",
        "            stats['average_sentiment'] = self.connection.execute(avg_query).fetchone()[0]\n",
        "\n",
        "            count_query = \"SELECT COUNT(*) FROM sentiment_scores\"\n",
        "            stats['total_headlines'] = self.connection.execute(count_query).fetchone()[0]\n",
        "\n",
        "            range_query = \"\"\"\n",
        "                SELECT\n",
        "                    MIN(date) as earliest_date,\n",
        "                    MAX(date) as latest_date\n",
        "                FROM sentiment_scores\n",
        "            \"\"\"\n",
        "            earliest, latest = self.connection.execute(range_query).fetchone()\n",
        "            stats['date_range'] = f\"{earliest} to {latest}\"\n",
        "\n",
        "            return stats\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error getting headline stats: {e}\")\n",
        "            return {}\n",
        "\n",
        "    def close(self):\n",
        "        \"\"\"\n",
        "        Closes the database connection if it is open.\n",
        "        \"\"\"\n",
        "        if self.connection:\n",
        "            self.connection.close()\n",
        "\n",
        "class SentimentAnalyzer:\n",
        "    \"\"\"\n",
        "    Performs sentiment analysis on text data using multiple NLP models.\n",
        "    Combines results from VADER, FinBERT, and RoBERTa models to produce a comprehensive sentiment score.\n",
        "    \"\"\"\n",
        "    def __init__(self):\n",
        "        self._initialize_models()\n",
        "        self._setup_device()\n",
        "\n",
        "    def _initialize_models(self):\n",
        "        \"\"\"\n",
        "        Initializes the necessary NLP models and tokenizers.\n",
        "        Downloads required NLTK data if not already present.\n",
        "        \"\"\"\n",
        "        for resource in ['vader_lexicon', 'punkt', 'stopwords']:\n",
        "            try:\n",
        "                nltk.data.find(f'tokenizers/{resource}')\n",
        "            except LookupError:\n",
        "                nltk.download(resource, quiet=True)\n",
        "\n",
        "        self.sia = SentimentIntensityAnalyzer()\n",
        "\n",
        "        self.finbert_tokenizer = AutoTokenizer.from_pretrained(\"ProsusAI/finbert\")\n",
        "        self.finbert_model = AutoModelForSequenceClassification.from_pretrained(\"ProsusAI/finbert\")\n",
        "        self.roberta_tokenizer = AutoTokenizer.from_pretrained(\"cardiffnlp/twitter-roberta-base-sentiment\")\n",
        "        self.roberta_model = AutoModelForSequenceClassification.from_pretrained(\"cardiffnlp/twitter-roberta-base-sentiment\")\n",
        "        self.summarizer = pipeline(\"summarization\", model=\"facebook/bart-large-cnn\")\n",
        "\n",
        "    def _setup_device(self):\n",
        "        \"\"\"\n",
        "        Configures the computational device (GPU if available, else CPU)\n",
        "        and moves the NLP models to the appropriate device.\n",
        "        \"\"\"\n",
        "        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n",
        "        self.finbert_model.to(self.device)\n",
        "        self.roberta_model.to(self.device)\n",
        "\n",
        "    @lru_cache(maxsize=1024)\n",
        "    def analyze_sentiment(self, text: str) -> float:\n",
        "        \"\"\"\n",
        "        Analyzes the sentiment of the provided text using VADER, FinBERT, and RoBERTa models.\n",
        "        Combines the scores from each model into a weighted average to produce a final sentiment score.\n",
        "\n",
        "        Parameters:\n",
        "            text (str): The text to analyze.\n",
        "\n",
        "        Returns:\n",
        "            float: A sentiment score between 0 and 10.\n",
        "        \"\"\"\n",
        "        if not text or len(text.strip()) < 10:\n",
        "            return 5.0\n",
        "\n",
        "        try:\n",
        "            with ThreadPoolExecutor(max_workers=3) as executor:\n",
        "                vader_future = executor.submit(self._get_vader_score, text)\n",
        "                finbert_future = executor.submit(self._get_finbert_score, text)\n",
        "                roberta_future = executor.submit(self._get_roberta_score, text)\n",
        "\n",
        "                vader_score, vader_confidence = vader_future.result()\n",
        "                finbert_score, finbert_confidence = finbert_future.result()\n",
        "                roberta_score, roberta_confidence = roberta_future.result()\n",
        "\n",
        "            weights = np.array([vader_confidence, finbert_confidence, roberta_confidence])\n",
        "            weights = weights / weights.sum()\n",
        "            scores = np.array([vader_score, finbert_score, roberta_score])\n",
        "\n",
        "            combined_score = np.dot(weights, scores)\n",
        "            return max(0, min(combined_score, 10))\n",
        "\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error in sentiment analysis: {e}\")\n",
        "            return 5.0\n",
        "\n",
        "    def _get_vader_score(self, text: str) -> Tuple[float, float]:\n",
        "        \"\"\"\n",
        "        Computes the sentiment score using the VADER model.\n",
        "\n",
        "        Parameters:\n",
        "            text (str): The text to analyze.\n",
        "\n",
        "        Returns:\n",
        "            Tuple[float, float]: The sentiment score scaled between 0 and 10, and the confidence score.\n",
        "        \"\"\"\n",
        "        scores = self.sia.polarity_scores(text)\n",
        "        return (scores['compound'] + 1) * 5, abs(scores['compound'])\n",
        "\n",
        "    def _get_finbert_score(self, text: str) -> Tuple[float, float]:\n",
        "        \"\"\"\n",
        "        Computes the sentiment score using the FinBERT model.\n",
        "\n",
        "        Parameters:\n",
        "            text (str): The text to analyze.\n",
        "\n",
        "        Returns:\n",
        "            Tuple[float, float]: The sentiment score scaled between 0 and 10, and the confidence score.\n",
        "        \"\"\"\n",
        "        with torch.no_grad():\n",
        "            inputs = self.finbert_tokenizer(text, return_tensors=\"pt\", truncation=True, max_length=512).to(self.device)\n",
        "            outputs = self.finbert_model(**inputs)\n",
        "            probs = torch.nn.functional.softmax(outputs.logits, dim=1)\n",
        "            sentiment = torch.argmax(probs).item()\n",
        "            return sentiment * 5, torch.max(probs).item()\n",
        "\n",
        "    def _get_roberta_score(self, text: str) -> Tuple[float, float]:\n",
        "        \"\"\"\n",
        "        Computes the sentiment score using the RoBERTa model.\n",
        "\n",
        "        Parameters:\n",
        "            text (str): The text to analyze.\n",
        "\n",
        "        Returns:\n",
        "            Tuple[float, float]: The sentiment score scaled between 0 and 10, and the confidence score.\n",
        "        \"\"\"\n",
        "        with torch.no_grad():\n",
        "            inputs = self.roberta_tokenizer(text, return_tensors=\"pt\", truncation=True, max_length=512).to(self.device)\n",
        "            outputs = self.roberta_model(**inputs)\n",
        "            probs = torch.nn.functional.softmax(outputs.logits, dim=1)\n",
        "            sentiment = torch.argmax(probs).item()\n",
        "            return sentiment * 5, torch.max(probs).item()\n",
        "\n",
        "class DataVisualizer:\n",
        "    \"\"\"\n",
        "    Generates various visualizations and dashboards based on the sentiment scores stored in the database.\n",
        "    Creates multiple subplots to display different aspects of the data, such as daily counts, sentiment trends,\n",
        "    distribution of summaries, and more.\n",
        "    \"\"\"\n",
        "    @staticmethod\n",
        "    def create_visualizations(db_name: str):\n",
        "        \"\"\"\n",
        "        Creates an enhanced set of visualizations that provide insights into the sentiment analysis data.\n",
        "        Includes timelines, distributions, heatmaps, and scatter plots to illustrate different metrics.\n",
        "\n",
        "        Parameters:\n",
        "            db_name (str): The name of the SQLite database to retrieve data from.\n",
        "        \"\"\"\n",
        "        with sqlite3.connect(db_name) as conn:\n",
        "            df = pd.read_sql_query(\"SELECT * FROM sentiment_scores\", conn)\n",
        "\n",
        "        if df.empty:\n",
        "            logging.warning(\"No data available for visualization\")\n",
        "            return\n",
        "\n",
        "        df['date'] = pd.to_datetime(df['date'])\n",
        "        df['hour'] = pd.to_datetime(df['time']).dt.hour\n",
        "\n",
        "        fig = make_subplots(\n",
        "            rows=3, cols=3,\n",
        "            subplot_titles=(\n",
        "                'Daily Entry Counts',\n",
        "                'Hourly Distribution',\n",
        "                'Sentiment Timeline',\n",
        "                'Summary Length Distribution',\n",
        "                'Sentiment Distribution',\n",
        "                'Weekly Patterns',\n",
        "                'Sentiment Moving Average',\n",
        "                'Headline Length vs Sentiment',\n",
        "                'Time of Day Sentiment'\n",
        "            ),\n",
        "            specs=[\n",
        "                [{'type': 'scatter'}, {'type': 'bar'}, {'type': 'scatter'}],\n",
        "                [{'type': 'histogram'}, {'type': 'histogram'}, {'type': 'heatmap'}],\n",
        "                [{'type': 'scatter'}, {'type': 'scatter'}, {'type': 'scatter'}]\n",
        "            ],\n",
        "            horizontal_spacing=0.12,\n",
        "            vertical_spacing=0.15\n",
        "        )\n",
        "\n",
        "        daily_counts = df.groupby('date').size()\n",
        "        fig.add_trace(\n",
        "            go.Scatter(\n",
        "                x=daily_counts.index,\n",
        "                y=daily_counts.values,\n",
        "                mode='lines+markers',\n",
        "                name='Daily Entries',\n",
        "                line=dict(width=2, color='royalblue'),\n",
        "                marker=dict(size=6)\n",
        "            ),\n",
        "            row=1, col=1\n",
        "        )\n",
        "\n",
        "        hourly_counts = df['hour'].value_counts().sort_index()\n",
        "        fig.add_trace(\n",
        "            go.Bar(\n",
        "                x=hourly_counts.index,\n",
        "                y=hourly_counts.values,\n",
        "                name='Hourly Distribution',\n",
        "                marker_color='lightblue'\n",
        "            ),\n",
        "            row=1, col=2\n",
        "        )\n",
        "\n",
        "        daily_sentiment = df.groupby('date')['score'].mean()\n",
        "        fig.add_trace(\n",
        "            go.Scatter(\n",
        "                x=daily_sentiment.index,\n",
        "                y=daily_sentiment.values,\n",
        "                mode='lines',\n",
        "                name='Daily Sentiment',\n",
        "                line=dict(color='green', width=2)\n",
        "            ),\n",
        "            row=1, col=3\n",
        "        )\n",
        "\n",
        "        summary_lengths = DataVisualizer.get_summary_lengths(db_name)\n",
        "        fig.add_trace(\n",
        "            go.Histogram(\n",
        "                x=summary_lengths,\n",
        "                name='Summary Lengths',\n",
        "                nbinsx=30,\n",
        "                marker_color='lightgreen'\n",
        "            ),\n",
        "            row=2, col=1\n",
        "        )\n",
        "\n",
        "        fig.add_trace(\n",
        "            go.Histogram(\n",
        "                x=df['score'],\n",
        "                name='Sentiment Distribution',\n",
        "                nbinsx=20,\n",
        "                histnorm='probability',\n",
        "                marker_color='coral'\n",
        "            ),\n",
        "            row=2, col=2\n",
        "        )\n",
        "\n",
        "        df['weekday'] = pd.to_datetime(df['date']).dt.day_name()\n",
        "        weekly_sentiment = df.pivot_table(\n",
        "            values='score',\n",
        "            index='weekday',\n",
        "            columns='hour',\n",
        "            aggfunc='mean'\n",
        "        )\n",
        "\n",
        "        weekday_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']\n",
        "        weekly_sentiment = weekly_sentiment.reindex(weekday_order)\n",
        "\n",
        "        fig.add_trace(\n",
        "            go.Heatmap(\n",
        "                z=weekly_sentiment.values,\n",
        "                x=weekly_sentiment.columns,\n",
        "                y=weekly_sentiment.index,\n",
        "                colorscale='RdYlBu',\n",
        "                name='Weekly Patterns',\n",
        "                colorbar=dict(\n",
        "                    title=\"Sentiment\",\n",
        "                    thickness=10,\n",
        "                    len=0.3,\n",
        "                    yanchor=\"middle\",\n",
        "                    y=0.5,\n",
        "                    xanchor=\"left\",\n",
        "                    x=1.02\n",
        "                )\n",
        "            ),\n",
        "            row=2, col=3\n",
        "        )\n",
        "\n",
        "        daily_sentiment = df.groupby('date')['score'].mean().reset_index()\n",
        "        daily_sentiment['MA7'] = daily_sentiment['score'].rolling(window=7).mean()\n",
        "        fig.add_trace(\n",
        "            go.Scatter(\n",
        "                x=daily_sentiment['date'],\n",
        "                y=daily_sentiment['MA7'],\n",
        "                mode='lines',\n",
        "                name='7-Day Moving Average',\n",
        "                line=dict(color='purple', width=2)\n",
        "            ),\n",
        "            row=3, col=1\n",
        "        )\n",
        "\n",
        "        df['title_length'] = df['title'].str.len()\n",
        "        fig.add_trace(\n",
        "            go.Scatter(\n",
        "                x=df['title_length'],\n",
        "                y=df['score'],\n",
        "                mode='markers',\n",
        "                name='Length vs Sentiment',\n",
        "                marker=dict(\n",
        "                    size=6,\n",
        "                    color=df['score'],\n",
        "                    colorscale='Viridis',\n",
        "                    showscale=False\n",
        "                )\n",
        "            ),\n",
        "            row=3, col=2\n",
        "        )\n",
        "\n",
        "        hourly_sentiment = df.groupby('hour')['score'].mean()\n",
        "        fig.add_trace(\n",
        "            go.Scatter(\n",
        "                x=hourly_sentiment.index,\n",
        "                y=hourly_sentiment.values,\n",
        "                mode='lines+markers',\n",
        "                name='Hourly Sentiment',\n",
        "                line=dict(shape='spline', color='orangered', width=2)\n",
        "            ),\n",
        "            row=3, col=3\n",
        "        )\n",
        "\n",
        "        fig.update_layout(\n",
        "            height=1200,\n",
        "            width=1600,\n",
        "            showlegend=True,\n",
        "            legend=dict(\n",
        "                orientation=\"h\",\n",
        "                yanchor=\"bottom\",\n",
        "                y=-0.2,\n",
        "                xanchor=\"center\",\n",
        "                x=0.5\n",
        "            ),\n",
        "            title_text=\"Enhanced News Analysis Dashboard\",\n",
        "            template=\"plotly_white\"\n",
        "        )\n",
        "\n",
        "        axis_labels = {\n",
        "            (1,1): ('Date', 'Number of Articles'),\n",
        "            (1,2): ('Hour of Day', 'Number of Articles'),\n",
        "            (1,3): ('Date', 'Average Sentiment Score'),\n",
        "            (2,1): ('Summary Length (characters)', 'Frequency'),\n",
        "            (2,2): ('Sentiment Score', 'Probability'),\n",
        "            (2,3): ('Hour of Day', 'Day of Week'),\n",
        "            (3,1): ('Date', '7-Day Moving Average'),\n",
        "            (3,2): ('Headline Length (characters)', 'Sentiment Score'),\n",
        "            (3,3): ('Hour of Day', 'Average Sentiment Score')\n",
        "        }\n",
        "\n",
        "        for (row, col), (xlabel, ylabel) in axis_labels.items():\n",
        "            fig.update_xaxes(title_text=xlabel, row=row, col=col)\n",
        "            fig.update_yaxes(title_text=ylabel, row=row, col=col)\n",
        "\n",
        "        fig.show()\n",
        "\n",
        "    @staticmethod\n",
        "    def get_summary_lengths(db_name: str) -> list:\n",
        "        \"\"\"\n",
        "        Retrieves the lengths of all summaries stored in the sentiment_scores table.\n",
        "\n",
        "        Parameters:\n",
        "            db_name (str): The name of the SQLite database to query.\n",
        "\n",
        "        Returns:\n",
        "            list: A list of summary lengths.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            with sqlite3.connect(db_name) as conn:\n",
        "                cursor = conn.cursor()\n",
        "                cursor.execute(\"SELECT LENGTH(summary) FROM sentiment_scores\")\n",
        "                return [row[0] for row in cursor.fetchall()]\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error getting summary lengths: {e}\")\n",
        "            return []\n",
        "\n",
        "class NewsFetcher:\n",
        "    \"\"\"\n",
        "    Fetches news articles from an RSS feed, processes them to analyze sentiment,\n",
        "    and stores the results in the database. Ensures that duplicate or similar\n",
        "    articles are not processed multiple times.\n",
        "    \"\"\"\n",
        "    def __init__(self, db_manager, sentiment_analyzer):\n",
        "        self.db_manager = db_manager\n",
        "        self.sentiment_analyzer = sentiment_analyzer\n",
        "        self._initialize_processed_titles()\n",
        "\n",
        "    def _initialize_processed_titles(self):\n",
        "        \"\"\"\n",
        "        Loads existing article titles from the database to avoid processing duplicates.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            with sqlite3.connect(self.db_manager.db_name) as conn:\n",
        "                cursor = conn.execute(\"SELECT title FROM sentiment_scores\")\n",
        "                self.processed_titles = {row[0] for row in cursor.fetchall()}\n",
        "            logging.info(f\"Loaded {len(self.processed_titles)} existing titles from database\")\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error loading existing titles: {e}\")\n",
        "            self.processed_titles = set()\n",
        "\n",
        "    def _is_similar_to_existing(self, title: str, threshold: float = 0.85) -> bool:\n",
        "        \"\"\"\n",
        "        Determines if a given title is similar to any already processed titles using SequenceMatcher.\n",
        "\n",
        "        Parameters:\n",
        "            title (str): The title to compare.\n",
        "            threshold (float): The similarity threshold above which a title is considered similar.\n",
        "\n",
        "        Returns:\n",
        "            bool: True if the title is similar to an existing one, False otherwise.\n",
        "        \"\"\"\n",
        "        for existing_title in self.processed_titles:\n",
        "            similarity = SequenceMatcher(None, title.lower(), existing_title.lower()).ratio()\n",
        "            if similarity > threshold:\n",
        "                logging.info(f\"Found similar title (similarity: {similarity:.2f})\")\n",
        "                return True\n",
        "        return False\n",
        "\n",
        "    def _is_valid_entry(self, entry) -> bool:\n",
        "        \"\"\"\n",
        "        Validates a news entry based on the presence and length of its summary and title,\n",
        "        and the existence of a publication date.\n",
        "\n",
        "        Parameters:\n",
        "            entry: The news entry to validate.\n",
        "\n",
        "        Returns:\n",
        "            bool: True if the entry is valid, False otherwise.\n",
        "        \"\"\"\n",
        "        if not hasattr(entry, 'summary') or len(entry.summary.strip()) < 17:\n",
        "            logging.debug(f\"Skipping entry with insufficient summary: {entry.title[:50]}...\")\n",
        "            return False\n",
        "\n",
        "        if not hasattr(entry, 'title') or len(entry.title.strip()) < 5:\n",
        "            logging.debug(\"Skipping entry with insufficient title\")\n",
        "            return False\n",
        "\n",
        "        if not hasattr(entry, 'published'):\n",
        "            logging.debug(\"Skipping entry without published date\")\n",
        "            return False\n",
        "\n",
        "        return True\n",
        "\n",
        "    def fetch_and_process_news(self):\n",
        "        \"\"\"\n",
        "        Fetches news articles from the RSS feed, analyzes their sentiment,\n",
        "        and stores valid and unique entries into the database.\n",
        "        \"\"\"\n",
        "        try:\n",
        "            logging.info(\"Fetching news from RSS feed...\")\n",
        "            feed = feedparser.parse(NEWS_FEED_URL)\n",
        "            new_entries_count = 0\n",
        "            skipped_entries_count = 0\n",
        "\n",
        "            for entry in feed.entries:\n",
        "                if not self._is_valid_entry(entry):\n",
        "                    skipped_entries_count += 1\n",
        "                    continue\n",
        "\n",
        "                if entry.title in self.processed_titles or self._is_similar_to_existing(entry.title):\n",
        "                    logging.debug(f\"Skipping duplicate/similar title: {entry.title[:50]}...\")\n",
        "                    skipped_entries_count += 1\n",
        "                    continue\n",
        "\n",
        "                try:\n",
        "                    published = datetime.strptime(entry.published, DATETIME_FORMAT)\n",
        "                    date = published.strftime(DATE_FORMAT)\n",
        "                    time = published.strftime(\"%H:%M:%S\")\n",
        "\n",
        "                    combined_text = f\"{entry.title} {entry.summary}\"\n",
        "                    sentiment_score = self.sentiment_analyzer.analyze_sentiment(combined_text)\n",
        "\n",
        "                    self.db_manager.store_score(\n",
        "                        date=date,\n",
        "                        time=time,\n",
        "                        title=entry.title,\n",
        "                        summary=entry.summary,\n",
        "                        score=sentiment_score\n",
        "                    )\n",
        "\n",
        "                    self.processed_titles.add(entry.title)\n",
        "                    new_entries_count += 1\n",
        "\n",
        "                    logging.info(f\"Processed new article: {entry.title[:50]}...\")\n",
        "\n",
        "                except Exception as e:\n",
        "                    logging.error(f\"Error processing entry: {e}\")\n",
        "                    skipped_entries_count += 1\n",
        "                    continue\n",
        "\n",
        "            logging.info(f\"Added {new_entries_count} new articles, skipped {skipped_entries_count} invalid/duplicate entries\")\n",
        "\n",
        "        except Exception as e:\n",
        "            logging.error(f\"Error fetching news: {e}\")\n",
        "            raise\n",
        "\n",
        "# Modified run_analysis function\n",
        "def run_analysis():\n",
        "    \"\"\"\n",
        "    Orchestrates the entire analysis process by initializing components,\n",
        "    fetching and processing news articles, generating visualizations,\n",
        "    and performing headline analysis. Ensures that all resources are properly closed.\n",
        "    \"\"\"\n",
        "    try:\n",
        "        logging.info(\"Starting analysis...\")\n",
        "\n",
        "        db_manager = DatabaseManager(DATABASE_NAME)\n",
        "        sentiment_analyzer = SentimentAnalyzer()\n",
        "        news_fetcher = NewsFetcher(db_manager, sentiment_analyzer)\n",
        "\n",
        "        logging.info(\"Fetching and processing news...\")\n",
        "        news_fetcher.fetch_and_process_news()\n",
        "\n",
        "        logging.info(\"Generating main dashboard...\")\n",
        "        DataVisualizer.create_visualizations(DATABASE_NAME)\n",
        "\n",
        "        logging.info(\"Analyzing headlines...\")\n",
        "        analyze_headlines()\n",
        "\n",
        "        logging.info(\"Analysis complete!\")\n",
        "\n",
        "    except Exception as e:\n",
        "        logging.error(f\"Error during analysis: {e}\")\n",
        "        raise\n",
        "    finally:\n",
        "        if 'db_manager' in locals():\n",
        "            db_manager.close()\n",
        "\n",
        "def analyze_headlines():\n",
        "    \"\"\"\n",
        "    Analyzes and displays recent headlines along with their sentiment scores.\n",
        "    Retrieves statistical data and extreme sentiment headlines, presenting them in a structured dashboard.\n",
        "    \"\"\"\n",
        "    try:\n",
        "        db_manager = DatabaseManager(DATABASE_NAME)\n",
        "\n",
        "        stats = db_manager.get_headline_stats()\n",
        "\n",
        "        recent_headlines = db_manager.get_headlines_with_scores(limit=10)\n",
        "        most_positive, most_negative = db_manager.get_extreme_sentiment_headlines()\n",
        "\n",
        "        fig = make_subplots(\n",
        "            rows=3, cols=1,\n",
        "            subplot_titles=(\n",
        "                'Recent Headlines with Sentiment Scores',\n",
        "                'Most Positive Headlines',\n",
        "                'Most Negative Headlines'\n",
        "            ),\n",
        "            specs=[[{\"type\": \"table\"}],\n",
        "                   [{\"type\": \"table\"}],\n",
        "                   [{\"type\": \"table\"}]],\n",
        "            vertical_spacing=0.1\n",
        "        )\n",
        "\n",
        "        fig.add_annotation(\n",
        "            text=(f\"Total Headlines: {stats.get('total_headlines', 'N/A')} | \"\n",
        "                  f\"Average Sentiment: {stats.get('average_sentiment', 0):.2f} | \"\n",
        "                  f\"Date Range: {stats.get('date_range', 'N/A')}\"),\n",
        "            xref=\"paper\", yref=\"paper\",\n",
        "            x=0, y=1.1,\n",
        "            showarrow=False,\n",
        "            font=dict(size=12)\n",
        "        )\n",
        "\n",
        "        fig.add_trace(\n",
        "            go.Table(\n",
        "                header=dict(\n",
        "                    values=['Date', 'Time', 'Headline', 'Sentiment Score'],\n",
        "                    fill_color='paleturquoise',\n",
        "                    align='left',\n",
        "                    font=dict(size=12)\n",
        "                ),\n",
        "                cells=dict(\n",
        "                    values=[recent_headlines['date'],\n",
        "                           recent_headlines['time'],\n",
        "                           recent_headlines['title'],\n",
        "                           recent_headlines['score'].round(2)],\n",
        "                    fill_color='lavender',\n",
        "                    align='left',\n",
        "                    font=dict(size=11)\n",
        "                )\n",
        "            ),\n",
        "            row=1, col=1\n",
        "        )\n",
        "\n",
        "        fig.add_trace(\n",
        "            go.Table(\n",
        "                header=dict(\n",
        "                    values=['Date', 'Time', 'Headline', 'Sentiment Score'],\n",
        "                    fill_color='lightgreen',\n",
        "                    align='left'\n",
        "                ),\n",
        "                cells=dict(\n",
        "                    values=[most_positive['date'],\n",
        "                           most_positive['time'],\n",
        "                           most_positive['title'],\n",
        "                           most_positive['score'].round(2)],\n",
        "                    fill_color='honeydew',\n",
        "                    align='left'\n",
        "                )\n",
        "            ),\n",
        "            row=2, col=1\n",
        "        )\n",
        "\n",
        "        fig.add_trace(\n",
        "            go.Table(\n",
        "                header=dict(\n",
        "                    values=['Date', 'Time', 'Headline', 'Sentiment Score'],\n",
        "                    fill_color='lightcoral',\n",
        "                    align='left'\n",
        "                ),\n",
        "                cells=dict(\n",
        "                    values=[most_negative['date'],\n",
        "                           most_negative['time'],\n",
        "                           most_negative['title'],\n",
        "                           most_negative['score'].round(2)],\n",
        "                    fill_color='mistyrose',\n",
        "                    align='left'\n",
        "                )\n",
        "            ),\n",
        "            row=3, col=1\n",
        "        )\n",
        "\n",
        "        fig.update_layout(\n",
        "            height=1000,\n",
        "            title_text=\"Headlines Analysis Dashboard\",\n",
        "            showlegend=False\n",
        "        )\n",
        "\n",
        "        fig.show()\n",
        "\n",
        "    except Exception as e:\n",
        "        logging.error(f\"Error analyzing headlines: {e}\")\n",
        "        raise\n",
        "    finally:\n",
        "        db_manager.close()\n",
        "\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    run_analysis()"
      ],
      "metadata": {
        "id": "up5A0BZQfLkR"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import sqlite3\n",
        "from difflib import SequenceMatcher\n",
        "import pandas as pd\n",
        "from typing import List, Tuple\n",
        "import logging\n",
        "\n",
        "def initialize_database(db_path: str = '/content/news_sentiment.db') -> None:\n",
        "    \"\"\"\n",
        "    Create the database and required tables if they don't exist.\n",
        "\n",
        "    This function initializes the SQLite database by creating the 'sentiment_scores' table\n",
        "    with the necessary columns if it doesn't already exist. It also creates indexes on\n",
        "    'date', 'title', and 'score' columns to optimize query performance.\n",
        "\n",
        "    Parameters:\n",
        "        db_path (str): The file path to the SQLite database. Defaults to '/content/news_sentiment.db'.\n",
        "\n",
        "    Raises:\n",
        "        Exception: If an error occurs while initializing the database.\n",
        "    \"\"\"\n",
        "    try:\n",
        "        with sqlite3.connect(db_path) as conn:\n",
        "            conn.execute('''CREATE TABLE IF NOT EXISTS sentiment_scores (\n",
        "                                date TEXT,\n",
        "                                time TEXT,\n",
        "                                title TEXT,\n",
        "                                summary TEXT,\n",
        "                                score REAL\n",
        "                            )''')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_date ON sentiment_scores(date)')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_title ON sentiment_scores(title)')\n",
        "            conn.execute('CREATE INDEX IF NOT EXISTS idx_score ON sentiment_scores(score)')\n",
        "\n",
        "        logging.info(f\"Database initialized successfully at {db_path}\")\n",
        "    except Exception as e:\n",
        "        logging.error(f\"Error initializing database: {e}\")\n",
        "        raise\n",
        "\n",
        "def calculate_similarity(text1: str, text2: str) -> float:\n",
        "    \"\"\"\n",
        "    Calculate similarity ratio between two texts using SequenceMatcher.\n",
        "\n",
        "    This function computes the similarity ratio between two input strings by utilizing\n",
        "    the SequenceMatcher from the difflib library. The comparison is case-insensitive.\n",
        "\n",
        "    Parameters:\n",
        "        text1 (str): The first text string.\n",
        "        text2 (str): The second text string.\n",
        "\n",
        "    Returns:\n",
        "        float: A value between 0 and 1 representing the similarity ratio.\n",
        "    \"\"\"\n",
        "    return SequenceMatcher(None, text1.lower(), text2.lower()).ratio()\n",
        "\n",
        "def find_similar_entries(conn: sqlite3.Connection, similarity_threshold: float = 0.85) -> List[Tuple[int, int, float]]:\n",
        "    \"\"\"\n",
        "    Find pairs of similar entries in the database.\n",
        "\n",
        "    This function retrieves all records from the 'sentiment_scores' table and compares\n",
        "    each entry's title and summary with others to find pairs that exceed the specified\n",
        "    similarity threshold. To optimize performance, it limits the number of comparisons\n",
        "    per entry using a sliding window.\n",
        "\n",
        "    Parameters:\n",
        "        conn (sqlite3.Connection): An active SQLite database connection.\n",
        "        similarity_threshold (float): The threshold above which two entries are considered similar. Defaults to 0.85.\n",
        "\n",
        "    Returns:\n",
        "        List[Tuple[int, int, float]]: A list of tuples, each containing the rowids of the two similar entries and their average similarity score.\n",
        "    \"\"\"\n",
        "    df = pd.read_sql_query(\"SELECT rowid, title, summary FROM sentiment_scores\", conn)\n",
        "    similar_pairs = []\n",
        "\n",
        "    for i in range(len(df)):\n",
        "        for j in range(i + 1, min(i + 50, len(df))):\n",
        "            title_similarity = calculate_similarity(df.iloc[i]['title'], df.iloc[j]['title'])\n",
        "            if title_similarity > similarity_threshold:\n",
        "                summary_similarity = calculate_similarity(df.iloc[i]['summary'] or '', df.iloc[j]['summary'] or '')\n",
        "                if summary_similarity > similarity_threshold:\n",
        "                    similar_pairs.append((df.iloc[i]['rowid'], df.iloc[j]['rowid'], (title_similarity + summary_similarity) / 2))\n",
        "    return similar_pairs\n",
        "\n",
        "def remove_duplicates(conn: sqlite3.Connection, similarity_threshold: float = 0.85) -> None:\n",
        "    \"\"\"\n",
        "    Remove duplicate or very similar entries from the database.\n",
        "\n",
        "    This function identifies similar entries in the 'sentiment_scores' table based on\n",
        "    the similarity threshold and removes the later entries to eliminate duplicates.\n",
        "\n",
        "    Parameters:\n",
        "        conn (sqlite3.Connection): An active SQLite database connection.\n",
        "        similarity_threshold (float): The threshold above which two entries are considered similar. Defaults to 0.85.\n",
        "\n",
        "    Returns:\n",
        "        None\n",
        "    \"\"\"\n",
        "    similar_pairs = find_similar_entries(conn, similarity_threshold)\n",
        "    if not similar_pairs:\n",
        "        print(\"No duplicate or similar entries found.\")\n",
        "        return\n",
        "\n",
        "    entries_to_remove = set()\n",
        "    for _, later_entry, _ in similar_pairs:\n",
        "        entries_to_remove.add(later_entry)\n",
        "\n",
        "    cursor = conn.cursor()\n",
        "    cursor.execute(\"DELETE FROM sentiment_scores WHERE rowid IN ({})\".format(','.join('?' * len(entries_to_remove))),\n",
        "                   tuple(entries_to_remove))\n",
        "\n",
        "    print(f\"Removed {len(entries_to_remove)} duplicate/similar entries.\")\n",
        "    print(f\"Remaining entries: {cursor.execute('SELECT COUNT(*) FROM sentiment_scores').fetchone()[0]}\")\n",
        "\n",
        "    conn.commit()\n",
        "\n",
        "def setup_and_clean_database(db_path: str = '/content/news_sentiment.db'):\n",
        "    \"\"\"\n",
        "    Initialize database if it doesn't exist and clean up duplicates.\n",
        "\n",
        "    This function ensures that the database is properly initialized by creating the necessary\n",
        "    tables and indexes. It then proceeds to remove any duplicate or very similar entries\n",
        "    to maintain data integrity.\n",
        "\n",
        "    Parameters:\n",
        "        db_path (str): The file path to the SQLite database. Defaults to '/content/news_sentiment.db'.\n",
        "\n",
        "    Returns:\n",
        "        None\n",
        "    \"\"\"\n",
        "    initialize_database(db_path)\n",
        "    with sqlite3.connect(db_path) as conn:\n",
        "        remove_duplicates(conn)\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    \"\"\"\n",
        "    The main entry point of the script.\n",
        "\n",
        "    Initializes logging configurations and executes the database setup and cleaning process.\n",
        "    \"\"\"\n",
        "    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n",
        "    setup_and_clean_database()"
      ],
      "metadata": {
        "id": "yLLOSKfWe0Ml",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "9fd485a9-5d23-40e1-f271-403f144a8e37"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "No duplicate or similar entries found.\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# cell 4\n",
        "import sqlite3\n",
        "\n",
        "def find_shortest_entries(db_path: str = '/content/news_sentiment.db', num_entries: int = 5) -> None:\n",
        "    \"\"\"\n",
        "    Find and print the entries with the smallest lengths in the database.\n",
        "\n",
        "    This function connects to the specified SQLite database and retrieves the top `num_entries`\n",
        "    from the `sentiment_scores` table based on the smallest lengths of the `title` and `summary`.\n",
        "    It orders the entries in ascending order of `title_length` and `summary_length` and prints\n",
        "    the truncated title and summary along with their respective lengths.\n",
        "\n",
        "    Parameters:\n",
        "        db_path (str): The file path to the SQLite database. Defaults to '/content/news_sentiment.db'.\n",
        "        num_entries (int): The number of shortest entries to retrieve and print. Defaults to 5.\n",
        "\n",
        "    Returns:\n",
        "        None\n",
        "    \"\"\"\n",
        "    try:\n",
        "        with sqlite3.connect(db_path) as conn:\n",
        "            cursor = conn.cursor()\n",
        "            cursor.execute(\"\"\"\n",
        "                SELECT title, summary, LENGTH(title) AS title_length, LENGTH(summary) AS summary_length\n",
        "                FROM sentiment_scores\n",
        "                ORDER BY title_length ASC, summary_length ASC\n",
        "                LIMIT ?\n",
        "            \"\"\", (num_entries,))\n",
        "\n",
        "            rows = cursor.fetchall()\n",
        "            for row in rows:\n",
        "                title, summary, title_length, summary_length = row\n",
        "                print(f\"Title: {title[:50]}..., Length: {title_length}\")\n",
        "                print(f\"Summary: {summary[:50]}..., Length: {summary_length}\")\n",
        "                print(\"-\" * 20)\n",
        "    except Exception as e:\n",
        "        print(f\"Error finding shortest entries: {e}\")\n",
        "\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    find_shortest_entries()"
      ],
      "metadata": {
        "id": "bA29R_4fu-8q",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "d0d9b256-2781-4b13-f940-500f43409b43"
      },
      "execution_count": 4,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Title: Just an accessory..., Length: 17\n",
            "Summary: After Hina lost her earring, she didn’t want to we..., Length: 81\n",
            "--------------------\n",
            "Title: Column | Invent a word..., Length: 22\n",
            "Summary: A new column from someone who doesn’t let the dict..., Length: 101\n",
            "--------------------\n",
            "Title: Vijay cannot be ignored..., Length: 23\n",
            "Summary: His speech has drawn admiration and criticism in e..., Length: 62\n",
            "--------------------\n",
            "Title: In the heart of the jungle..., Length: 26\n",
            "Summary: At the Agumbe Rainforest Research Station, Taran d..., Length: 107\n",
            "--------------------\n",
            "Title: Fighting high-stakes bypolls..., Length: 28\n",
            "Summary: Wayanad, Palakkad, and Chelakkara will all witness..., Length: 75\n",
            "--------------------\n"
          ]
        }
      ]
    }
  ]
}