INITIAL COMMIT

.gitignore

@@ -0,0 +1,162 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/

app.py

@@ -1,7 +1,310 @@
-import gradio as gr
-
-def greet(name):
-    return "Hello " + name + "!!"
-
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
-demo.launch()
+from src.pipelines.completePipeline import Pipeline
+import gradio as gr
+import os
+
+# os.system("apt-get update -y")
+# os.system("apt-get upgrade -y")
+# os.system("apt install poppler-utils -y")
+
+chain = None
+pipeline = Pipeline()
+
+
+def getTextResponse(text: str, inputQuery: str):
+    global chain
+    if chain is None:
+        chain = pipeline.plainText(text = text)
+    else:
+        pass
+    response = chain.invoke(
+        {
+            "question": inputQuery
+        }
+    )
+    return response
+
+
+def getSearchablePdfResponse(path: str, inputQuery: str):
+    global chain
+    if chain is None:
+        chain = pipeline.searchablePdf(path = path)
+    else:
+        pass
+    response = chain.invoke(
+        {
+            "question": inputQuery
+        }
+    )
+    return response
+
+def getScannablePdfResponse(path: str, inputQuery: str):
+    global chain
+    if chain is None:
+        chain = pipeline.scannablePdf(path = path)
+    else:
+        pass
+    response = chain.invoke(
+        {
+            "question": inputQuery
+        }
+    )
+    return response
+
+def clearFunction():
+    global chain
+    chain = None
+
+with gr.Blocks() as textInterface:
+    with gr.Row():
+        inputText = gr.Textbox(
+            label = "Input Text",
+            placeholder = "Enter you text here"
+        )
+    with gr.Row():
+        question = gr.Textbox(
+            label = "Question",
+            placeholder = "Enter your question here"
+        )
+        answer = gr.Textbox(
+            label = "Response",
+            interactive = False
+        )
+    with gr.Row():
+        submitButton = gr.Button(
+            value = "Submit",
+            variant = "primary"
+        )
+        clearButton = gr.ClearButton(
+            components = [inputText, question, answer],
+            value = "Clear",
+            variant = "secondary"
+        )
+    submitButton.click(
+        fn = getTextResponse,
+        inputs = [inputText, question],
+        outputs = [answer]
+    )
+    clearButton.click(
+        fn = clearFunction
+    )
+
+
+with gr.Blocks() as searchablePdf:
+    with gr.Row():
+        inputFile = gr.File(
+            file_types = [".pdf"],
+            file_count = "single",
+            label = "Select PDF"
+        )
+    with gr.Row():
+        question = gr.Textbox(
+            label = "Question",
+            placeholder = "Enter your question here"
+        )
+        answer = gr.Textbox(
+            label = "Response",
+            interactive = False
+        )
+    with gr.Row():
+        submitButton = gr.Button(
+            value = "Submit",
+            variant = "primary"
+        )
+        clearButton = gr.ClearButton(
+            components = [inputFile, question, answer],
+            value = "Clear",
+            variant = "secondary"
+        )
+    submitButton.click(
+        fn = getSearchablePdfResponse,
+        inputs = [inputFile, question],
+        outputs = [answer]
+    )
+    clearButton.click(
+        fn = clearFunction
+    )
+
+
+with gr.Blocks() as scannablePdf:
+    with gr.Row():
+        inputFile = gr.File(
+            file_types = [".pdf"],
+            file_count = "single",
+            label = "Select PDF"
+        )
+    with gr.Row():
+        question = gr.Textbox(
+            label = "Question",
+            placeholder = "Enter your question here"
+        )
+        answer = gr.Textbox(
+            label = "Response",
+            interactive = False
+        )
+    with gr.Row():
+        submitButton = gr.Button(
+            value = "Submit",
+            variant = "primary"
+        )
+        clearButton = gr.ClearButton(
+            components = [inputFile, question, answer],
+            value = "Clear",
+            variant = "secondary"
+        )
+    submitButton.click(
+        fn = getScannablePdfResponse,
+        inputs = [inputFile, question],
+        outputs = [answer]
+    )
+    clearButton.click(
+        fn = clearFunction
+    )
+
+
+def getLinksButtonFn(baseUrl: str):
+    links = pipeline.webCrawler.getLinks(url = baseUrl)
+    checkboxes = gr.CheckboxGroup(
+        choices = links,
+        label = "Fetched Links",
+        visible = True
+    )
+    row2 = gr.Row(visible = True)
+    row3 = gr.Row(visible = True)
+    return (
+        checkboxes,
+        row2,
+        row3
+    )
+
+def getWebsiteResponse(links: list[str], inputQuery: str):
+    global chain
+    if chain is None:
+        print(links)
+        chain = pipeline.webCrawl(urls = links)
+    else:
+        pass
+    response = chain.invoke(
+        {
+            "question": inputQuery
+        }
+    )
+    return response
+
+def clearWebsiteResponse():
+    global chain
+    chain = None
+    checkboxes = gr.CheckboxGroup(
+        choices = [],
+        label = "Fetched Links",
+        visible = False
+    )
+    return checkboxes
+
+with gr.Blocks() as websiteCrawler:
+    with gr.Row():
+        inputUrl = gr.Textbox(
+            label = "Base URL",
+            placeholder = "Enter the Base URL to fetch other links",
+            scale = 3
+        )
+        getLinksButton = gr.Button(
+            value = "Get Links",
+            variant = "primary",
+            scale = 1
+        )
+    checkboxes = gr.CheckboxGroup(
+        choices = [],
+        label = "Fetched Links",
+    )
+    with gr.Row(visible = False) as row2:
+        question = gr.Textbox(
+            label = "Question",
+            placeholder = "Enter your question here"
+        )
+        answer = gr.Textbox(
+            label = "Response",
+            interactive = False
+        )
+    with gr.Row(visible = False) as row3:
+        submitButton = gr.Button(
+            value = "Submit",
+            variant = "primary"
+        )
+        clearButton = gr.ClearButton(
+            components = [question, answer],
+            value = "Clear",
+            variant = "secondary"
+        )
+    getLinksButton.click(
+        fn = getLinksButtonFn,
+        inputs = [inputUrl],
+        outputs = [checkboxes, row2, row3]
+    )
+    submitButton.click(
+        fn = getWebsiteResponse,
+        inputs = [checkboxes, question],
+        outputs = [answer]
+    )
+    clearButton.click(
+        fn = clearWebsiteResponse,
+        inputs = None,
+        outputs = [checkboxes]
+    )
+
+
+def getYoutubeResponse(links: str, inputQuery: str):
+    global chain
+    links = [link.strip() for link in links.split(",")]
+    if chain is None:
+        chain = pipeline.youtubeLinks(urls = links)
+    else:
+        pass
+    response = chain.invoke(
+        {
+            "question": inputQuery
+        }
+    )
+    return response
+
+
+with gr.Blocks() as youtubeInterface:
+    with gr.Row():
+        inputLinks = gr.Textbox(
+            label = "Youtube Links",
+            placeholder = 'Enter comma(,)-separated youtube video links'
+        )
+    with gr.Row():
+        question = gr.Textbox(
+            label = "Question",
+            placeholder = "Enter your question here"
+        )
+        answer = gr.Textbox(
+            label = "Response",
+            interactive = False
+        )
+    with gr.Row():
+        submitButton = gr.Button(
+            value = "Submit",
+            variant = "primary"
+        )
+        clearButton = gr.ClearButton(
+            components = [inputLinks, question, answer],
+            value = "Clear",
+            variant = "secondary"
+        )
+    submitButton.click(
+        fn = getYoutubeResponse,
+        inputs = [inputLinks, question],
+        outputs = [answer]
+    )
+    clearButton.click(
+        fn = clearFunction
+    )
+
+
+application = gr.TabbedInterface(
+    [textInterface, searchablePdf, scannablePdf, websiteCrawler, youtubeInterface],
+    ["Text", "Searchable PDF", "Scannable PDF", "Website Text", "Youtube Transcripts"]
+)
+
+application.launch()

config.ini

@@ -0,0 +1,25 @@
+[EMBEDDINGS]
+embeddingModel = sentence-transformers/all-MiniLM-L6-v2
+device = cpu
+normalize_embeddings = true
+
+[VECTORSTORE]
+chunkSize = 1250
+chunkOverlap = 250
+addStartIndex = true
+
+[LLM]
+llmModel = llama-3.1-70b-versatile
+maxTokens = 512
+temperature = 0.75
+
+[RETRIEVER]
+searchType = mmr
+k = 5
+fetchK = 10
+
+[WEBCRAWLER]
+timeout = 30
+
+[EASYOCR]
+gpu = false

params.yaml

@@ -0,0 +1,19 @@
+prompt: |
+  INSTRUCTIONS:
+  =====================================
+  ### Role
+  **Primary Function**: You are an AI chatbot designed to provide accurate and efficient assistance to users based on provided context data. Your responses must be reliable, friendly, and directly address user inquiries or issues. Always clarify any unclear questions, and conclude responses positively.
+  ### Constraints
+  1. **No Data Disclosure**: Never reveal access to training data or any context explicitly.
+  2. **Maintaining Focus**: Politely redirect any off-topic conversations back to relevant issues without breaking character.
+  3. **Exclusive Reliance on Context Data**: Base all answers strictly on the provided context data. If the context doesn’t cover the query, use a fallback response. Always maintain a third-person perspective.
+  4. **Restrictive Role Focus**: Do not engage in tasks or answer questions unrelated to your role or context data.
+  Ensure all instructions are strictly followed. Responses must be meaningful and concise, within 512 words. Make sure the user is always happy and satisfied with the outputs you return.
+  CONTEXT:
+  =====================================
+  {context}
+  ======================================
+  QUESTION:
+  =====================================
+  {question}
+  NOTE: Generate responses directly without using phrases like "Response:" or "Answer:". NEVER mention the user about usage of any context to generate an answer.

requirements.txt

@@ -0,0 +1,15 @@
+gradio
+langchain
+langchain-core
+langchain-groq
+langchain-community
+langchain_huggingface
+pymupdf
+easyocr 
+numpy
+pdf2image
+requests
+python-dotenv
+beautifulsoup4
+youtube-transcript-api
+urllib3

secrets.env

@@ -0,0 +1 @@
+GROQ_API_KEY=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

setup.py

@@ -0,0 +1,15 @@
+from setuptools import setup, find_packages
+
+def getRequirements(requirementsPath: str) -> list[str]:
+    with open(requirementsPath) as file:
+        requirements = file.read().split("\n")
+    return requirements
+
+setup(
+    name = "ConversAI",
+    author = "Rauhan Ahmed Siddiqui",
+    author_email = "rauhaan.siddiqui@gmail.com",
+    version = "0.1",
+    packages = find_packages(),
+    install_requires = getRequirements(requirementsPath = "requirements.txt")
+)

src/components/loaders/pdfLoader.py

@@ -0,0 +1,42 @@
+from src.utils.functions import cleanText, getConfig
+from concurrent.futures import ThreadPoolExecutor
+from src.utils.exceptions import CustomException
+from pdf2image import convert_from_path
+from src.utils.logging import logger
+import numpy as np
+import pymupdf
+import easyocr
+
+
+class PdfLoader:
+    def __init__(self) -> None:
+        self.config = getConfig(path = "config.ini")
+        self.reader = easyocr.Reader(['en'], gpu = self.config.getboolean("EASYOCR", "gpu"))
+    
+    def extractTextFromPage(self, page):
+        return cleanText(text = page.get_text())
+
+    def searchablePdf(self, pdfPath: str):
+        try: 
+            logger.info("Text Extraction Started from Searchable PDF")
+            doc = pymupdf.open(pdfPath)
+            pages = [doc.load_page(i) for i in range(len(doc))]
+            with ThreadPoolExecutor() as executor:
+                texts = list(executor.map(self.extractTextFromPage, pages))
+            doc.close()
+            return "\n".join(texts)
+        except Exception as e:
+            logger.error(CustomException(e))
+    
+    def getText(self, image):
+        text = "\n".join([text[1] for text in self.reader.readtext(np.array(image), paragraph=True)])
+        return cleanText(text = text)
+
+    def scannablePdf(self, pdfPath: str):
+        try:
+            logger.info("Text Extraction Started from Scannable PDF")
+            allImages = convert_from_path(pdfPath)
+            texts = [self.getText(image) for image in allImages]
+            return "\n".join(texts)
+        except Exception as e:
+            logger.error(CustomException(e))

src/components/loaders/websiteCrawler.py

@@ -0,0 +1,65 @@
+from concurrent.futures import ThreadPoolExecutor
+from src.utils.exceptions import CustomException
+from urllib.parse import urlparse, urljoin
+from src.utils.functions import getConfig
+from src.utils.functions import cleanText
+from src.utils.logging import logger
+from bs4 import BeautifulSoup
+import time
+import requests
+
+
+class WebsiteCrawler:
+    def __init__(self):
+        self.config = getConfig(path = "config.ini")
+
+    def getLinksFromPage(self, url: str):
+        response = requests.get(url)
+        soup = BeautifulSoup(response.content, "html.parser")
+        anchors = soup.find_all("a")
+        links = []
+        for anchor in anchors:
+            if "href" in anchor.attrs:
+                if urlparse(anchor.attrs["href"]).netloc == urlparse(url).netloc:
+                    links.append(anchor.attrs["href"])
+                elif not anchor.attrs["href"].startswith(("//", "file", "javascript", "tel", "mailto", "http")):
+                    links.append(urljoin(url + "/", anchor.attrs["href"]))
+                else:
+                    pass
+                links = [link for link in links if "#" not in link]
+                links = list(set(links))
+            else:
+                continue
+        return links
+    
+    def getLinks(self, url: str):
+        try:
+            logger.info("fetching links from url")
+            start = time.time()
+            links = self.getLinksFromPage(url)
+            uniqueLinks = set()
+            for link in links:
+                now = time.time()
+                if now - start > self.config.getint("WEBCRAWLER", "timeout"):
+                    break
+                else:
+                    uniqueLinks = uniqueLinks.union(set(self.getLinksFromPage(link)))
+            return list(set([x[:len(x) - 1] if x[-1] == "/" else x for x in uniqueLinks]))
+        except Exception as e:
+            logger.error(CustomException(e))
+
+    def extractTextFromUrl(self, url: str):
+        response = requests.get(url)
+        response.raise_for_status()
+        html = response.text
+        soup = BeautifulSoup(html, 'html.parser')
+        return cleanText(text = soup.get_text(separator=' ', strip=True))
+
+    def extractTextFromUrlList(self, urls: list[str]):
+        try:
+            logger.info("extracting text from urls")
+            with ThreadPoolExecutor() as executor:
+                texts = list(executor.map(self.extractTextFromUrl, urls))
+            return "\n".join(texts) 
+        except Exception as e:
+            logger.error(CustomException(e))

src/components/loaders/youtubeLoader.py

@@ -0,0 +1,24 @@
+from langchain_community.document_loaders import YoutubeLoader
+from src.utils.exceptions import CustomException
+from src.utils.functions import cleanText
+from src.utils.logging import logger
+
+
+class YoutubeTranscriptLoader:
+    def __init__(self):
+        pass
+        
+    def getTranscripts(self, urls: str):
+        texts = []
+        for url in set(urls):
+            try:
+                loader = YoutubeLoader.from_youtube_url(
+                    url, add_video_info=False
+                )
+                doc = " ".join([x.page_content for x in loader.load()])
+                texts.append(cleanText(text = doc))
+            except Exception as e:
+                logger.error(CustomException(e))
+                doc = ""
+                texts.append(doc)
+        return "\n".join(texts)

src/components/rag/RAG.py

@@ -0,0 +1,42 @@
+from src.components.vectors.vectorstore import VectorStore
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.runnables import RunnableLambda
+from src.utils.exceptions import CustomException
+from src.utils.functions import getConfig
+from src.utils.functions import loadYaml
+from src.utils.logging import logger
+from langchain_groq import ChatGroq
+
+
+class Chain:
+    def __init__(self):
+        self.config = getConfig(path = "config.ini")
+        self.store = VectorStore()
+        prompt = loadYaml(path = "params.yaml")["prompt"]
+        self.prompt = ChatPromptTemplate.from_template(prompt)
+
+    def formatDocs(self, docs):
+        context = ""
+        for doc in docs:
+            context += f"{doc}\n\n\n"
+        if context == "":
+            context = "No Context Found"
+        else:
+            pass
+        return context
+
+    def returnChain(self, text: str):
+        try:
+            logger.info("preparing chain")
+            store = self.store.setupStore(text = text)
+            chain = (
+                    {"context": RunnableLambda(lambda x: x["question"]) | store | RunnableLambda(self.formatDocs),
+                     "question": RunnableLambda(lambda x: x["question"])}
+                    | self.prompt
+                    | ChatGroq(model_name = self.config.get("LLM", "llmModel"), temperature = self.config.getfloat("LLM", "temperature"), max_tokens = self.config.getint("LLM", "maxTokens"))
+                    | StrOutputParser()
+            )
+            return chain
+        except Exception as e:
+            logger.error(CustomException(e))

src/components/vectors/vectorstore.py

@@ -0,0 +1,38 @@
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+from langchain_core.vectorstores import InMemoryVectorStore
+from langchain_community.docstore.document import Document
+from langchain_huggingface import HuggingFaceEmbeddings
+from src.utils.exceptions import CustomException
+from src.utils.functions import getConfig
+from src.utils.logging import logger
+
+class VectorStore:
+    def __init__(self):
+        self.config = getConfig(path = "config.ini")
+        self.vectorEmbeddings = HuggingFaceEmbeddings(
+            model_name = self.config.get("EMBEDDINGS", "embeddingModel"),
+            model_kwargs = {"device": self.config.get("EMBEDDINGS", "device")},
+            encode_kwargs = {"normalize_embeddings": self.config.getboolean("EMBEDDINGS", "normalize_embeddings")}
+        )
+        self.splitter = RecursiveCharacterTextSplitter(
+            chunk_size = self.config.getint("VECTORSTORE", "chunkSize"),
+            chunk_overlap = self.config.getint("VECTORSTORE", "chunkOverlap"),
+            add_start_index = self.config.getboolean("VECTORSTORE", "addStartIndex")
+        )
+
+    def setupStore(self, text: str):
+        try:
+            store = InMemoryVectorStore(self.vectorEmbeddings)
+            textDocument = Document(page_content = text)
+            documents = self.splitter.split_documents([textDocument])
+            store.add_documents(documents = documents)
+            return store.as_retriever(
+                search_type = self.config.get("RETRIEVER", "searchType"),
+                search_kwargs = {
+                    "k": self.config.getint("RETRIEVER", "k"),
+                    "fetch_k": self.config.getint("RETRIEVER", "fetchK")                
+                }
+            )            
+        except Exception as e:
+            print(CustomException(e))
+            logger.error(CustomException(e))

src/pipelines/completePipeline.py

@@ -0,0 +1,39 @@
+from src.components.loaders.websiteCrawler import WebsiteCrawler
+from src.components.loaders.youtubeLoader import YoutubeTranscriptLoader
+from src.components.loaders.pdfLoader import PdfLoader
+from src.components.rag.RAG import Chain
+from dotenv import load_dotenv
+
+load_dotenv("secrets.env")
+
+class Pipeline:
+    def __init__(self):
+        self.pdfLoader = PdfLoader()
+        self.webCrawler = WebsiteCrawler()
+        self.youtubeLoader = YoutubeTranscriptLoader()
+        self.ragChain = Chain()
+        
+    def plainText(self, text: str):
+        chain = self.ragChain.returnChain(text = text)
+        return chain
+
+    def searchablePdf(self, path: str):
+        extractedText = self.pdfLoader.searchablePdf(pdfPath = path)
+        chain = self.ragChain.returnChain(text = extractedText)
+        return chain
+
+    def scannablePdf(self, path: str):
+        extractedText = self.pdfLoader.scannablePdf(pdfPath = path)
+        chain = self.ragChain.returnChain(text = extractedText)
+        return chain
+    
+    def webCrawl(self, urls: list[str]):
+        extractedText = self.webCrawler.extractTextFromUrlList(urls = urls)
+        chain = self.ragChain.returnChain(text = extractedText)
+        return chain     
+    
+    def youtubeLinks(self, urls: list[str]):
+        extractedText = self.youtubeLoader.getTranscripts(urls = urls)
+        print(extractedText)
+        chain = self.ragChain.returnChain(text = extractedText)
+        return chain     

src/utils/exceptions.py

@@ -0,0 +1,16 @@
+import sys
+
+def error_message_detail(error):
+    _, _, exc_info = sys.exc_info()
+    filename = exc_info.tb_frame.f_code.co_filename
+    lineno = exc_info.tb_lineno
+    error_message = "Error encountered in line no [{}], filename : [{}], saying [{}]".format(lineno, filename, error)
+    return error_message
+
+class CustomException(Exception):
+    def __init__(self, error_message):
+        super().__init__(error_message)
+        self.error_message = error_message_detail(error_message)
+
+    def __str__(self) -> str:
+        return self.error_message

src/utils/functions.py

@@ -0,0 +1,17 @@
+import configparser
+import string
+import yaml
+
+def getConfig(path: str):
+    config = configparser.ConfigParser()
+    config.read(path)
+    return config
+
+def cleanText(text: str):
+    text = text.replace("\n", " ")
+    text = text.translate(str.maketrans('', '', string.punctuation.replace(".", "")))
+    return text
+
+def loadYaml(path: str):
+    with open(path) as file:
+        return yaml.safe_load(file)

src/utils/logging.py

@@ -0,0 +1,12 @@
+import logging
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+logFormat = "[%(asctime)s: %(levelname)s: %(module)s: %(message)s]"
+logFormatter = logging.Formatter(fmt = logFormat, style = "%")
+
+streamHandler = logging.StreamHandler()
+streamHandler.setFormatter(logFormatter)
+
+logger.addHandler(streamHandler)