summarizer.py

"""
Text summarization methods for RQ3/Objective 4.

Implements multiple summarization strategies to test information loss.
"""
import numpy as np
from typing import List, Dict
from sklearn.feature_extraction.text import TfidfVectorizer
from sumy.parsers.plaintext import PlaintextParser
from sumy.nlp.tokenizers import Tokenizer
from sumy.summarizers.text_rank import TextRankSummarizer
from sumy.summarizers.lex_rank import LexRankSummarizer
from sumy.summarizers.lsa import LsaSummarizer
import warnings
warnings.filterwarnings('ignore')


class DocumentSummarizer:
    """
    Multiple text summarization methods for comparing information loss.
    """

    def __init__(self, language: str = 'english'):
        """
        Initialize summarizer.

        Parameters:
        -----------
        language : str
            Language for tokenization
        """
        self.language = language

    def _get_target_length(self, text: str, ratio: float, min_length: int = 50,
                          max_length: int = 512) -> int:
        """
        Calculate target summary length based on ratio.

        Parameters:
        -----------
        text : str
            Original text
        ratio : float
            Summary ratio (0.0 - 1.0)
        min_length : int
            Minimum summary length
        max_length : int
            Maximum summary length

        Returns:
        --------
        int
            Target length in tokens
        """
        words = text.split()
        target = int(len(words) * ratio)
        return max(min_length, min(target, max_length))

    def extractive_tfidf(self, text: str, ratio: float = 0.25,
                        n_sentences: int = 3) -> str:
        """
        Extractive summarization using TF-IDF sentence scoring.

        Parameters:
        -----------
        text : str
            Input text
        ratio : float
            Summary ratio
        n_sentences : int
            Fallback number of sentences if ratio cannot be applied

        Returns:
        --------
        str
            Summary text
        """
        if not text or len(text.strip()) == 0:
            return ""

        # Split into sentences (simple)
        sentences = [s.strip() for s in text.split('.') if len(s.strip()) > 10]

        # Derive sentence count from ratio; fall back to n_sentences for very short docs
        target_n = max(1, int(len(sentences) * ratio)) if len(sentences) > 0 else n_sentences

        if len(sentences) <= target_n:
            return text

        # Vectorize sentences
        try:
            vectorizer = TfidfVectorizer(stop_words='english', max_features=100)
            tfidf_matrix = vectorizer.fit_transform(sentences)

            # Score sentences by sum of TF-IDF
            sentence_scores = np.array(tfidf_matrix.sum(axis=1)).flatten()

            # Get top sentences
            top_indices = sentence_scores.argsort()[-target_n:][::-1]
            top_indices = sorted(top_indices)  # Maintain original order

            summary_sentences = [sentences[i] for i in top_indices]
            return '. '.join(summary_sentences) + '.'

        except:
            # Fallback: return first target_n sentences
            return '. '.join(sentences[:target_n]) + '.'

    def textrank(self, text: str, ratio: float = 0.25, n_sentences: int = 3) -> str:
        """
        TextRank summarization (graph-based).

        Parameters:
        -----------
        text : str
            Input text
        ratio : float
            Summary ratio
        n_sentences : int
            Number of sentences

        Returns:
        --------
        str
            Summary text
        """
        if not text or len(text.strip()) == 0:
            return ""

        try:
            parser = PlaintextParser.from_string(text, Tokenizer(self.language))
            total_sents = len(list(parser.document.sentences))
            target_n = max(1, int(total_sents * ratio)) if total_sents > 0 else n_sentences
            summarizer = TextRankSummarizer()
            summary = summarizer(parser.document, target_n)

            return ' '.join([str(sentence) for sentence in summary])

        except:
            # Fallback
            return self.extractive_tfidf(text, ratio, n_sentences)

    def lexrank(self, text: str, ratio: float = 0.25, n_sentences: int = 3) -> str:
        """
        LexRank summarization (graph-based).

        Parameters:
        -----------
        text : str
            Input text
        ratio : float
            Summary ratio
        n_sentences : int
            Number of sentences

        Returns:
        --------
        str
            Summary text
        """
        if not text or len(text.strip()) == 0:
            return ""

        try:
            parser = PlaintextParser.from_string(text, Tokenizer(self.language))
            total_sents = len(list(parser.document.sentences))
            target_n = max(1, int(total_sents * ratio)) if total_sents > 0 else n_sentences
            summarizer = LexRankSummarizer()
            summary = summarizer(parser.document, target_n)

            return ' '.join([str(sentence) for sentence in summary])

        except:
            # Fallback
            return self.extractive_tfidf(text, ratio, n_sentences)

    def lsa(self, text: str, ratio: float = 0.25, n_sentences: int = 3) -> str:
        """
        LSA (Latent Semantic Analysis) summarization.

        Parameters:
        -----------
        text : str
            Input text
        ratio : float
            Summary ratio
        n_sentences : int
            Number of sentences

        Returns:
        --------
        str
            Summary text
        """
        if not text or len(text.strip()) == 0:
            return ""

        try:
            parser = PlaintextParser.from_string(text, Tokenizer(self.language))
            total_sents = len(list(parser.document.sentences))
            target_n = max(1, int(total_sents * ratio)) if total_sents > 0 else n_sentences
            summarizer = LsaSummarizer()
            summary = summarizer(parser.document, target_n)

            return ' '.join([str(sentence) for sentence in summary])

        except:
            # Fallback
            return self.extractive_tfidf(text, ratio, n_sentences)

    def first_n_sentences(self, text: str, ratio: float = 0.25,
                         n_sentences: int = 3) -> str:
        """
        Baseline: Extract first N sentences.

        Parameters:
        -----------
        text : str
            Input text
        ratio : float
            Summary ratio (ignored, using n_sentences)
        n_sentences : int
            Number of sentences

        Returns:
        --------
        str
            Summary text
        """
        if not text or len(text.strip()) == 0:
            return ""

        sentences = [s.strip() for s in text.split('.') if len(s.strip()) > 10]

        target_n = max(1, int(len(sentences) * ratio)) if len(sentences) > 0 else n_sentences

        if len(sentences) <= target_n:
            return text

        return '. '.join(sentences[:target_n]) + '.'

    def summarize_documents(self, docs: List[str], method: str = 'extractive_tfidf',
                           ratio: float = 0.25, n_sentences: int = 3) -> List[str]:
        """
        Summarize a list of documents.

        Parameters:
        -----------
        docs : List[str]
            List of documents
        method : str
            Summarization method ('extractive_tfidf', 'textrank', 'lexrank', 'lsa', 'first_n')
        ratio : float
            Summary ratio
        n_sentences : int
            Number of sentences

        Returns:
        --------
        List[str]
            List of summaries
        """
        print(f"Summarizing {len(docs)} documents using {method} (ratio={ratio})...")

        summaries = []

        for i, doc in enumerate(docs):
            if i % 1000 == 0 and i > 0:
                print(f"  Processed {i}/{len(docs)} documents...")

            if method == 'extractive_tfidf':
                summary = self.extractive_tfidf(doc, ratio, n_sentences)
            elif method == 'textrank':
                summary = self.textrank(doc, ratio, n_sentences)
            elif method == 'lexrank':
                summary = self.lexrank(doc, ratio, n_sentences)
            elif method == 'lsa':
                summary = self.lsa(doc, ratio, n_sentences)
            elif method == 'first_n':
                summary = self.first_n_sentences(doc, ratio, n_sentences)
            else:
                raise ValueError(f"Unknown method: {method}")

            summaries.append(summary if summary else doc)  # Keep original if summary fails

        print(f"Summarization complete. Generated {len(summaries)} summaries.")
        return summaries

    def calculate_compression_ratio(self, original_docs: List[str],
                                   summaries: List[str]) -> Dict[str, float]:
        """
        Calculate actual compression ratio achieved.

        Parameters:
        -----------
        original_docs : List[str]
            Original documents
        summaries : List[str]
            Summarized documents

        Returns:
        --------
        Dict[str, float]
            Compression statistics
        """
        original_lengths = [len(doc.split()) for doc in original_docs]
        summary_lengths = [len(summary.split()) for summary in summaries]

        avg_original = np.mean(original_lengths)
        avg_summary = np.mean(summary_lengths)
        compression_ratio = avg_summary / avg_original if avg_original > 0 else 0

        return {
            'avg_original_length': avg_original,
            'avg_summary_length': avg_summary,
            'compression_ratio': compression_ratio,
            'total_original_words': sum(original_lengths),
            'total_summary_words': sum(summary_lengths)
        }

    def summarize_with_multiple_methods(self, docs: List[str],
                                       methods: List[str],
                                       ratios: List[float],
                                       n_sentences: int = 3) -> Dict[str, Dict[float, List[str]]]:
        """
        Summarize documents using multiple methods and ratios.

        Parameters:
        -----------
        docs : List[str]
            Documents to summarize
        methods : List[str]
            List of methods to use
        ratios : List[float]
            List of compression ratios to test
        n_sentences : int
            Number of sentences for extractive methods

        Returns:
        --------
        Dict[str, Dict[float, List[str]]]
            Nested dict: {method: {ratio: summaries}}
        """
        results = {}

        for method in methods:
            results[method] = {}
            for ratio in ratios:
                print(f"\nMethod: {method}, Ratio: {ratio}")
                summaries = self.summarize_documents(docs, method, ratio, n_sentences)
                results[method][ratio] = summaries

                # Print compression stats
                stats = self.calculate_compression_ratio(docs, summaries)
                print(f"  Actual compression: {stats['compression_ratio']:.2%}")
                print(f"  Avg original: {stats['avg_original_length']:.1f} words")
                print(f"  Avg summary: {stats['avg_summary_length']:.1f} words")

        return results