add txt and docx class loaders

app/loaders/__init__.py

@@ -1,6 +1,6 @@
 
 import os
-from app.loaders.loaders import (DataLoader, txt_loader, pdf_loader, docx_loader, xlsx_loader, DataLoaderCls, PDFLoader)
+from app.loaders.loaders import (DataLoader, txt_loader, pdf_loader, docx_loader, xlsx_loader, DataLoaderCls, PDFLoader, DocxLoader, TxtLoader)
 from langchain.schema.document import Document
 from typing import List
 from app.text_splitters import ITextSplitter
@@ -14,7 +14,10 @@
 }
 
 supported_loaders_cls = {
- 'pdf': PDFLoader
+ 'pdf': PDFLoader,
+ 'doc': DocxLoader,
+ 'docx': DocxLoader,
+ 'txt': TxtLoader,
 }
 
 

app/loaders/loaders.py

@@ -52,6 +52,26 @@ def txt_loader(file: str) -> Callable:
  loader = TextLoader(file)
  return loader.load()
 
+class TxtLoader(DocumentLoader):
+ def __init__(self, file:str) -> None:
+ self.loader = TextLoader(file)
+
+ def load(self) -> List[Document]:
+ return self.loader.load_and_split()
+
+ def load_and_split_text(self, text_splitter: ITextSplitter) -> List[Document]:
+ pages = self.load()
+ split_pages = []
+ for page in pages:
+ page_content = page.page_content.lower()
+ metadatas = page.metadata
+ metadatas.update({"full_page": page_content})
+
+ text_chunks = text_splitter.split_text(page_content)
+ for chunk in text_chunks:
+ split_pages.append(Document(page_content=chunk, metadata=metadatas))
+ return split_pages
+
 
 class PDFLoader(DocumentLoader):
  def __init__(self, file: str) -> None:
@@ -92,6 +112,27 @@ def pdf_loader(file: str) -> Callable:
  return pages
 
 
+class DocxLoader(DocumentLoader):
+ def __init__(self, file:str) -> None:
+ self.loader = UnstructuredWordDocumentLoader(file)
+
+ def load(self) -> List[Document]:
+ return self.loader.load_and_split()
+
+ def load_and_split_text(self, text_splitter: ITextSplitter) -> List[Document]:
+ pages = self.load()
+ split_pages = []
+ for page in pages:
+ page_content = page.page_content.lower()
+ metadatas = page.metadata
+ metadatas.update({"full_page": page_content})
+
+ text_chunks = text_splitter.split_text(page_content)
+ for chunk in text_chunks:
+ split_pages.append(Document(page_content=chunk, metadata=metadatas))
+ return split_pages
+
+
 def docx_loader(file: str) -> Callable:
  loader = Docx2txtLoader(file)
  return loader.load()

app/tests/test_document_loader.py

@@ -3,6 +3,8 @@
  DataLoader,
  DataLoaderCls,
  PDFLoader,
+ DocxLoader,
+ TxtLoader,
  pdf_loader,
  txt_loader,
  docx_loader,
@@ -126,6 +128,18 @@ def test_pdf_loader_cls(self):
  split_pages: List[Document] = pdf_loader.load_and_split_text(self.text_splitter)
  self.assertEqual(list, type(split_pages))
 
+ def test_docx_loader_cls(self):
+ file_path = f'{self.path}/{self.file_name}.docx'
+ docx_loader = DocxLoader(file_path)
+ split_pages: List[Document] = docx_loader.load_and_split_text(self.text_splitter)
+ self.assertEqual(list, type(split_pages))
+
+ def test_txt_loader_cls(self):
+ file_path = f'{self.path}/{self.file_name}.txt'
+ docx_loader = TxtLoader(file_path)
+ split_pages: List[Document] = docx_loader.load_and_split_text(self.text_splitter)
+ self.assertEqual(list, type(split_pages))
+
  def test_load_file_url_and_split_text(self):
  file_path = f'{self.path}/{self.file_name}.pdf'
  file_type = "pdf"