src/application/content_detection.py

"""
Author: Khanh Phan
Date: 2024-12-04
"""

import pandas as pd

from src.application.config import (
    MIN_RATIO_PARAPHRASE_NUM,
    PARAPHRASE_THRESHOLD,
    PARAPHRASE_THRESHOLD_MACHINE,
)
from src.application.formatting_fact_checker import create_fact_checker_table
from src.application.formatting_governor import create_governor_table
from src.application.formatting_ordinary_user import create_ordinary_user_table
from src.application.image.image import ImageDetector
from src.application.image.image_detection import (
    detect_image_by_ai_model,
    detect_image_by_reverse_search,
    detect_image_from_news_image,
)
from src.application.text.entity import highlight_entities
from src.application.text.helper import (
    postprocess_label,
    split_into_paragraphs,
)
from src.application.text.model_detection import (
    detect_text_by_ai_model,
    predict_generation_model,
)
from src.application.text.search_detection import find_sentence_source
from src.application.text.text import TextDetector


class NewsVerification:
    def __init__(self):
        """
        Initializes the NewsVerification object.
        """
        self.news_text: str = ""
        self.news_title: str = ""
        self.news_content: str = ""
        self.news_image: str = ""

        self.text = TextDetector()
        self.image = ImageDetector()

        self.news_prediction_label: str = ""
        self.news_prediction_score: float = -1

        # news' urls to find img
        self.found_img_url: list[str] = []

        # Analyzed results
        self.aligned_sentences_df: pd.DataFrame = pd.DataFrame(
            columns=[
                "input",
                "source",
                "label",
                "similarity",
                "paraphrase",
                "url",
                # "entities",
            ],
        )

    def load_news(self, news_title: str, news_content: str, news_image: str):
        """
        Loads news data into the object's attributes.

        Args:
            news_title (str): The title of the news article.
            news_content (str): The content of the news article.
            news_image (str): The url of image in news article.
        """
        # Combine title and content for a full text representation.
        self.news_text = news_title + "\n\n" + news_content

        # if not isinstance(news_title, str) or not isinstance(
        #     news_content,
        #     str,
        # ):
        #     raise TypeError("News title and content must be strings.")

        # if not isinstance(news_image, str) or news_image is not None:
        #     Warning("News image must be a string.")

        self.news_title = news_title
        self.news_content = news_content
        self.news_image = news_image

        self.text.input = self.news_text
        self.image.input = news_image

    def group_by_url(self):
        """
        Groups aligned sentences by URL
        Then, concatenates text the 'input' and 'source' text for each group.
        """

        def concat_text(series):
            """
            Concatenates the elements of a pd.Series into a single string.
            """
            return " ".join(
                series.astype(str).tolist(),
            )  # Handle mixed data types and NaNs

        # Group sentences by URL and concatenate 'input' and 'source' text.
        self.text.grouped_url_df = (
            self.aligned_sentences_df.groupby("url")
            .agg(
                {
                    "input": concat_text,
                    "source": concat_text,
                },
            )
            .reset_index()
        )  # Reset index to make 'url' a regular column

        # Add new columns for label and score
        self.text.grouped_url_df["label"] = None
        self.text.grouped_url_df["score"] = None

        print(f"aligned_sentences_df:\n {self.aligned_sentences_df}")

    def determine_text_origin_by_url(self):
        """
        Determines the text origin for each URL group.
        """
        for index, row in self.text.grouped_url_df.iterrows():
            # Verify text origin using URL-based verification.
            label, score = self.verify_text(row["url"])

            # If URL-based verification returns 'UNKNOWN', use AI detection
            if label == "UNKNOWN":
                # Concatenate text from "input" column in sentence_df
                text = " ".join(row["input"])

                # Detect text origin using an AI model.
                label, score = detect_text_by_ai_model(text)

            print(f"labels = {label}")
            self.text.grouped_url_df.at[index, "label"] = label
            self.text.grouped_url_df.at[index, "score"] = score

    def determine_text_origin(self):
        """
        Determines the origin of the input text by analyzing
            its sources and applying AI detection models.

        This method groups sentences by their source URLs,
            applies verification and AI detection, and then determines
            an overall label and score for the input text.
        """
        # Find the text URLs associated with the input text
        self.find_text_source()

        # Group sentences by URL and concatenate 'input' and 'source' text.
        self.group_by_url()

        # Determine the text origin for each URL group
        self.determine_text_origin_by_url()

        # Determine the overall label and score for the entire input text.
        if not self.text.grouped_url_df.empty:
            # Check for 'gpt-4o' labels in the grouped URLs.
            machine_label = self.text.grouped_url_df[
                self.text.grouped_url_df["label"].str.contains(
                    "gpt-4o",
                    case=False,
                    na=False,
                )
            ]
            print(f" machine_label = {machine_label}")

            if not machine_label.empty:
                # If 'gpt-4o' labels are found, post-process and assign.
                labels = machine_label["label"].tolist()
                label = postprocess_label(labels)

                # labels = " and ".join(machine_label["label"].tolist())
                # label = remove_duplicate_words(label)
                self.text.prediction_label[0] = label
                self.text.prediction_score[0] = machine_label["score"].mean()
            else:
                # If no 'gpt-4o' labels, assign for 'HUMAN' labels.
                machine_label = self.aligned_sentences_df[
                    self.aligned_sentences_df["label"] == "HUMAN"
                ]
                self.text.prediction_label[0] = "HUMAN"
                self.text.prediction_score[0] = self.text.grouped_url_df[
                    "score"
                ].mean()
        else:
            # If no found URLs, use AI detection on the entire input text.
            print("No source found in the input text")
            text = " ".join(self.aligned_sentences_df["input"].tolist())

            # Detect text origin using an AI model.
            label, score = detect_text_by_ai_model(text)
            self.text.prediction_label[0] = label
            self.text.prediction_score[0] = score

    def find_text_source(self):
        """
        Determines the origin of the given text based on paraphrasing
            detection and human authorship analysis.

        1. Splits the input news text into sentences,
        2. Searches for sources for each sentence
        3. Updates the aligned_sentences_df with the found sources.
        """
        print("CHECK TEXT:")
        print("\tFrom search engine:")

        input_paragraphs = split_into_paragraphs(self.news_text)

        # Initialize an empty DataFrame if it doesn't exist,
        # otherwise extend it.
        if (
            not hasattr(self, "aligned_sentences_df")
            or self.aligned_sentences_df is None
        ):
            self.aligned_sentences_df = pd.DataFrame(
                columns=[
                    "input",
                    "source",
                    "label",
                    "similarity",
                    "paraphrase",
                    "url",
                    "entities",
                ],
            )

        # Setup DataFrame for input_sentences
        for _ in range(len(input_paragraphs)):
            self.aligned_sentences_df = pd.concat(
                [
                    self.aligned_sentences_df,
                    pd.DataFrame(
                        [
                            {
                                "input": None,
                                "source": None,
                                "label": None,
                                "similarity": None,
                                "paraphrase": None,
                                "url": None,
                                "entities": None,
                            },
                        ],
                    ),
                ],
                ignore_index=True,
            )

        # Find a source for each sentence
        for index, _ in enumerate(input_paragraphs):
            similarity = self.aligned_sentences_df.loc[index, "similarity"]
            if similarity is not None:
                if similarity > PARAPHRASE_THRESHOLD_MACHINE:
                    continue

            print(f"\n-------index = {index}-------")
            print(f"current_text = {input_paragraphs[index]}\n")

            self.aligned_sentences_df, img_urls = find_sentence_source(
                input_paragraphs,
                index,
                self.aligned_sentences_df,
            )

            # Initialize found_img_url if it does not exist.
            if not hasattr(self, "found_img_url"):
                self.found_img_url = []
            self.found_img_url.extend(img_urls)

    def verify_text(self, url):
        """
        Verifies the text origin based on similarity scores and labels
            associated with a given URL.

        1. Filters sentences by URL and similarity score,
        2. Determines if the text is likely generated by a machine or a human.
        3. Calculates an average similarity score.

        Args:
            url (str): The URL to filter sentences by.

        Returns:
            tuple: A
                - Label ("MACHINE", "HUMAN", or "UNKNOWN")
                - Score
        """
        label = "UNKNOWN"
        score = 0

        # calculate the average similarity when the similary score
        # in each row of sentences_df is higher than 0.8

        # Filter sentences by URL.
        filtered_by_url = self.aligned_sentences_df[
            self.aligned_sentences_df["url"] == url
        ]

        # Filter sentences by similarity score (> PARAPHRASE_THRESHOLD).
        filtered_by_similarity = filtered_by_url[
            filtered_by_url["similarity"] > PARAPHRASE_THRESHOLD
        ]

        # Check if a ratio of remaining filtering-sentences is more than 50%.
        if (
            len(filtered_by_similarity) / len(filtered_by_url)
            > MIN_RATIO_PARAPHRASE_NUM
        ):
            # check if "MACHINE" is in self.aligned_sentences_df["label"]:
            contains_machine = (
                filtered_by_similarity["label"]
                .str.contains(
                    "MACHINE",
                    case=False,
                    na=False,
                )
                .any()
            )
            print(f"contain_machine = \n{contains_machine}")

            # TODO: integrate with determine_text_origin
            if contains_machine:
                # If "MACHINE" label is present, set label and calculate score.
                machine_rows = filtered_by_similarity[
                    filtered_by_similarity["label"].str.contains(
                        "MACHINE",
                        case=False,
                        na=False,
                    )
                ]
                generated_model, _ = predict_generation_model(self.news_text)
                label = f"Partially generated by {generated_model}"
                score = machine_rows["similarity"].mean()
            else:
                # If no "MACHINE" label,
                # assign "HUMAN" label and calculate score.
                label = "HUMAN"
                human_rows = filtered_by_similarity[
                    filtered_by_similarity["label"].str.contains(
                        "HUMAN",
                        case=False,
                        na=False,
                    )
                ]
                score = human_rows["similarity"].mean()

        return label, score

    def determine_image_origin(self):
        """
        Determines the origin of the news image using 3 detection methods.

        1.  Matching against previously found image URLs.
        2.  Reverse image search.
        3.  AI-based image detection.

        If none of these methods succeed, the image origin is "UNKNOWN".
        """
        print("CHECK IMAGE:")

        # Handle the case where no image is provided.
        if self.news_image is None:
            self.image.prediction_label = "UNKNOWN"
            self.image.prediction_score = 0.0
            self.image.referent_url = None
            return

        # Attempt to match the image against previously found image URLs.
        print("\tFrom found image URLs...")
        matched_url, similarity = detect_image_from_news_image(
            self.news_image,
            self.found_img_url,
        )
        if matched_url is not None:
            print(f"matched image: {matched_url}\nsimilarity: {similarity}\n")
            self.image.prediction_label = "HUMAN"
            self.image.prediction_score = similarity
            self.image.referent_url = matched_url
            return

        # Attempt to find the image origin using reverse image search.
        print("\tFrom reverse image search...")
        matched_url, similarity = detect_image_by_reverse_search(
            self.news_image,
        )
        if matched_url is not None:
            print(f"matched image: {matched_url}\tScore: {similarity}%\n")
            self.image.prediction_label = "HUMAN"
            self.image.prediction_score = similarity
            self.image.referent_url = matched_url
            return

        # Attempt to detect the image origin using an AI model.
        print("\tFrom an AI model...")
        detected_label, score = detect_image_by_ai_model(self.news_image)
        if detected_label:
            print(f"detected_label: {detected_label} ({score})")
            self.image.prediction_label = detected_label
            self.image.prediction_score = score
            self.image.referent_url = None
            return

        # If all detection methods fail, mark the image origin as "UNKNOWN".
        self.image.prediction_label = "UNKNOWN"
        self.image.prediction_score = 50
        self.image.referent_url = None

    def determine_origin(self):
        """
        Determine origins by analyzing the news text and image.
        """
        if self.news_text != "":
            self.determine_text_origin()
        if self.news_image != "":
            self.determine_image_origin()

        # Handle entity recognition and processing.
        self.handle_entities()

    def generate_report(self) -> tuple[str, str, str]:
        """
        Generates reports tailored for different user roles
            (ordinary users, fact checkers, governors).

        Returns:
            tuple: A tuple containing three html-formatted reports:
                - ordinary_user_table: Report for ordinary users.
                - fact_checker_table: Report for fact checkers.
                - governor_table: Report for governors.
        """
        ordinary_user_table = create_ordinary_user_table(
            self.aligned_sentences_df,
            self.text,
            self.image,
        )
        fact_checker_table = create_fact_checker_table(
            self.aligned_sentences_df,
            self.text,
            self.image,
        )
        governor_table = create_governor_table(
            self.aligned_sentences_df,
            self.text,
            self.image,
        )

        return ordinary_user_table, fact_checker_table, governor_table

    def handle_entities(self):
        """
        Highlights and assigns entities with colors to aligned sentences
            based on grouped URLs.

        For each grouped URL:
        1. Highlights entities in the input and source text
        2. Then assigns these highlighted entities to the corresponding
            sentences in the aligned sentences DataFrame.
        """

        entities_with_colors = []
        for index, row in self.text.grouped_url_df.iterrows():
            # Get entity-words (in pair) with colors
            entities_with_colors = highlight_entities(
                row["input"],
                row["source"],
            )

            # Assign the highlighted entities to the corresponding sentences
            # in aligned_sentences_df.
            for index, sentence in self.aligned_sentences_df.iterrows():
                if sentence["url"] == row["url"]:
                    # Use .at to modify the DataFrame efficiently.
                    self.aligned_sentences_df.at[index, "entities"] = (
                        entities_with_colors
                    )

    def get_text_urls(self) -> set:
        """
        Returns a set of unique URLs referenced in the text analysis.

        Returns:
            set: A set containing the unique URLs referenced in the text.
        """
        return set(self.text_referent_url)