app.py

import logging
from typing import List
from pydantic import NoneStr
import os
from langchain.chains.question_answering import load_qa_chain
from langchain.document_loaders import UnstructuredFileLoader
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.llms import OpenAI
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import FAISS
import gradio as gr
import openai
from langchain import PromptTemplate, OpenAI, LLMChain
import validators
import requests
import mimetypes
import tempfile
import pandas as pd
import re

class ChemicalIdentifier:
    def __init__(self):

        openai.api_key = os.getenv("OPENAI_API_KEY") 
        self.logger = logging.getLogger("ChemicalIdentifier")
        self.logger.setLevel(logging.DEBUG)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.DEBUG)
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)


    def upload_via_url(self,url:str)->List:
        """
        Uploads a file from a given URL and returns the loaded document.

        Args:
            url (str): The URL of the file to be uploaded.

        Returns:
            Document: The loaded document.

        Raises:
            ValueError: If the URL is not valid or the file cannot be fetched.
        """

        try:
          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
                  )

              content_type = r.headers.get("content-type")
              file_extension = mimetypes.guess_extension(content_type)
              temp_file = tempfile.NamedTemporaryFile(suffix=file_extension, delete=False)
              temp_file.write(r.content)
              file_path = temp_file.name
              loader = UnstructuredFileLoader(file_path, strategy="fast")
              docs = loader.load()
              return docs
          else:
              raise ValueError("Please enter a valid URL")
        except Exception as e:
             self.logger.error("Error occurred while uploading the file: %s", str(e))
             raise ValueError("Error occurred while uploading the file") from e


    def find_chemicals(self,text:str)->str:
        """
        Extracts chemical names from the given text.

        Args:
            text (str): The text to extract chemical names from.

        Returns:
            str: The extracted chemical names in bullet form.

        Raises:
            ValueError: If an error occurs during the extraction process.
        """

        try:
          prompt = f"List out only all the Chemicals Names in the give text in bullet form.{text}"
          response = openai.Completion.create(
              model="text-davinci-003",
              prompt=prompt,
              temperature=0,
              max_tokens=500,
              top_p=1,
              frequency_penalty=0,
              presence_penalty=0,
          )

          message = response.choices[0].text.strip()
          if ":" in message:
              message = re.sub(r'^.*:', '', message)
          return message.strip()
        except Exception as e:
            self.logger.error("Error occurred while finding chemicals: %s", str(e))
            raise ValueError("Error occurred while finding chemicals") from e


    def get_chemicals(self,urls:str)->str:
        """
        Retrieves chemicals from the provided URLs.

        Args:
            urls (str): Comma-separated URLs of the files to be processed.

        Returns:
            str: The extracted chemical names.

        Raises:
            ValueError: If an error occurs during the process.
        """

        try:
          total_chemical=[]
          for url in urls.split(','):
            webpage_text = self.upload_via_url(url)
            chemicals = self.find_chemicals(webpage_text)
            total_chemical.append(chemicals)
          list_of_chemicals = "".join(total_chemical)
          return list_of_chemicals

        except Exception as e:
            self.logger.error("Error occurred while getting chemicals from URLs: %s", str(e))
            raise ValueError("Error occurred while getting chemicals from URLs") from e

    def get_empty_state(self):

        """ Create empty Knowledge base"""

        return {"knowledge_base": None}

    def create_knowledge_base(self,docs):

        """Create a knowledge base from the given documents.
        Args:
            docs (List[str]): List of documents.
        Returns:
            FAISS: Knowledge base built from the documents.
        """

        # Initialize a CharacterTextSplitter to split the documents into chunks
        # Each chunk has a maximum length of 500 characters
        # There is no overlap between the chunks
        text_splitter = CharacterTextSplitter(
            separator="\n", chunk_size=1000, chunk_overlap=200, length_function=len
        )

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

        # Initialize an OpenAIEmbeddings model to compute embeddings of the chunks
        embeddings = OpenAIEmbeddings()

        # Build a knowledge base using FAISS from the chunks and their embeddings
        knowledge_base = FAISS.from_documents(chunks, embeddings)

        # Return the resulting knowledge base
        return knowledge_base


    def upload_file(self,file_paths):
        """Upload a file and create a knowledge base from its contents.
        Args:
            file_paths : The files to uploaded.
        Returns:
            tuple: A tuple containing the file name and the knowledge base.
        """

        file_paths = [single_file_path.name for single_file_path in file_paths]

        loaders = [UnstructuredFileLoader(file_obj, strategy="fast") for file_obj in file_paths]

        # Load the contents of the file using the loader
        docs = []
        for loader in loaders:
            docs.extend(loader.load())

        # Create a knowledge base from the loaded documents using the create_knowledge_base() method
        knowledge_base = self.create_knowledge_base(docs)


        # Return a tuple containing the file name and the knowledge base
        return file_paths, {"knowledge_base": knowledge_base}



    def answer_question(self,urls, state):
        """Answer a question based on the current knowledge base.
        Args:
            state (dict): The current state containing the knowledge base.
        Returns:
            str: The answer to the question.
        """

        result = self.get_chemicals(urls)
        # Retrieve the knowledge base from the state dictionary
        knowledge_base = state["knowledge_base"]

        # Set the question for which we want to find the answer
        question = "Identify the Chemical Capabilities Only"

        # Perform a similarity search on the knowledge base to retrieve relevant documents
        docs = knowledge_base.similarity_search(question)

        # Initialize an OpenAI language model for question answering
        template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer.
        Identify the Chemical Capabilities Only.
        {context}
        Question :{question}.
        The result should be in bullet points format.
        """

        prompt = PromptTemplate(template=template,input_variables=["context","question"])

        llm = OpenAI(temperature=0.4)
        llm_chain = LLMChain(prompt=prompt, llm=llm)

        # Load a question-answering chain using the language model
        chain = load_qa_chain(llm, chain_type="stuff",prompt=prompt)

        # Run the question-answering chain on the input documents and question
        response = chain.run(input_documents=docs, question=question)

        Answer = response+"\n"+result

        # Return the response as the answer to the question
        return Answer


    def extract_excel_data(self,file_path):
        # Read the Excel file
        df = pd.read_excel(file_path)

        # Flatten the data to a single list
        data_list = []
        for _, row in df.iterrows():
            data_list.extend(row.tolist())

        return data_list

    def comparing_chemicals(self,excel_file_path,chemicals):
        chemistry_capability = self.extract_excel_data(excel_file_path.name)
        response = openai.Completion.create(
        engine="text-davinci-003",
        prompt= f"""Analyse the following text delimited by triple backticks to return the comman chemicals.
                  text : ```{chemicals}  {chemistry_capability}```.
                  result should be in bullet points format.
                 """,
        max_tokens=100,
        n=1,
        stop=None,
        temperature=0,
        top_p=1.0,
        frequency_penalty=0.0,
        presence_penalty=0.0
        )

        result = response.choices[0].text.strip()

        return result

    def gradio_interface(self)->None:
        """
        Starts the Gradio interface for chemical identification.
        """

        try:
          with gr.Blocks(css="style.css",theme='karthikeyan-adople/hudsonhayes-dark1') as demo:
            #gr.HTML("""<center><img src="https://hudsonandhayes.co.uk/wp-content/uploads/2023/03/Hudson_meta.jpg" height="210px" width="310"></center>""")
            gr.HTML("""<center><div class="crop"> <img src="file=logo.png" alt="logo"/></div></center>""") 
            state = gr.State(self.get_empty_state())
            gr.HTML("""<center><h1 style="color:#fff">Chemical Identifier for Syngenta</h1></center>""")
                # btn = gr.Button(value="Submit")
                # chemicals_textbox = gr.Textbox(label="Chemicals",lines=6)
            with gr.Column(elem_id="col-container"):
              with gr.Row(elem_id="row-flex"):
                  url = gr.Textbox(label="URL")
              with gr.Row(elem_id="row-flex"):
                  with gr.Column(scale=0.90, min_width=160):
                      file_output = gr.File(elem_classes="heightfit")
                  with gr.Column(scale=0.10, min_width=160):
                      upload_button = gr.UploadButton(
                          "Browse File", file_types=[".txt", ".pdf", ".doc", ".docx"],
                          elem_classes="heightfit",variant="primary",
                          file_count = "multiple")
              with gr.Row():
                with gr.Column(scale=1, min_width=0):
                  excel_input = gr.File(elem_classes="heightfit1",label = "excel file",file_types = [".xlsx"])
              with gr.Row():
                with gr.Column(scale=1, min_width=0):
                  analyse_btn = gr.Button(value="Analyse",variant="primary")
              with gr.Row():
                with gr.Column(scale=1, min_width=0):
                  answer = gr.Textbox(value="",label='Chemicals :',show_label=True, placeholder="",lines=5)
              with gr.Row():
                with gr.Column(scale=1, min_width=0):
                  compare_btn = gr.Button(value="valid",variant="primary")
              with gr.Row():
                with gr.Column(scale=1, min_width=0):
                  compared_result = gr.Textbox(value="",label='valid chemicals :',show_label=True, placeholder="",lines=5)


            upload_button.upload(self.upload_file, upload_button, [file_output,state])

            analyse_btn.click(self.answer_question, [url,state], [answer])

            compare_btn.click(self.comparing_chemicals,[excel_input,answer],compared_result)

            # btn.click(fn=self.get_chemicals, inputs=url, outputs=chemicals_textbox)
          demo.launch()

        except Exception as e:
            self.logger.error("Error occurred while launching Gradio interface: %s", str(e))
            raise ValueError("Error occurred while launching Gradio interface") from e

if __name__ == "__main__":
  logging.basicConfig(level=logging.DEBUG)
  chemical_identifier = ChemicalIdentifier()
  chemical_identifier.gradio_interface()