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
|
import json
import logging
from abc import ABC, abstractmethod
from typing import BinaryIO
import charset_normalizer
import docx
import pypdf
import yaml
from bs4 import BeautifulSoup
from pylatexenc.latex2text import LatexNodes2Text
logger = logging.getLogger(__name__)
class ParserStrategy(ABC):
@abstractmethod
def read(self, file: BinaryIO) -> str:
...
# Basic text file reading
class TXTParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
charset_match = charset_normalizer.from_bytes(file.read()).best()
logger.debug(
f"Reading {getattr(file, 'name', 'file')} "
f"with encoding '{charset_match.encoding}'"
)
return str(charset_match)
# Reading text from binary file using pdf parser
class PDFParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
parser = pypdf.PdfReader(file)
text = ""
for page_idx in range(len(parser.pages)):
text += parser.pages[page_idx].extract_text()
return text
# Reading text from binary file using docs parser
class DOCXParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
doc_file = docx.Document(file)
text = ""
for para in doc_file.paragraphs:
text += para.text
return text
# Reading as dictionary and returning string format
class JSONParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
data = json.load(file)
text = str(data)
return text
class XMLParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
soup = BeautifulSoup(file, "xml")
text = soup.get_text()
return text
# Reading as dictionary and returning string format
class YAMLParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
data = yaml.load(file, Loader=yaml.SafeLoader)
text = str(data)
return text
class HTMLParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
soup = BeautifulSoup(file, "html.parser")
text = soup.get_text()
return text
class LaTeXParser(ParserStrategy):
def read(self, file: BinaryIO) -> str:
latex = file.read().decode()
text = LatexNodes2Text().latex_to_text(latex)
return text
class FileContext:
def __init__(self, parser: ParserStrategy, logger: logging.Logger):
self.parser = parser
self.logger = logger
def set_parser(self, parser: ParserStrategy) -> None:
self.logger.debug(f"Setting Context Parser to {parser}")
self.parser = parser
def decode_file(self, file: BinaryIO) -> str:
self.logger.debug(
f"Reading {getattr(file, 'name', 'file')} with parser {self.parser}"
)
return self.parser.read(file)
extension_to_parser = {
".txt": TXTParser(),
".md": TXTParser(),
".markdown": TXTParser(),
".csv": TXTParser(),
".pdf": PDFParser(),
".docx": DOCXParser(),
".json": JSONParser(),
".xml": XMLParser(),
".yaml": YAMLParser(),
".yml": YAMLParser(),
".html": HTMLParser(),
".htm": HTMLParser(),
".xhtml": HTMLParser(),
".tex": LaTeXParser(),
}
def is_file_binary_fn(file: BinaryIO):
"""Given a file path load all its content and checks if the null bytes is present
Args:
file (_type_): _description_
Returns:
bool: is_binary
"""
file_data = file.read()
file.seek(0)
if b"\x00" in file_data:
return True
return False
def decode_textual_file(file: BinaryIO, ext: str, logger: logging.Logger) -> str:
if not file.readable():
raise ValueError(f"{repr(file)} is not readable")
parser = extension_to_parser.get(ext.lower())
if not parser:
if is_file_binary_fn(file):
raise ValueError(f"Unsupported binary file format: {ext}")
# fallback to txt file parser (to support script and code files loading)
parser = TXTParser()
file_context = FileContext(parser, logger)
return file_context.decode_file(file)
|
