A data indexing library 100% open source with no need to use any closed source embeddings or opaque code.
from safe_store import TextVectorizer
from safe_store import TextVectorizer, VectorizationMethod, VisualizationMethod
from safe_store.generic_data_loader import GenericDataLoader
from pathlib import Path

# Create an instance of TextVectorizer
vectorizer = TextVectorizer(
    vectorization_method=VectorizationMethod.TFIDF_VECTORIZER,#=VectorizationMethod.BM25_VECTORIZER,
    database_path=Path(__file__).parent.parent/"vectorized_dbs"/"database.json",
    data_visualization_method=VisualizationMethod.TSNE,#VisualizationMethod.PCA,
    save_db=False
)

# Add a document for vectorization
database_path = Path(__file__).parent.parent/"test_database"
documents = [d for d in database_path.iterdir()]
for doc in documents:
    text = GenericDataLoader.read_file(doc)
    vectorizer.add_document(str(doc), text, chunk_size=100, overlap_size=20, force_vectorize=False, add_as_a_bloc=False)

# Index the documents (perform vectorization)
vectorizer.index()

# Embed a query and retrieve similar documents
query_text = "How did wild life evolve?"
similar_texts, _ = vectorizer.recover_text(query_text, top_k=3)

vectorizer.show_document(query_text,show_interactive_form=True)

print("Similar Documents:")
for i, text in enumerate(similar_texts):
    print(f"{i + 1}: {text}")

examples/lollms_with_data/test_long_text_corpus.py
ParisNeo/safe_store
from safe_store import TextVectorizer
from safe_store import TextVectorizer, VectorizationMethod, VisualizationMethod
from safe_store.generic_data_loader import GenericDataLoader
from pathlib import Path
# Create an instance of TextVectorizer
vectorizer = TextVectorizer(
    vectorization_method=VectorizationMethod.TFIDF_VECTORIZER,#=VectorizationMethod.BM25_VECTORIZER,
    database_path=Path(__file__).parent.parent/"vectorized_dbs"/"database.json",
    data_visualization_method=VisualizationMethod.TSNE,#VisualizationMethod.PCA,
    save_db=False
)

# # Add a document for vectorization
database_path = Path(__file__).parent.parent/"test_database"
documents = [d for d in database_path.iterdir()]
for doc in documents:
    text = GenericDataLoader.read_file(doc)
    vectorizer.add_document(str(doc), text, chunk_size=100, overlap_size=20, force_vectorize=False, add_as_a_bloc=False)

# Index the documents (perform vectorization)
vectorizer.index()

# Embed a query and retrieve similar documents
query_text = "What are future space technologies"
similar_texts, _ = vectorizer.recover_text(query_text, top_k=3)

vectorizer.show_document(query_text,show_interactive_form=True)


print("Similar Documents:")
for i, text in enumerate(similar_texts):
    print(f"{i + 1}: {text}")

examples/view and search from db/test_long_text_corpus.py
ParisNeo/safe_store
from safe_store import TextVectorizer
from safe_store import TextVectorizer, VectorizationMethod, VisualizationMethod
from safe_store.generic_data_loader import GenericDataLoader
from pathlib import Path
# Create an instance of TextVectorizer
vectorizer = TextVectorizer(
    vectorization_method=VectorizationMethod.TFIDF_VECTORIZER,#=VectorizationMethod.BM25_VECTORIZER,
    database_path=Path(__file__).parent.parent/"vectorized_dbs"/"database.json",
    data_visualization_method=VisualizationMethod.TSNE,#VisualizationMethod.PCA,
    save_db=True
)

# # Add a document for vectorization
database_path = Path(__file__).parent.parent/"test_database"
documents = [d for d in database_path.iterdir()]
for doc in documents:
    text = GenericDataLoader.read_file(doc)
    vectorizer.add_document(str(doc), text, chunk_size=100, overlap_size=20, force_vectorize=False, add_as_a_bloc=False)

# Index the documents (perform vectorization)
vectorizer.index()

# Embed a query and retrieve similar documents
query_text = "What are future space technologies"
similar_texts, _ = vectorizer.recover_text(query_text, top_k=3)

vectorizer.show_document(query_text,show_interactive_form=True)
vectorizer.save_to_json()

print("Similar Documents:")
for i, text in enumerate(similar_texts):
    print(f"{i + 1}: {text}")

examples/view and search from db/test_long_text_corpus_with_save.py
ParisNeo/safe_store
"""
BM25 (Best Matching 25) Explanation:

BM25 is a ranking function used for information retrieval and text search. It is an extension of the TF-IDF (Term Frequency-Inverse Document Frequency) weighting scheme, designed to overcome some of its limitations. BM25 takes into account the distribution of terms in the document corpus and the frequency of terms in a given document.

The key components of BM25 are as follows:

1. TF (Term Frequency): BM25 considers the frequency of each term in a document. It rewards documents that contain more occurrences of query terms.

2. IDF (Inverse Document Frequency): BM25 calculates the inverse document frequency of each term in the corpus. It penalizes terms that appear in a large number of documents and gives more weight to rare terms.

3. Document Length Normalization: BM25 normalizes the term frequency by taking into account the length of the document. Longer documents are penalized less than shorter ones.

4. Query Parameters (k1 and b): BM25 has two tuning parameters, k1 and b, that control the impact of term frequency and document length, respectively. Adjusting these parameters can fine-tune the ranking results.

The BM25 scoring formula is as follows:

BM25(d, Q) = Σ(wi * IDF(ti) * ((fi * (k1 + 1)) / (fi + k1 * (1 - b + (b * |d| / avgdl))))

Where:
- d: Document being scored.
- Q: Query terms.
- ti: Each term in the query Q.
- fi: Term frequency of term ti in document d.
- |d|: Length of document d.
- avgdl: Average document length in the corpus.
- k1 and b: Tuning parameters.

In practice, to find the most relevant documents for a query, you calculate BM25 scores for each document in the corpus with respect to the query, and then rank the documents by their BM25 scores. The higher the BM25 score, the more relevant the document is to the query.

This class, BM25Vectorizer, provides a way to compute BM25 scores for a corpus of text documents and retrieve the most relevant documents for a given query.
"""
import numpy as np
import re

def split_string(doc:str):
    # Split the string using regular expressions to keep punctuation as separate tokens
    words = re.findall(r'\b\w+\b|[.,!?;]', doc)
    return words
class BM25Vectorizer:
    def __init__(self, k1=1.5, b=0.75):
        self.k1 = k1
        self.b = b
        self.doc_lengths = []
        self.avg_doc_length = 0
        self.doc_term_freqs = []
        self.doc_count = 0
        self.term_idf = {}

    def fit(self, corpus):
        self.doc_count = len(corpus)
        self.doc_lengths = [len(split_string(doc)) for doc in corpus]
        self.avg_doc_length = np.mean(self.doc_lengths)

        self.doc_term_freqs = []
        for doc in corpus:
            term_freqs = {}
            for term in split_string(doc):
                term_freqs[term] = term_freqs.get(term, 0) + 1
            self.doc_term_freqs.append(term_freqs)

        # Calculate IDF for each term in the corpus
        term_doc_count = {}
        for term_freqs in self.doc_term_freqs:
            for term in term_freqs:
                term_doc_count[term] = term_doc_count.get(term, 0) + 1

        for term, doc_count in term_doc_count.items():
            self.term_idf[term] = np.log((self.doc_count - doc_count + 0.5) / (doc_count + 0.5) + 1.0)

    def transform(self, query):
        scores = []

        for doc_term_freqs, doc_length in zip(self.doc_term_freqs, self.doc_lengths):
            doc_score = 0
            for term in split_string(query):
                if term in doc_term_freqs:
                    idf_term = self.term_idf[term]
                    tf_term = doc_term_freqs[term]
                    doc_score += (idf_term * (tf_term * (self.k1 + 1))) / (tf_term + self.k1 * (1 - self.b + self.b * (doc_length / self.avg_doc_length)))

            scores.append(doc_score)

        return np.array(scores)

safe_store/BM25Vectorizer.py
ParisNeo/safe_store
from safe_store.document_decomposer import *
from safe_store.generic_data_loader import *
from safe_store.text_vectorizer import *
from safe_store.tfidf_loader import *
from safe_store.utils import *
safe_store/__init__.py
ParisNeo/safe_store
import re

   
class DocumentDecomposer:
    @staticmethod
    def clean_text(text):
        # Remove extra returns and leading/trailing spaces
        text = text.replace('\r', '').strip()
        return text

    @staticmethod
    def split_into_paragraphs(text):
        # Split the text into paragraphs using two or more consecutive newlines
        paragraphs = [p+"\n" for p in re.split(r'\n{2,}', text)]
        return paragraphs

    @staticmethod
    def tokenize_sentences(paragraph):
        # Custom sentence tokenizer using simple regex-based approach
        sentences = [s+"." for s in paragraph.split(".")]
        sentences = [sentence.strip() for sentence in sentences if sentence.strip()]
        return sentences

    @staticmethod
    def decompose_document(text, max_chunk_size, overlap_size, tokenize=None, detokenize=None):
        cleaned_text = DocumentDecomposer.clean_text(text)
        paragraphs = DocumentDecomposer.split_into_paragraphs(cleaned_text)

        # List to store the final clean chunks
        clean_chunks = []

        current_chunk = []  # To store the current chunk being built
        l = 0
        for paragraph in paragraphs:
            # Tokenize the paragraph into sentences
            sentences = DocumentDecomposer.tokenize_sentences(paragraph)

            for sentence in sentences:
                if tokenize is not None:
                    tokens = tokenize(sentence)
                else:
                    # Split the text using regular expression to preserve line breaks and multiple spaces
                    tokens = re.split(r'(\s+|\n+)', sentence)
                    # tokens = sentence.split()  # Use words as units

                nb_tokens = len(tokens)
                if nb_tokens > max_chunk_size:
                    while nb_tokens > max_chunk_size:
                        current_chunk += tokens[:max_chunk_size - l - 1]
                        clean_chunks.append(current_chunk)
                        tokens = tokens[max_chunk_size - l - 1 - overlap_size:]
                        nb_tokens -= max_chunk_size - l - 1 - overlap_size
                        l = 0
                        current_chunk = current_chunk[-overlap_size:]
                else:
                    if l + nb_tokens + 1 > max_chunk_size:
                        clean_chunks.append(current_chunk)
                        if overlap_size == 0:
                            current_chunk = []
                        else:
                            current_chunk = current_chunk[-overlap_size:]
                        l = 0

                    # Add the current sentence to the chunk
                    current_chunk += tokens
                    l += nb_tokens

        # Add the remaining chunk from the paragraph to the clean_chunks
        if current_chunk:
            clean_chunks.append(current_chunk)
            current_chunk = ""

        return clean_chunks
safe_store/document_decomposer.py
ParisNeo/safe_store
from pathlib import Path
 
class GenericDataLoader:
    @staticmethod        
    def read_file(file_path:Path):
        if file_path.suffix ==".pdf":
            return GenericDataLoader.read_pdf_file(file_path)
        elif file_path.suffix == ".docx":
            return GenericDataLoader.read_docx_file(file_path)
        elif file_path.suffix == ".json":
            return GenericDataLoader.read_json_file(file_path)
        elif file_path.suffix == ".html":
            return GenericDataLoader.read_html_file(file_path)
        
        elif file_path.suffix == ".pptx":
            return GenericDataLoader.read_pptx_file(file_path)
        if file_path.suffix in [".txt", ".rtf", ".md", ".log", ".cpp", ".java", ".js", ".py", ".rb", ".sh", ".sql", ".css", ".html", ".php", ".json", ".xml", ".yaml", ".yml", ".h", ".hh", ".hpp", ".inc", ".snippet", ".snippets", ".asm", ".s", ".se", ".sym", ".ini", ".inf", ".map", ".bat"]:
            return GenericDataLoader.read_text_file(file_path)
        elif file_path.suffix == ".csv":
            return GenericDataLoader.read_csv_file(file_path)
        elif file_path.suffix == ".xlsx":
            return GenericDataLoader.read_excel_file(file_path)
        else:
            raise ValueError("Unknown file type")
    def get_supported_file_types():
        return ["pdf", "txt", "docx", "json", "html", "pptx",".txt", ".md", ".log", ".cpp", ".java", ".js", ".py", ".rb", ".sh", ".sql", ".css", ".html", ".php", ".json", ".xml", ".yaml", ".yml", ".h", ".hh", ".hpp", ".inc", ".snippet", ".snippets", ".asm", ".s", ".se", ".sym", ".ini", ".inf", ".map", ".bat", ".rtf"]    
    @staticmethod        
    def read_pdf_file(file_path):
        import PyPDF2
        def extract_text_from_pdf(file_path):
            text = ""
            with open(file_path, 'rb') as pdf_file:
                pdf_reader = PyPDF2.PdfReader(pdf_file)
                for page in pdf_reader.pages:
                    text += page.extract_text()
            return text
        # Extract text from the PDF
        text = extract_text_from_pdf(file_path)

        # Convert to Markdown (You may need to implement custom logic based on your specific use case)
        markdown_text = text.replace('\n', '  \n')  # Adding double spaces at the end of each line for Markdown line breaks
        
        return markdown_text

    @staticmethod
    def read_docx_file(file_path):
        from docx import Document

        doc = Document(file_path)
        text = ""
        for paragraph in doc.paragraphs:
            text += paragraph.text + "\n"
        return text

    @staticmethod
    def read_json_file(file_path):
        import json
        with open(file_path, 'r') as file:
            data = json.load(file)
        return data
    
    @staticmethod
    def read_csv_file(file_path):
        import pandas as pd
        try:
            df = pd.read_csv(file_path)
            return df
        except FileNotFoundError:
            print(f"File '{file_path}' not found.")
            return None

    @staticmethod
    def read_excel_file(file_path, sheet_name=None):
        import pandas as pd
        try:
            df = pd.read_excel(file_path, sheet_name)
            return df
        except FileNotFoundError:
            print(f"File '{file_path}' not found.")
            return None

    @staticmethod
    def read_html_file(file_path):
        from bs4 import BeautifulSoup
        with open(file_path, 'r') as file:
            soup = BeautifulSoup(file, 'html.parser')
            text = soup.get_text()
        return text
    
    @staticmethod
    def read_pptx_file(file_path):
        from pptx import Presentation
        prs = Presentation(file_path)
        text = ""
        for slide in prs.slides:
            for shape in slide.shapes:
                if shape.has_text_frame:
                    for paragraph in shape.text_frame.paragraphs:
                        for run in paragraph.runs:
                            text += run.text
        return text
    
    @staticmethod
    def read_text_file(file_path):
        with open(file_path, 'r', encoding='utf-8') as file:
            content = file.read()
        return content

safe_store/generic_data_loader.py
ParisNeo/safe_store
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from safe_store.BM25Vectorizer import BM25Vectorizer, split_string  # Import BM25Vectorizer
import numpy as np
from pathlib import Path
import json
from ascii_colors import ASCIIColors, trace_exception
from safe_store.document_decomposer import DocumentDecomposer
from safe_store.tfidf_loader import TFIDFLoader
from safe_store.utils import NumpyEncoderDecoder

from enum import Enum

class VectorizationMethod(Enum):
    MODEL_EMBEDDING = "model_embedding"
    TFIDF_VECTORIZER = "tfidf_vectorizer"
    BM25_VECTORIZER = "bm25_vectorizer"

class VisualizationMethod(Enum):
    PCA = "PCA"
    TSNE = "TSNE"

class TextVectorizer:
    def __init__(
                    self, 
                    vectorization_method:VectorizationMethod|str, # supported "model_embedding" or "tfidf_vectorizer"
                    model=None, #needed in case of using model_embedding
                    database_path=None,
                    save_db=False,
                    data_visualization_method:VisualizationMethod|str=VisualizationMethod.PCA,
                    database_dict=None
                    ):
        if isinstance(vectorization_method, str):
            try:
                vectorization_method = VectorizationMethod(vectorization_method)
            except ValueError:
                raise ValueError("Invalid vectorization_method string. Please use valid enum values or strings.")
        elif not isinstance(vectorization_method, VectorizationMethod):
            raise ValueError("Invalid vectorization_method. Please use VectorizationMethod enum values or strings.")
        
        if isinstance(data_visualization_method, str):
            try:
                data_visualization_method = VisualizationMethod(vectorization_method)
            except ValueError:
                raise ValueError("Invalid vectorization_method string. Please use valid enum values or strings.")
        elif not isinstance(data_visualization_method, VisualizationMethod):
            raise ValueError("Invalid vectorization_method. Please use VisualizationMethod enum values or strings.")
        
        
        
        self.vectorization_method = vectorization_method
        self.save_db = save_db
        self.model = model
        self.database_file = database_path
        
        self.data_visualization_method = data_visualization_method
        
        if database_dict is not None:
            self.chunks =  database_dict["chunks"]
            self.vectorizer = database_dict["vectorizer"]
            self.infos =   database_dict["infos"]
            self.ready = True
        else:
            self.chunks = {}
            self.ready = False
            self.vectorizer = None
        
            if vectorization_method==VectorizationMethod.MODEL_EMBEDDING:
                try:
                    if not self.model or self.model.embed("hi")==None: # test
                        self.vectorization_method=VectorizationMethod.TFIDF_VECTORIZER
                        self.infos={
                            "vectorization_method":VectorizationMethod.TFIDF_VECTORIZER.value
                        }                        
                    else:
                        self.infos={
                            "vectorization_method":VectorizationMethod.MODEL_EMBEDDING.value
                        }
                except Exception as ex:
                    ASCIIColors.error("Couldn't embed the text, so trying to use tfidf instead.")
                    trace_exception(ex)
                    self.infos={
                        "vectorization_method":VectorizationMethod.TFIDF_VECTORIZER.value
                    }
            elif vectorization_method == VectorizationMethod.BM25_VECTORIZER:
                self.infos = {
                    "vectorization_method": VectorizationMethod.BM25_VECTORIZER.value
                }

            else:
                self.infos={
                    "vectorization_method":VectorizationMethod.TFIDF_VECTORIZER.value
                }

        # Load previous state from the JSON file
        if self.save_db:
            if Path(self.database_file).exists():
                ASCIIColors.success(f"Database file found : {self.database_file}")
                self.load_from_json()
                self.ready = True
            else:
                ASCIIColors.info(f"No database file found : {self.database_file}")
                if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER:
                    self.vectorizer = TfidfVectorizer()
                elif self.vectorization_method == VectorizationMethod.BM25_VECTORIZER:
                    self.vectorizer = BM25Vectorizer()
        else:
            if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER:
                self.vectorizer = TfidfVectorizer()
            elif self.vectorization_method == VectorizationMethod.BM25_VECTORIZER:
                self.vectorizer = BM25Vectorizer()

                    
    def show_document(self, query_text=None, save_fig_path=None, show_interactive_form=False, add_hover_detection=False, add_click_detection=False):
        import textwrap
        import seaborn as sns
        import matplotlib.pyplot as plt
        import mplcursors
        from tkinter import Tk, Text, Scrollbar, Frame, Label, TOP, BOTH, RIGHT, LEFT, Y, N, END
        from sklearn.manifold import TSNE
        from sklearn.decomposition import PCA
        if save_fig_path is None and show_interactive_form==False:
            raise Exception("[show_document] parameters error. Either select a save_fig_path to save the graph or set show_interactive_form to True")

        if self.vectorization_method==VectorizationMethod.BM25_VECTORIZER:
            ASCIIColors.red("BM25 does not make a representatrion of the data. So we don't show a distribution.")
            return

        if self.data_visualization_method == VisualizationMethod.PCA:
            use_pca = True
        else:
            use_pca = False

        if use_pca:
            print("Showing pca representation:")
        else:
            print("Showing t-sne representation:")

        embeddings = {key: chunk["embeddings"] for key, chunk in self.chunks.items()}
        emb = list(embeddings.values())
        ref = list(embeddings.keys())

        if len(emb) >= 2:
            # Normalize embeddings
            emb = np.vstack(emb)
            norms = np.linalg.norm(emb, axis=1)
            normalized_embeddings = emb / norms[:, np.newaxis]

            # Embed the query text
            if query_text is not None:
                query_embedding = self.embed_query(query_text)
                query_normalized_embedding = query_embedding / np.linalg.norm(query_embedding)

                # Combine the query embeddings with the document embeddings
                combined_embeddings = np.vstack((normalized_embeddings, query_normalized_embedding))
                ref.append("Query")  # Use a unique name for the query
            else:
                # Combine the query embeddings with the document embeddings
                combined_embeddings = normalized_embeddings

            if use_pca:
                # Use PCA for dimensionality reduction
                pca = PCA(n_components=2)
                try:
                    embeddings_2d = pca.fit_transform(combined_embeddings)
                except Exception as ex:
                    embeddings_2d = []
            else:
                # Use t-SNE for dimensionality reduction
                perplexity = min(30, combined_embeddings.shape[0] - 1)
                tsne = TSNE(n_components=2, perplexity=perplexity)
                embeddings_2d = tsne.fit_transform(combined_embeddings)



            trans={tuple(e.tolist()):r for e,r in zip(embeddings_2d,ref)}

            # Create a dictionary to map document paths to unique colors
            document_paths = ["_".join(path.split("_")[:-1]) if "/" in path or "\\" in path else path for path in ref]
            # Define the number of colors you need
            num_colors = len(document_paths)

            # Use Seaborn's "Set1" palette to get a list of distinct colors
            colors = sns.color_palette("Set1", n_colors=num_colors)

            # Now you can use the 'colors' list for your document paths
            document_path_colors = colors
            unique_document_paths = list(set(document_paths))
            legend_labels= []
            embeddings_by_document={doc:[] for doc in unique_document_paths}
            for i,document_name in enumerate(document_paths):
                legend_labels.append(Path(document_name).stem if "/" in document_name or "\\" in document_name else document_name)
                embeddings_by_document[document_name].append(embeddings_2d[i,:][None,:])

            labels=[]
            for i, (document_name, emb) in enumerate(embeddings_by_document.items()):
                label = Path(document_name).stem if "/" in document_name or "\\" in document_name else document_name
                labels.append(label)
                embeddings_2d = np.vstack(emb)
                plt.scatter(embeddings_2d[:,0], embeddings_2d[:,1],  color=document_path_colors[i], label=label, marker='o' if document_name!="Query" else "x")
                # Add labels to the scatter plot
                for j, (x, y) in enumerate(embeddings_2d[:]):
                    if label!="Query":
                        plt.text(x, y, f"{i}_{j}", fontsize=8)
                    else:
                        plt.text(x, y, f"query", fontsize=8)
            # Create the legend
            plt.legend(title='Document Name', loc='upper right')



            plt.xlabel('Dimension 1')
            plt.ylabel('Dimension 2')

            if use_pca:
                plt.title('Embeddings Scatter Plot based on PCA')
            else:
                plt.title('Embeddings Scatter Plot based on t-SNE')

            # Enable mplcursors to show tooltips on hover
            cursor = mplcursors.cursor(hover=True)

            # Define the hover event handler
            if add_hover_detection:
                @cursor.connect("add")
                def on_hover(sel):
                    index = sel.target.index
                    if trans[tuple(sel.target.real.tolist())]!="Query":
                        text = self.chunks[trans[tuple(sel.target.real.tolist())]]["chunk_text"]
                    else:
                        text = query_text
                    wrapped_text = textwrap.fill(text, width=50)  # Wrap the text into multiple lines
                    sel.annotation.set_text(f"Index: {index}\nText:\n{wrapped_text}")

            if add_click_detection:
                # Define the click event handler using matplotlib event handling mechanism
                def on_click(event):
                    if event.xdata is not None and event.ydata is not None:
                        x, y = event.xdata, event.ydata
                        distances = ((embeddings_2d[:, 0] - x) ** 2 + (embeddings_2d[:, 1] - y) ** 2)
                        index = distances.argmin()
                        if distances[index]<0.1:

                            # Check if the click is outside any data point
                            if index < len(self.chunks):
                                text = self.chunks[trans[tuple(embeddings_2d[index,:].tolist())]]["chunk_text"]
                            elif index == len(self.chunks):
                                text = query_text
                            else:
                                # Clicked outside any data point, close the Tkinter window if it's open
                                plt.close('all')
                                return

                            # Open a new Tkinter window with the content of the text
                            root = Tk()
                            root.title(f"Text for Index {index}")
                            frame = Frame(root)
                            frame.pack(fill=BOTH, expand=True)

                            label = Label(frame, text="Text:")
                            label.pack(side=TOP, padx=5, pady=5)

                            text_box = Text(frame)
                            text_box.pack(side=TOP, padx=5, pady=5, fill=BOTH, expand=True)
                            text_box.insert(END, text)

                            scrollbar = Scrollbar(frame)
                            scrollbar.pack(side=RIGHT, fill=Y)
                            scrollbar.config(command=text_box.yview)
                            text_box.config(yscrollcommand=scrollbar.set)

                            text_box.config(state="disabled")

                            root.mainloop()


                # Connect the click event handler to the figure
                plt.gcf().canvas.mpl_connect("button_press_event", on_click)

            if save_fig_path:
                try:
                    plt.savefig(save_fig_path)
                except Exception as ex:
                    trace_exception(ex)

            if show_interactive_form:
                plt.show()

    
    def file_exists(self, document_name:str)->bool:
        # Loop through the list of dictionaries
        for dictionary in self.chunks:
            if 'document_name' in dictionary and dictionary['document_name'] == document_name:
                # If the document_name is found in the current dictionary, set the flag to True and break the loop
                document_name_found = True
                return True
        return False
    
    def remove_document(self, document_name:str):
        for dictionary in self.chunks:
            if 'document_name' in dictionary and dictionary['document_name'] == document_name:
                # If the document_name is found in the current dictionary, set the flag to True and break the loop
                self.chunks.remove(dictionary)
                return True
        return False

    def add_document(self, document_name:Path, text:str, chunk_size: int, overlap_size:int, force_vectorize=False,add_as_a_bloc=False):
        if self.file_exists(document_name) and not force_vectorize:
            print(f"Document {document_name} already exists. Skipping vectorization.")
            return
        if add_as_a_bloc:
            chunks_text = [self.model.tokenize(text)]
            for i, chunk in enumerate(chunks_text):
                chunk_id = f"{document_name}_chunk_{i + 1}"
                chunk_dict = {
                    "document_name": str(document_name),
                    "chunk_index": i+1,
                    "chunk_text":self.model.detokenize(chunk),
                    "chunk_tokens": chunk,
                    "embeddings":[]
                }
                self.chunks[chunk_id] = chunk_dict
        else:
            if self.model:
                chunks_text = DocumentDecomposer.decompose_document(text, chunk_size, overlap_size, self.model.tokenize, self.model.detokenize)
            else:
                chunks_text = DocumentDecomposer.decompose_document(text, chunk_size, overlap_size)

            for i, chunk in enumerate(chunks_text):
                chunk_id = f"{document_name}_chunk_{i + 1}"
                chunk_dict = {
                    "document_name": str(document_name),
                    "chunk_index": i+1,
                    "chunk_text":self.model.detokenize(chunk) if (self.model and self.model.detokenize) else ' '.join(chunk),
                    "chunk_tokens": chunk,
                    "embeddings":[]
                }
                self.chunks[chunk_id] = chunk_dict
        
    def index(self):
        if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER:
            #if self.debug:
            #    ASCIIColors.yellow(','.join([len(chunk) for chunk in chunks]))
            data=[]
            for k,chunk in self.chunks.items():
                try:
                    data.append(chunk["chunk_text"]) 
                except Exception as ex:
                    print("oups")
            self.vectorizer.fit(data)
        elif self.vectorization_method==VectorizationMethod.BM25_VECTORIZER:
            #if self.debug:
            #    ASCIIColors.yellow(','.join([len(chunk) for chunk in chunks]))
            data=[]
            for k,chunk in self.chunks.items():
                try:
                    data.append(chunk["chunk_text"]) 
                except Exception as ex:
                    print("oups")
            self.vectorizer.fit(data)

            
        # Generate embeddings for each chunk
        for chunk_id, chunk in self.chunks.items():
            # Store chunk ID, embeddings, and original text
            try:
                if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER:
                    chunk["embeddings"] = self.vectorizer.transform([chunk["chunk_text"]]).toarray()
                elif self.vectorization_method==VectorizationMethod.BM25_VECTORIZER:
                    chunk["BM25_data"] = (self.vectorizer.doc_term_freqs, self.vectorizer.doc_lengths) 
                else:
                    chunk["embeddings"] = self.model.embed(chunk["chunk_text"])
            except Exception as ex:
                print("oups")

        if self.save_db:
            self.save_to_json()
            
        self.ready = True


    def embed_query(self, query_text):
        # Generate query embeddings
        if self.vectorization_method == VectorizationMethod.TFIDF_VECTORIZER:
            query_embedding = self.vectorizer.transform([query_text]).toarray()
        elif self.vectorization_method == VectorizationMethod.BM25_VECTORIZER:
            raise Exception("BM25 don't use embedding")
        else:
            query_embedding = self.model.embed(query_text)
            if query_embedding is None:
                ASCIIColors.warning("The model doesn't implement embeddings extraction")
                self.vectorization_method = VectorizationMethod.TFIDF_VECTORIZER
                query_embedding = self.vectorizer.transform([query_text]).toarray()

        return query_embedding

    def __len__(self):
        return len(list(self.chunks.keys()))

    def recover_chunk_by_index(self, index):
        chunk_id = [ch for ch in self.chunks.keys()][index]
        return self.chunks[chunk_id]["chunk_text"]

    def recover_chunk_by_document_name(self, document_name):
        chunks = [ch for ch in self.chunks.values() if ch["document_name"]==document_name]
        return chunks

    def recover_text(self, query, top_k=3):
        if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER or self.vectorization_method==VectorizationMethod.MODEL_EMBEDDING:
            similarities = {}
            query_embedding = self.embed_query(query)
            for chunk_id, chunk in self.chunks.items():
                similarity = cosine_similarity(query_embedding, chunk["embeddings"])
                similarities[chunk_id] = similarity

            # Sort the similarities and retrieve the top-k most similar embeddings
            sorted_similarities = sorted(similarities.items(), key=lambda x: x[1], reverse=True)[:top_k]

            # Retrieve the original text associated with the most similar embeddings
            texts = [self.chunks[chunk_id]["chunk_text"] for chunk_id, _ in sorted_similarities]
        elif self.vectorization_method==VectorizationMethod.BM25_VECTORIZER:
            # Use the BM25Vectorizer to compute BM25 scores for the query
            bm25_scores = self.vectorizer.transform(query)

            # Find the top-k documents with the highest BM25 scores
            top_k_indices = np.argsort(bm25_scores)[::-1][:top_k]

            # Retrieve the original text associated with the top-k documents
            chunk_keys = [key for key,_ in self.chunks.items()]
            texts = [self.chunks[chunk_keys[chunk_id]]["chunk_text"] for chunk_id in top_k_indices]   
            sorted_similarities = np.sort(bm25_scores)
        return texts, sorted_similarities

    def toJson(self):
        state = {
            "chunks": self.chunks,
            "infos": self.infos,
            "vectorizer": TFIDFLoader.create_dict_from_vectorizer(self.vectorizer) if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER else None
        }
        return state
    
    def setVectorizer(self, vectorizer_dict:dict):
        self.vectorizer=TFIDFLoader.create_vectorizer_from_dict(vectorizer_dict)

    def save_to_json(self):
        state = {
            "chunks": self.chunks,
            "infos": self.infos,
            "vectorizer": TFIDFLoader.create_dict_from_vectorizer(self.vectorizer) if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER else None
        }
        with open(self.database_file, "w") as f:
            json.dump(state, f, cls=NumpyEncoderDecoder, indent=4)

    def load_from_json(self):
        ASCIIColors.info("Loading vectorized documents")
        with open(self.database_file, "r") as f:
            database = json.load(f, object_hook=NumpyEncoderDecoder.as_numpy_array)
            self.chunks = database["chunks"]
            self.infos= database["infos"]
            self.ready = True
        if self.vectorization_method==VectorizationMethod.TFIDF_VECTORIZER:
            self.vectorizer = TFIDFLoader.create_vectorizer_from_dict(database["vectorizer"])
        if self.vectorization_method==VectorizationMethod.BM25_VECTORIZER:
            self.vectorizer = BM25Vectorizer()
            data=[]
            for k,chunk in self.chunks.items():
                try:
                    data.append(chunk["chunk_text"]) 
                except Exception as ex:
                    print("oups")
            self.vectorizer.fit(data)

                
                    
                    
    def clear_database(self):
        self.ready = False
        self.vectorizer=None
        self.chunks = {}
        self.infos={}
        if self.save_db:
            self.save_to_json()
            
safe_store/text_vectorizer.py
ParisNeo/safe_store
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy

class TFIDFLoader:
    @staticmethod
    def create_vectorizer_from_dict(tfidf_info):
        vectorizer = TfidfVectorizer(**tfidf_info['params'])
        vectorizer.ngram_range = tuple(vectorizer.ngram_range)
        vectorizer.vocabulary_ = tfidf_info['vocabulary']
        vectorizer.idf_ = tfidf_info['idf_values']
        dt = vectorizer.dtype[8:-2]
        vectorizer.dtype = eval(dt)
        return vectorizer

    @staticmethod
    def create_dict_from_vectorizer(vectorizer):
        tfidf_info = {
            "vocabulary": vectorizer.vocabulary_,
            "idf_values": vectorizer.idf_,# dict(zip(vectorizer.get_feature_names(), )),
            "params": vectorizer.get_params()
        }
        tfidf_info["params"]["dtype"]=str(tfidf_info["params"]["dtype"])

        return tfidf_info
safe_store/tfidf_loader.py
ParisNeo/safe_store
import numpy as np
import json
def serialize_enum_to_json(enum_member):
    return enum_member.value

class NumpyEncoderDecoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, np.ndarray):
            return {'__numpy_array__': True, 'data': obj.tolist()}
        return super(NumpyEncoderDecoder, self).default(obj)

    @staticmethod
    def as_numpy_array(dct):
        if '__numpy_array__' in dct:
            return np.array(dct['data'])
        return dct
safe_store/utils.py
ParisNeo/safe_store
from setuptools import setup, find_packages

# Package metadata
NAME = 'safe_store'
DESCRIPTION = 'A library for safe document storage and vectorization.'
URL = 'https://github.com/ParisNeo/safe_store'
AUTHOR = 'ALOUI Saifeddine (ParisNeo)'
AUTHOR_EMAIL = 'aloui.seifeddine@email.com'
LICENSE = 'Apache 2.0'
VERSION = '0.2.1'  # Update with your desired version

# Read dependencies from requirements.txt
with open('requirements.txt', 'r') as req_file:
    INSTALL_REQUIRES = req_file.read().splitlines()

# Packages to include (find_packages() automatically discovers and includes sub-packages)
PACKAGES = find_packages()

# Define long description (usually from README.md)
with open('README.md', 'r') as readme_file:
    LONG_DESCRIPTION = readme_file.read()

# Additional package data (e.g., data files)
PACKAGE_DATA = {}

setup(
    name=NAME,
    version=VERSION,
    description=DESCRIPTION,
    long_description=LONG_DESCRIPTION,
    long_description_content_type='text/markdown',
    url=URL,
    author=AUTHOR,
    author_email=AUTHOR_EMAIL,
    license=LICENSE,
    packages=PACKAGES,
    package_data=PACKAGE_DATA,
    install_requires=INSTALL_REQUIRES,
    classifiers=[
        'Development Status :: 3 - Alpha',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: 3.9',
        'Programming Language :: Python :: 3.10',
        'Programming Language :: Python :: 3.11',
        'Programming Language :: Python :: 3.12',
    ],
)

setup.py
ParisNeo/safe_store