app.py

import openai
import os
import pdfplumber
from langchain.chains.mapreduce import MapReduceChain
from langchain.text_splitter import CharacterTextSplitter
from langchain.chains.summarize import load_summarize_chain
from langchain.chat_models import ChatOpenAI
from langchain.document_loaders import UnstructuredFileLoader
from langchain.prompts import PromptTemplate
import logging
import json
from typing import List
import mimetypes
import validators
import requests
import tempfile
from bs4 import BeautifulSoup
from langchain.chains import create_extraction_chain
from GoogleNews import GoogleNews
import pandas as pd
import gradio as gr
import re
from langchain.document_loaders import WebBaseLoader
from langchain.chains.llm import LLMChain
from langchain.chains.combine_documents.stuff import StuffDocumentsChain
from transformers import pipeline
import plotly.express as px

class KeyValueExtractor:

    def __init__(self):

        """
        Initialize the ContractSummarizer object.

        Parameters:
            pdf_file_path (str): The path to the input PDF file.
        """
        self.model = "facebook/bart-large-mnli"

    def get_news(self,keyword):

        googlenews = GoogleNews(lang='en', region='US', period='1d', encode='utf-8')
        googlenews.clear()
        googlenews.search(keyword)
        googlenews.get_page(2)
        news_result = googlenews.result(sort=True)
        news_data_df = pd.DataFrame.from_dict(news_result)

        news_data_df.info()

        # Display header of dataframe.
        news_data_df.head()

        tot_news_link = []
        for index, headers in news_data_df.iterrows():
          news_link = str(headers['link'])
          tot_news_link.append(news_link)

        return tot_news_link

    def url_format(self,urls):

        tot_url_links = []
        for url_text in urls:
            # Define a regex pattern to match URLs starting with 'http' or 'https'
            pattern = r'(https?://[^\s]+)'

            # Search for the URL in the text using the regex pattern
            match = re.search(pattern, url_text)

            if match:
                extracted_url = match.group(1)
                tot_url_links.append(extracted_url)

            else:
                print("No URL found in the given text.")

        return tot_url_links

    def clear_error_ulr(self,urls):
        error_url = []
        for url in urls:
                  if validators.url(url):
                      headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36',}
                      r = requests.get(url,headers=headers)
                      if r.status_code != 200:
                          # raise ValueError("Check the url of your file; returned status code %s" % r.status_code)
                                      print(f"Error fetching {url}:")
                                      error_url.append(url)
                                      continue
        cleaned_list_url = [item for item in urls if item not in error_url]
        return cleaned_list_url

    def get_each_link_summary(self,urls):

      each_link_summary = ""

      for url in urls:
        loader = WebBaseLoader(url)
        docs = loader.load()
        text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
        chunk_size=3000, chunk_overlap=200
          )

        # Split the documents into chunks
        split_docs = text_splitter.split_documents(docs)

        # Prepare the prompt template for summarization
        prompt_template = """Write a concise summary of the following:
        {text}
        CONCISE SUMMARY:"""
        prompt = PromptTemplate.from_template(prompt_template)

        # Prepare the template for refining the summary with additional context
        refine_template = (
            "Your job is to produce a final summary\n"
            "We have provided an existing summary up to a certain point: {existing_answer}\n"
            "We have the opportunity to refine the existing summary"
            "(only if needed) with some more context below.\n"
            "------------\n"
            "{text}\n"
            "------------\n"
            "Given the new context, refine the original summary"
            "If the context isn't useful, return the original summary."
        )
        refine_prompt = PromptTemplate.from_template(refine_template)

        # Load the summarization chain using the ChatOpenAI language model
        chain = load_summarize_chain(
            llm = ChatOpenAI(temperature=0),
            chain_type="refine",
            question_prompt=prompt,
            refine_prompt=refine_prompt,
            return_intermediate_steps=True,
            input_key="input_documents",
            output_key="output_text",
        )

        # Generate the refined summary using the loaded summarization chain
        result = chain({"input_documents": split_docs}, return_only_outputs=True)
        print(result["output_text"])

        # Return the refined summary
        each_link_summary = each_link_summary + result["output_text"]

      return each_link_summary

    def save_text_to_file(self,each_link_summary) -> str:

        """
        Load the text from the saved file and split it into documents.

        Returns:
            List[str]: List of document texts.
        """

        # Get the path to the text file where the extracted text will be saved
        file_path = "extracted_text.txt"
        try:
            with open(file_path, 'w') as file:
                # Write the extracted text into the text file
                file.write(each_link_summary)
            # Return the file path of the saved text file
            return file_path
        except IOError as e:
            # If an IOError occurs during the file saving process, log the error
            logging.error(f"Error while saving text to file: {e}")

    def document_loader(self,file_path) -> List[str]:

        """
        Load the text from the saved file and split it into documents.

        Returns:
            List[str]: List of document texts.
        """

        # Initialize the UnstructuredFileLoader
        loader = UnstructuredFileLoader(file_path, strategy="fast")
        # Load the documents from the file
        docs = loader.load()

        # Return the list of loaded document texts
        return docs

    def document_text_spilliter(self,docs) -> List[str]:

        """
        Split documents into chunks for efficient processing.

        Returns:
            List[str]: List of split document chunks.
        """

        # Initialize the text splitter with specified chunk size and overlap
        text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
            chunk_size=3000, chunk_overlap=200
        )

        # Split the documents into chunks
        split_docs = text_splitter.split_documents(docs)

        # Return the list of split document chunks
        return split_docs

    def extract_key_value_pair(self,content) -> None:

        """
        Extract key-value pairs from the refined summary.

        Prints the extracted key-value pairs.
        """

        try:

          # Use OpenAI's Completion API to analyze the text and extract key-value pairs
          response = openai.Completion.create(
              engine="text-davinci-003",  # You can choose a different engine as well
              temperature = 0,
              prompt=f"Get maximum count meaningfull key value pairs. content in backticks.```{content}```.",
              max_tokens=1000 # You can adjust the length of the response
          )

          # Extract and return the chatbot's reply
          result = response['choices'][0]['text'].strip()
          return result
        except Exception as e:
            # If an error occurs during the key-value extraction process, log the error
            logging.error(f"Error while extracting key-value pairs: {e}")
            print("Error:", e)

    def refine_summary(self,split_docs) -> str:

        """
        Refine the summary using the provided context.

        Returns:
            str: Refined summary.
        """

        # Prepare the prompt template for summarization
        prompt_template = """Write a detalied broad abractive summary of the following:
        {text}
        CONCISE SUMMARY:"""
        prompt = PromptTemplate.from_template(prompt_template)

        # Prepare the template for refining the summary with additional context
        refine_template = (
            "Your job is to produce a final summary\n"
            "We have provided an existing summary up to a certain point: {existing_answer}\n"
            "We have the opportunity to refine the existing summary"
            "(only if needed) with some more context below.\n"
            "------------\n"
            "{text}\n"
            "------------\n"
            "Given the new context, refine the original summary"
            "If the context isn't useful, return the original summary."
        )
        refine_prompt = PromptTemplate.from_template(refine_template)

        # Load the summarization chain using the ChatOpenAI language model
        chain = load_summarize_chain(
            llm = ChatOpenAI(temperature=0),
            chain_type="refine",
            question_prompt=prompt,
            refine_prompt=refine_prompt,
            return_intermediate_steps=True,
            input_key="input_documents",
            output_key="output_text",
        )

        # Generate the refined summary using the loaded summarization chain
        result = chain({"input_documents": split_docs}, return_only_outputs=True)

        key_value_pair = self.extract_key_value_pair(result["output_text"])

        # Return the refined summary
        return result["output_text"],key_value_pair

    def analyze_sentiment_for_graph(self, text):
        pipe = pipeline("zero-shot-classification", model=self.model)
        label=["Positive", "Negative", "Neutral"]
        result = pipe(text, label)
        sentiment_scores = {
            result['labels'][0]: result['scores'][0],
            result['labels'][1]: result['scores'][1],
            result['labels'][2]: result['scores'][2]
        }
        return sentiment_scores

    def display_graph(self,text):

        sentiment_scores = self.analyze_sentiment_for_graph(text)
        labels = sentiment_scores.keys()
        scores = sentiment_scores.values()
        fig = px.bar(x=scores, y=labels, orientation='h', color=labels, color_discrete_map={"Negative": "red", "Positive": "green", "Neutral": "gray"})
        fig.update_traces(texttemplate='%{x:.2f}%', textposition='outside')
        fig.update_layout(title="Sentiment Analysis",width=800)

        formatted_pairs = []
        for key, value in sentiment_scores.items():
            formatted_value = round(value, 2)  # Round the value to two decimal places
            formatted_pairs.append(f"{key} : {formatted_value}")

        result_string = '\t'.join(formatted_pairs)

        return fig

    def main(self,keyword):

      urls = self.get_news(keyword)
      tot_urls = self.url_format(urls)
      clean_url = self.clear_error_ulr(tot_urls)
      each_link_summary  =  self.get_each_link_summary(clean_url)
      file_path = self.save_text_to_file(each_link_summary)
      docs = self.document_loader(file_path)
      split_docs = self.document_text_spilliter(docs)
      result = self.refine_summary(split_docs)


      return result

    def gradio_interface(self):

        with gr.Blocks(css="style.css",theme= 'karthikeyan-adople/hudsonhayes-gray') as app:
              gr.HTML("""<center class="darkblue" style='background-color:rgb(0,1,36); text-align:center;padding:25px;'><center><h1 class ="center">
                <img src="file=logo.png" height="110px" width="280px"></h1></center>
                  <br><h1 style="color:#fff">summarizer</h1></center>""")
              with gr.Row(elem_id="col-container"):
                with gr.Column(scale=1.0, min_width=150, ):
                  input_news = gr.Textbox(label="NEWS")
              with gr.Row(elem_id="col-container"):
                with gr.Column(scale=1.0, min_width=150):
                  analyse = gr.Button("Analyse")
              with gr.Row(elem_id="col-container"):
                with gr.Column(scale=0.50, min_width=150):
                  result_summary = gr.Textbox(label="Summary")
                with gr.Column(scale=0.50, min_width=150):
                  key_value_pair_result = gr.Textbox(label="Key Value Pair")
              with gr.Row(elem_id="col-container"):
                  with gr.Column(scale=0.70, min_width=0):
                      plot =gr.Plot(label="Customer", size=(500, 600))
              with gr.Row(elem_id="col-container"):
                with gr.Column(scale=1.0, min_width=150):
                  analyse_sentiment = gr.Button("Analyse")

              analyse.click(self.main, input_news, [result_summary,key_value_pair_result])
              analyse_sentiment.click(self.display_graph,result_summary,[plot])

        app.launch(debug=True)

if __name__ == "__main__":

  text_process = KeyValueExtractor()
  text_process.gradio_interface()