Spaces:
Sleeping
Sleeping
File size: 5,966 Bytes
a4538e5 9adf25b a4538e5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | from pathlib import Path
from typing import List, Any
from langchain_community.document_loaders import PyPDFLoader, TextLoader, CSVLoader
from langchain_community.document_loaders import Docx2txtLoader
from langchain_community.document_loaders.excel import UnstructuredExcelLoader
from langchain_community.document_loaders import JSONLoader
def load_all_documents(data_dir: str) -> List[Any] :
data_path = Path(data_dir).resolve()
print(f"[DEBUG] Data Path : {data_path}")
document = []
## Pdf files
pdf_files = list(data_path.glob('**/*.pdf'))
print(f"[DEBUG] Found {len(pdf_files)} PDF files : { [str(f) for f in pdf_files] }")
for pdf_file in pdf_files:
print(f"[DEBUG] Loading PDF: {pdf_file}")
try:
loader = PyPDFLoader(str(pdf_file))
loaded = loader.load()
print(f"[DEBUG] Loaded {len(loaded)} PDF docs from {pdf_file}")
document.extend(loaded)
except Exception as e:
print(f"[ERROR] Failed to load PDF {pdf_file} : {e}")
## TXT files
txt_files = list(data_path.glob('**/*.txt'))
print(f'[DEBUG] Founded {len(txt_files)} TXT files: {[str(f) for f in txt_files]} ')
for txt_file in txt_files:
print(f"[DEBUG] Loading TXT : {txt_file}")
try:
loader = TextLoader(str(txt_file), encoding="utf-8")
loaded = loader.load()
print(f"[DEBUG] Loaded {len(loaded)} TXT docs from {txt_file}")
document.extend(loaded)
except Exception as e:
print(f"[ERROR] failed to load TXT {txt_file} : {e}")
## CSV files
csv_files = list(data_path.glob('**/*.csv'))
print(f"[DEBUG] Found {len((csv_files))} CSV files : {[str(f) for f in csv_files]}")
for csv_file in csv_files:
print(f"[DEBUG] Loading CSV : {csv_file}")
try :
loader = CSVLoader(str(csv_file))
loaded = loader.load()
print(f"[DEBUG] Loaded {len(loaded)} CSV docs from {csv_file}")
document.extend(loaded)
except Exception as e:
print(f"[ERROR] failed to load CSV {csv_file} : {e}")
## Excel files
xlsx_files = list(data_path.glob('**/*.xlsx'))
print(f"[DEBUG] Found {len(xlsx_files)} xlsx files : {[str(f) for f in xlsx_files]}")
for xlsx_file in xlsx_files:
print(f"[DEBUG] Loading XLSX : {xlsx_file}")
try :
loader = UnstructuredExcelLoader(str(xlsx_file))
loaded = loader.load()
print(f"[DEBUG] Loaded {(len(loaded))} XLSX docs from {xlsx_file}")
document.extend(loaded)
except Exception as e :
print(f"[ERROR] failed to load Excel {xlsx_file} : {e} ")
## Word file
word_files = list(data_path.glob('**/*.docx'))
print(f"[DEBUG] Found {len(word_files)} word files : {[str(f) for f in word_files]}")
for word_file in word_files:
print(f"[DEBUG] Loaded WORD : {word_file}")
try :
loader = Docx2txtLoader(str(word_file))
loaded = loader.load()
print(f"[DEBUG] Loaded {(len(loaded))} Word docs from {word_file}")
document.extend(loaded)
except Exception as e:
print(f"[ERROR] failed to load Word Document {word_file} : {e}")
## JSON files
json_files = list(data_path.glob('**/*.json'))
print(f"[DEBUG] Found {(len(json_files))} json files : {[str(f) for f in json_files]}")
for json_file in json_files:
print(f"[DEBUG] Loaded JSON : {(str(json_file))}")
try:
loader = JSONLoader(
str(json_file),
jq_schema=".",
text_content=False
)
loaded = loader.load()
print(f"[DEBUG] Loaded {(len(loaded))} JSON from {json_file}")
document.extend(loaded)
except Exception as e:
print(f"[ERROR] Failed to load JSON file{json_file} : {e}")
print(f"[DEBUG] Total Loaded Documents : {len(document)}")
# Filter out documents with None or non-string page_content
valid_documents = []
invalid_count = 0
for doc in document:
content = getattr(doc, 'page_content', None)
if content is not None and isinstance(content, str):
valid_documents.append(doc)
else:
invalid_count += 1
if invalid_count > 0:
print(f"[WARNING] Filtered out {invalid_count} documents with invalid page_content (None or non-string)")
print(f"[DEBUG] Valid Documents after filtering : {len(valid_documents)}")
return valid_documents
if __name__ == '__main__':
docs = load_all_documents('Research/data/pdf')
print(f"Loaded {len(docs)} document.")
print("Example document:", docs[0] if docs else None)
# def process_all_pdf(pdf_directory):
# all_documents = []
# pdf_dir = Path(pdf_directory)
# pdf_files = list(pdf_dir.glob('**/*.pdf'))
# print(f"found {len(pdf_files)} PDF files to Process")
# for pdf_files in pdf_files:
# print(f"\npreprocessing : {pdf_files.name}")
# try:
# loader = PyPDFLoader(str(pdf_files))
# documents = loader.load()
# for doc in documents:
# doc.metadata['source_file'] = pdf_files.name
# doc.metadata['file_type'] = 'pdf'
# all_documents.extend(documents)
# print(f" Loaded {len(documents)} pages")
# except Exception as e:
# print(f" Error : {e}")
# print(f"\n Total documents loaded : {len(all_documents)}")
# return all_documents
# all_pdf_files = process_all_pdf('./data') |