pre-commit

gpt_test.py

@@ -1,28 +1,35 @@
+import csv
 import os
 
 from dotenv import load_dotenv
-from openai import AzureOpenAI, OpenAIError
+from openai import (
+    AzureOpenAI,
+    OpenAIError,
+)
 
 
-import csv
-
 def get_first_column(csv_filepath):
     """
-    Reads a CSV file with a header and returns a list containing only the 
+    Reads a CSV file with a header and returns a list containing only the
     values from the first column.
 
     Args:
         csv_filepath: The path to the CSV file.
 
     Returns:
-        A list of strings, where each string is a value from the first 
-        column of the CSV file.  Returns an empty list if there's an error 
-        opening or reading the file, or if the file has no rows after the header.
+        A list of strings, where each string is a value from the first
+        column of the CSV file.
+        Returns an empty list if there's an error opening or reading
+        the file, or if the file has no rows after the header.
         Prints an error message to the console in case of file errors.
     """
     first_column_values = []
     try:
-        with open(csv_filepath, 'r', newline='', encoding='utf-8') as csvfile:  # Handle potential encoding issues
+        with open(
+            csv_filepath,
+            newline="",
+            encoding="utf-8",
+        ) as csvfile:  # Handle potential encoding issues
             reader = csv.reader(csvfile)
             next(reader, None)  # Skip the header row (if it exists)
 
@@ -32,21 +39,29 @@ def get_first_column(csv_filepath):
 
     except FileNotFoundError:
         print(f"Error: File not found at {csv_filepath}")
-    except Exception as e: # Catch other potential errors (e.g., UnicodeDecodeError)
+    except (
+        Exception
+    ) as e:  # Catch other potential errors (e.g., UnicodeDecodeError)
         print(f"An error occurred: {e}")
-    
+
     return first_column_values
 
+
 def add_text_to_csv(csv_filepath, text_to_add, index=0):
     """
     Adds text to a single-column CSV file (UTF-8 encoding).
 
     Args:
         csv_filepath: The path to the CSV file.
-        text_to_add: The text to append to the CSV file (one value per new row).
+        text_to_add: The text to append to CSV file (one value per new row).
     """
     try:
-        with open(csv_filepath, 'a', newline='', encoding='utf-8') as csvfile:  # 'a' for append mode
+        with open(
+            csv_filepath,
+            "a",
+            newline="",
+            encoding="utf-8",
+        ) as csvfile:  # 'a' for append mode
             writer = csv.writer(csvfile)
 
             # Check if file is empty to determine if header needs to be written
@@ -58,13 +73,18 @@ def add_text_to_csv(csv_filepath, text_to_add, index=0):
 
             if isinstance(text_to_add, list):  # Check if text_to_add is a list
                 for text_item in text_to_add:
-                    writer.writerow([index, text_item]) # Write text_item as a single-element row
+                    writer.writerow(
+                        [index, text_item],
+                    )  # Write text_item as a single-element row
             else:  # If not a list, assume it's a single string
-                writer.writerow([index, text_to_add])  # Write text_to_add as a single-element row
+                writer.writerow(
+                    [index, text_to_add],
+                )  # Write text_to_add as a single-element row
 
     except Exception as e:
         print(f"An error occurred: {e}")
 
+
 load_dotenv()
 AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
 AZURE_OPENAI_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT")
@@ -76,25 +96,25 @@ azure_client = AzureOpenAI(
     api_version="2024-05-01-preview",
 )
 
-deplopment_name = "gpt-4o-mini" # "o1-mini"  # or "gpt-4o"
+deplopment_name = "gpt-4o-mini"  # "o1-mini"  # or "gpt-4o"
 TEXT_PROMPT = """
 Paraphrase the following news, only output the paraphrased text:
 
 """
 text = get_first_column("data/MAGE.csv")
-count  = 0
+count = 0
 for index, news in enumerate(text):
     if count > 1000:
         break
     prompt = TEXT_PROMPT + news
     print(f"{index:5}:\t{news[:50]}")
-    #print(f"{index:5}:\t{prompt}")
-    
+    # print(f"{index:5}:\t{prompt}")
+
     try:
         response = azure_client.chat.completions.create(
             model=deplopment_name,  # model = "deployment_name".
             messages=[
-                # {"role": "system", "content": "You are a helpful assistant."},
+                # {"role": "system", "content": "You're an assistant."},
                 {"role": "user", "content": prompt},
             ],
             # max_tokens=512,
@@ -103,8 +123,8 @@ for index, news in enumerate(text):
     except OpenAIError as e:
         print(f"Error interacting with OpenAI API: {e}")
         continue
-    
+
     count += 1
     paraphrased_news = response.choices[0].message.content
-    
+
     add_text_to_csv("data/MAGE_4o_mini.csv", paraphrased_news, count)

src/application/content_detection.py

@@ -1,6 +1,5 @@
 from difflib import SequenceMatcher
 
-import numpy as np
 import pandas as pd
 
 from src.application.image.image_detection import (
@@ -13,7 +12,10 @@ from src.application.text.entity import (
     highlight_entities,
 )
 from src.application.text.helper import extract_equal_text
-from src.application.text.model_detection import detect_text_by_ai_model, predict_generation_model
+from src.application.text.model_detection import (
+    detect_text_by_ai_model,
+    predict_generation_model,
+)
 from src.application.text.preprocessing import split_into_paragraphs
 from src.application.text.search_detection import (
     PARAPHRASE_THRESHOLD_MACHINE,
@@ -30,17 +32,17 @@ class NewsVerification:
 
         self.text_prediction_label: list[str] = ["UNKNOWN"]
         self.text_prediction_score: list[float] = [0.0]
-        
+
         self.image_prediction_label: list[str] = ["UNKNOWN"]
         self.image_prediction_score: list[str] = [0.0]
         self.image_referent_url: list[str] = []
-        
+
         self.news_prediction_label = ""
         self.news_prediction_score = -1
 
         # news' urls to find img
         self.found_img_url: list[str] = []
-        
+
         # Analyzed results
         self.aligned_paragraphs_df: pd.DataFrame = pd.DataFrame(
             columns=[
@@ -69,24 +71,26 @@ class NewsVerification:
 
     def determine_text_origin(self):
         self.find_text_source()
-        
+
         # Group inout and source by url
         def concat_text(series):
-            return ' '.join(series.astype(str).tolist()) #Handle mixed data types and NaNs
-        
-        self.grouped_url_df = self.aligned_paragraphs_df.groupby('url').agg(
+            return " ".join(
+                series.astype(str).tolist(),
+            )  # Handle mixed data types and NaNs
+
+        self.grouped_url_df = self.aligned_paragraphs_df.groupby("url").agg(
             {
-                'input': concat_text,
-                'source': concat_text,
-                }
-            )
+                "input": concat_text,
+                "source": concat_text,
+            },
+        )
         self.grouped_url_df = self.grouped_url_df.reset_index()
         # Add new columns for label and score
         self.grouped_url_df["label"] = None
         self.grouped_url_df["score"] = None
-        
+
         print(f"aligned_paragraphs_df:\n {self.aligned_paragraphs_df}")
-        
+
         for index, row in self.grouped_url_df.iterrows():
             label, score = self.verify_text(row["url"])
             if label == "UNKNOWN":
@@ -95,18 +99,21 @@ class NewsVerification:
 
                 # detect by baseline model
                 label, score = detect_text_by_ai_model(text)
-                
+
             self.grouped_url_df.at[index, "label"] = label
             self.grouped_url_df.at[index, "score"] = score
 
         # Overall label or score for the whole input text
         if len(self.grouped_url_df) > 0:
-            # filter self.aligned_paragraphs_df["label"] if inclucind substring MACHINE
             machine_label = self.grouped_url_df[
-                self.grouped_url_df["label"].str.contains("MACHINE", case=False, na=False)
+                self.grouped_url_df["label"].str.contains(
+                    "MACHINE",
+                    case=False,
+                    na=False,
+                )
             ]
             # machine_label = self.aligned_paragraphs_df[
-            #     self.aligned_paragraphs_df["label"] == "MACHINE" 
+            #     self.aligned_paragraphs_df["label"] == "MACHINE"
             # ]
             if len(machine_label) > 0:
                 label = " ".join(machine_label["label"].tolist())
@@ -118,14 +125,14 @@ class NewsVerification:
                 ]
                 self.text_prediction_label[0] = "HUMAN"
                 self.text_prediction_score[0] = machine_label["score"].mean()
-        else:  # no source found in the input text 
+        else:  # no source found in the input text
             print("No source found in the input text")
             text = " ".join(self.aligned_paragraphs_df["input"].tolist())
             # detect by baseline model
-            label, score = detect_text_by_ai_model(text)      
+            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
@@ -148,15 +155,22 @@ class NewsVerification:
 
         for _ in range(len(input_sentences)):
             self.aligned_paragraphs_df = pd.concat(
-                [self.aligned_paragraphs_df, pd.DataFrame([{
-                    "input": None,
-                    "source": None,
-                    "label": None,
-                    "similarity": None,
-                    "paraphrase": None,
-                    "url": None,
-                    "entities": None,
-                }])],
+                [
+                    self.aligned_paragraphs_df,
+                    pd.DataFrame(
+                        [
+                            {
+                                "input": None,
+                                "source": None,
+                                "label": None,
+                                "similarity": None,
+                                "paraphrase": None,
+                                "url": None,
+                                "entities": None,
+                            },
+                        ],
+                    ),
+                ],
                 ignore_index=True,
             )
 
@@ -183,7 +197,8 @@ class NewsVerification:
     def verify_text(self, url):
         label = "UNKNOWN"
         score = 0
-        # calculate the average similarity when the similary score in each row of sentences_df is higher than 0.8
+        # calculate the average similarity when the similary score
+        # in each row of sentences_df is higher than 0.8
         filtered_by_url = self.aligned_paragraphs_df[
             self.aligned_paragraphs_df["url"] == url
         ]
@@ -192,17 +207,24 @@ class NewsVerification:
         ]
         if len(filtered_by_similarity) / len(self.aligned_paragraphs_df) > 0.5:
             # check if "MACHINE" is in self.aligned_sentences_df["label"]:
-            contains_machine = filtered_by_similarity["label"].str.contains(
-                "MACHINE", case=False, na=False
-                ).any()
+            contains_machine = (
+                filtered_by_similarity["label"]
+                .str.contains(
+                    "MACHINE",
+                    case=False,
+                    na=False,
+                )
+                .any()
+            )
             if contains_machine:
                 label = "MACHINE"
                 machine_rows = filtered_by_similarity[
                     filtered_by_similarity["label"].str.contains(
                         "MACHINE",
                         case=False,
-                        na=False)
-                    ]
+                        na=False,
+                    )
+                ]
                 generated_model, _ = predict_generation_model(self.news_text)
                 label += f"<br>({generated_model})"
                 score = machine_rows["similarity"].mean()
@@ -212,12 +234,12 @@ class NewsVerification:
                     filtered_by_similarity["label"].str.contains(
                         "HUMAN",
                         case=False,
-                        na=False)
-                    ]
+                        na=False,
+                    )
+                ]
                 score = human_rows["similarity"].mean()
-        
+
         return label, score
-        
 
     def determine_image_origin(self):
         print("CHECK IMAGE:")
@@ -267,14 +289,14 @@ class NewsVerification:
             self.determine_image_origin()
 
     def analyze_details(self):
-        self.handle_entities()    
+        self.handle_entities()
         ordinary_user_table = self.create_ordinary_user_table()
         fact_checker_table = self.create_fact_checker_table()
         governor_table = self.create_governor_table()
 
         return ordinary_user_table, fact_checker_table, governor_table
-    
-    def handle_entities(self):        
+
+    def handle_entities(self):
         entities_with_colors = []
         for index, row in self.grouped_url_df.iterrows():
             # Get entity-words (in pair) with colors
@@ -283,12 +305,11 @@ class NewsVerification:
                 row["source"],
             )
 
-            #self.grouped_url_df.at[index, "entities"] = entities_with_colors  # must use at
-            
             for index, paragraph in self.aligned_paragraphs_df.iterrows():
                 if paragraph["url"] == row["url"]:
-                    self.aligned_paragraphs_df.at[index, "entities"] = entities_with_colors # must use at
-
+                    self.aligned_paragraphs_df.at[index, "entities"] = (
+                        entities_with_colors  # must use at
+                    )
 
     def get_text_urls(self):
         return set(self.text_referent_url)
@@ -336,13 +357,13 @@ class NewsVerification:
         rows.append(self.format_image_fact_checker_row(max_length))
 
         for _, row in self.aligned_paragraphs_df.iterrows():
-            if row["input"] == None:
+            if row["input"] is None:
                 continue
-            
-            if row["source"] == None:
+
+            if row["source"] is None:
                 equal_idx_1 = equal_idx_2 = []
-            
-            else: # Get index of equal phrases in input and source sentences
+
+            else:  # Get index of equal phrases in input and source sentences
                 equal_idx_1, equal_idx_2 = extract_equal_text(
                     row["input"],
                     row["source"],
@@ -354,33 +375,42 @@ class NewsVerification:
                     equal_idx_1,
                     equal_idx_2,
                     row["entities"],
-                    row["url"]
+                    row["url"],
                 ],
             )
-        
+
         previous_url = None
         span_row = 1
-        for index, row in enumerate(self.fact_checker_table): 
+        for index, row in enumerate(self.fact_checker_table):
             current_url = row[4]
             last_url_row = False
-            
+
             # First row or URL change
             if index == 0 or current_url != previous_url:
                 first_url_row = True
                 previous_url = current_url
                 # Increase counter  "span_row" when the next url is the same
-                while index + span_row < len(self.fact_checker_table) \
-                    and self.fact_checker_table[index + span_row][4] == current_url:
+                while (
+                    index + span_row < len(self.fact_checker_table)
+                    and self.fact_checker_table[index + span_row][4]
+                    == current_url
+                ):
                     span_row += 1
-                
+
             else:
                 first_url_row = False
                 span_row -= 1
-                
+
             if span_row == 1:
                 last_url_row = True
-                
-            formatted_row = self.format_text_fact_checker_row(row, first_url_row, last_url_row, span_row, max_length)
+
+            formatted_row = self.format_text_fact_checker_row(
+                row,
+                first_url_row,
+                last_url_row,
+                span_row,
+                max_length,
+            )
             rows.append(formatted_row)
 
         table = "\n".join(rows)
@@ -436,7 +466,7 @@ class NewsVerification:
                 source_sentence = row[0]["source"]
                 highlight_idx_input = []
                 highlight_idx_source = []
-                
+
             if row[3] is not None:
                 entity_count = len(row[3])
 
@@ -453,7 +483,7 @@ class NewsVerification:
             )  # text, index of highlight words
 
             # Replace _ to get correct formatting
-            # Original one having _ for correct word counting 
+            # Original one having _ for correct word counting
             input_sentence = input_sentence.replace(
                 "span_style",
                 "span style",
@@ -468,24 +498,22 @@ class NewsVerification:
 
         url = row[0]["url"]
         # Displayed label and score by url
-        filterby_url = self.grouped_url_df[
-            self.grouped_url_df["url"] == url
-        ]
+        filterby_url = self.grouped_url_df[self.grouped_url_df["url"] == url]
         if len(filterby_url) > 0:
             label = filterby_url["label"].values[0]
             score = filterby_url["score"].values[0]
-        else:        
+        else:
             label = self.text_prediction_label[0]
             score = self.text_prediction_score[0]
 
         # Format displayed url
-        
+
         short_url = self.shorten_url(url, max_length)
         source_text_url = f"""<a href="{url}">{short_url}</a>"""
 
         # Format displayed entity count
         entity_count_text = self.get_entity_count_text(entity_count)
-        
+
         border_top = "border-top: 1px solid transparent;"
         border_bottom = "border-bottom: 1px solid transparent;"
         if first_url_row is True:
@@ -580,7 +608,7 @@ class NewsVerification:
         source_text_urls = ""
         urls = []
         for _, row in self.aligned_paragraphs_df.iterrows():
-            if row["input"] == None:
+            if row["input"] is None:
                 continue
             input_sentences += row["input"] + "<br><br>"
             url = row["url"]
@@ -620,13 +648,13 @@ class NewsVerification:
         rows.append(self.format_image_governor_row(max_length))
 
         for _, row in self.aligned_paragraphs_df.iterrows():
-            if row["input"] == None:
+            if row["input"] is None:
                 continue
-                
-            if row["source"] == None:
+
+            if row["source"] is None:
                 equal_idx_1 = equal_idx_2 = []
-            
-            else:    
+
+            else:
                 # Get index of equal phrases in input and source sentences
                 equal_idx_1, equal_idx_2 = extract_equal_text(
                     row["input"],
@@ -680,19 +708,25 @@ class NewsVerification:
             if row[0]["input"] is None:
                 continue
 
-            if row[0]["source"] is not None and row[3] is not None:  # source is not empty
+            if (
+                row[0]["source"] is not None and row[3] is not None
+            ):  # source is not empty
                 # highlight entities
                 input_sentence, highlight_idx_input = apply_highlight(
                     row[0]["input"],
                     row[3],  # entities_with_colors
                     "input",  # key
-                    entity_count[-2],  # since the last one is for current counting
+                    entity_count[
+                        -2
+                    ],  # since the last one is for current counting
                 )
                 source_sentence, highlight_idx_source = apply_highlight(
                     row[0]["source"],
                     row[3],  # entities_with_colors
                     "source",  # key
-                    entity_count[-2],  # since the last one is for current counting
+                    entity_count[
+                        -2
+                    ],  # since the last one is for current counting
                 )
 
                 # Color overlapping words
@@ -722,12 +756,11 @@ class NewsVerification:
                 else:
                     source_sentence = row[0]["source"]
                 input_sentence = row[0]["input"]
-                
 
             # convert score to HUMAN-based score:
             input_sentences += input_sentence + "<br><br>"
             source_sentences += source_sentence + "<br><br>"
-            
+
             url = row[0]["url"]
             if url not in urls:
                 urls.append(url)
@@ -736,7 +769,7 @@ class NewsVerification:
                 sentence_count += 1
                 if row[3] is not None:
                     entity_count.append(len(row[3]))
-        
+
         entity_count_text = self.get_entity_count_text(sum(entity_count))
 
         return f"""
@@ -791,7 +824,7 @@ class NewsVerification:
 
         starts, ends = self.extract_starts_ends(colored_idx)
         starts, ends = self.filter_indices(starts, ends, highlighted_idx)
-        
+
         previous_end = 0
         for start, end in zip(starts, ends):
             paragraph += " ".join(words[previous_end:start])
@@ -892,4 +925,4 @@ class NewsVerification:
                 starts.append(start)
                 ends.append(end)
 
-        return starts, ends
+        return starts, ends

src/application/text/entity.py

@@ -161,7 +161,7 @@ def assign_colors_to_entities(entities):
 
 
 def highlight_entities(text1, text2):
-    if text1 == None or text2 == None:
+    if text1 is None or text2 is None:
         return None
 
     entities_text = extract_entities_gpt(text1, text2)

src/application/text/helper.py

@@ -147,7 +147,7 @@ def extract_equal_text(text1, text2):
         text = text.lower()
         text = text.translate(str.maketrans("", "", string.punctuation))
         return text
-    
+
     splited_text1 = cleanup(text1).split()
     splited_text2 = cleanup(text2).split()
 
@@ -163,7 +163,8 @@ def extract_equal_text(text1, text2):
             equal_idx_2.append({"start": j1, "end": j2})
             # subtext_1 = " ".join(text1[i1:i2])
             # subtext_2 = " ".join(text2[j1:j2])
-            # print(f'{tag:7}   a[{i1:2}:{i2:2}] --> b[{j1:2}:{j1:2}] {subtext_1!r:>55} --> {subtext_2!r}')
+            # print(f'{tag:7}   a[{i1:2}:{i2:2}] --> b[{j1:2}:{j1:2}]
+            # {subtext_1!r:>55} --> {subtext_2!r}')
     return equal_idx_1, equal_idx_2
 
 

src/application/text/model_detection.py

@@ -1,11 +1,16 @@
-from transformers import pipeline
 import os
 
-from dotenv import load_dotenv
-from openai import AzureOpenAI, OpenAIError
-from sentence_transformers import SentenceTransformer, util
 import torch
-
+from dotenv import load_dotenv
+from openai import (
+    AzureOpenAI,
+    OpenAIError,
+)
+from sentence_transformers import (
+    SentenceTransformer,
+    util,
+)
+from transformers import pipeline
 
 load_dotenv()
 AZURE_OPENAI_API_KEY = os.getenv("AZURE_OPENAI_API_KEY")
@@ -74,7 +79,7 @@ def detect_text_by_ai_model(
         return UNKNOWN, 0.5  # Return UNKNOWN and 0.0 confidence if error
 
 
-def predict_generation_model(text:str) -> tuple[str, float]:
+def predict_generation_model(text: str) -> tuple[str, float]:
     """
     Predicts if text is generated by gpt-4o or gpt-4o-mini models.
     Compare the input text against the paraphrased text by the models.
@@ -94,7 +99,7 @@ def predict_generation_model(text:str) -> tuple[str, float]:
         if similarity > best_similarity:
             best_similarity = similarity
             best_model = model
-    
+
     return best_model, best_similarity
 
 
@@ -125,8 +130,9 @@ Paraphrase the following news, only output the paraphrased text:
         return paraphrased_text
     except OpenAIError as e:  # Add exception handling
         print(f"Error in AI model inference: {e}")
-        return None 
-    
+        return None
+
+
 def measure_text_similarity(text1: str, text2: str) -> float:
     """
     Measure the similarity between two texts.
@@ -151,4 +157,3 @@ def measure_text_similarity(text1: str, text2: str) -> float:
     similarity = util.cos_sim(embeddings1, embeddings2).cpu().numpy()
     print(similarity[0][0])
     return similarity[0][0]
-

src/application/text/search_detection.py

@@ -75,23 +75,26 @@ def find_paragraph_source(text, text_index, sentences_df):
                 )
 
                 if aligned_sentence["paraphrase"] is False:
-                    sentences_df.loc[text_index, "input"] = aligned_sentence["input"]
-                    sentences_df.loc[text_index, "paraphrase"] = aligned_sentence["paraphrase"]
+                    sentences_df.loc[text_index, "input"] = aligned_sentence[
+                        "input"
+                    ]
+                    sentences_df.loc[text_index, "paraphrase"] = (
+                        aligned_sentence["paraphrase"]
+                    )
                     return sentences_df, []
-                
+
                 # assign values
                 columns = [
                     "input",
-                    "source", 
-                    "label", 
-                    "similarity", 
-                    "paraphrase", 
+                    "source",
+                    "label",
+                    "similarity",
+                    "paraphrase",
                     "url",
-                    ]
+                ]
                 for c in columns:
                     if c in sentences_df.columns:
                         sentences_df.loc[text_index, c] = aligned_sentence[c]
-                
 
                 for idx, _ in sentences_df.iterrows():
                     similarity = sentences_df.loc[idx, "similarity"]
@@ -106,12 +109,20 @@ def find_paragraph_source(text, text_index, sentences_df):
                         url,
                     )
 
-                    if similarity is None or \
-                        aligned_sentence["similarity"] > similarity:
-                            columns = ["input", "source", "label", "similarity", "url"]
-                            for c in columns:
-                                if c in sentences_df.columns:
-                                    sentences_df.loc[idx, c] = aligned_sentence[c]
+                    if (
+                        similarity is None
+                        or aligned_sentence["similarity"] > similarity
+                    ):
+                        columns = [
+                            "input",
+                            "source",
+                            "label",
+                            "similarity",
+                            "url",
+                        ]
+                        for c in columns:
+                            if c in sentences_df.columns:
+                                sentences_df.loc[idx, c] = aligned_sentence[c]
                 return sentences_df, content.images
 
     sentences_df.loc[text_index, "input"] = text[text_index]
@@ -266,7 +277,7 @@ def check_paraphrase(input_text, page_text, url):
 
         label, is_paraphrased = determine_label(max_similarity)
         best_matched_paragraph = page_paragraphs[max_sim_index]
-        
+
         alignment = {
             "input": paragraph,
             "source": best_matched_paragraph,
@@ -317,6 +328,7 @@ def check_human(alligned_sentences):
         return True
     return False
 
+
 def determine_label(similarity):
     if similarity >= PARAPHRASE_THRESHOLD_HUMAN:
         return "HUMAN", True

test.py

@@ -1,2 +1,2 @@
 my_list = [0, 0]
-print(my_list[-2])
+print(my_list[-2])