Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/loaders/office.py: 14%
114 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Office and email document loaders for RAG."""
3from __future__ import annotations
5import asyncio
6import email
7import email.policy
8from pathlib import Path
10from lexigram.ai.rag.chunking.types import Chunk
11from lexigram.ai.rag.loaders._io_utils import read_file_bytes, read_file_text
12from lexigram.ai.rag.types import RAGError
14# ---------------------------------------------------------------------------
15# DocxLoader
16# ---------------------------------------------------------------------------
19class DocxLoader:
20 """Load Microsoft Word (.docx) documents.
22 Requires: python-docx
23 Install: pip install openpyxl
24 """
26 async def load(self, source: str | Path) -> list[Chunk]:
27 """Load a .docx file into chunks — one chunk per paragraph.
29 Args:
30 source: Path to the .docx file.
32 Returns:
33 List of chunks, one per non-empty paragraph.
35 Raises:
36 ImportError: If python-docx is not installed.
37 RAGError: If the file cannot be read or parsed.
38 """
39 try:
40 try:
41 import docx # type: ignore[import-not-found]
42 except ImportError as e:
43 msg = "DocxLoader requires 'python-docx'. Install with: pip install python-docx"
44 raise ImportError(msg) from e
46 path = Path(source)
47 raw = await read_file_bytes(path)
49 def _parse() -> list[Chunk]:
50 import io as _io
52 doc = docx.Document(_io.BytesIO(raw))
53 result: list[Chunk] = []
54 for idx, para in enumerate(doc.paragraphs):
55 text = para.text.strip()
56 if not text:
57 continue
58 result.append(
59 Chunk(
60 text=text,
61 source=str(path),
62 chunk_index=len(result),
63 metadata={
64 "source": str(path),
65 "type": "docx",
66 "paragraph_index": idx,
67 "style": para.style.name if para.style else None,
68 },
69 )
70 )
71 return result
73 return await asyncio.to_thread(_parse)
75 except (ImportError, RAGError):
76 raise
77 except (OSError, FileNotFoundError) as e:
78 msg = f"Failed to read DOCX file {source}: {e}"
79 raise RAGError(msg) from e
80 except Exception as e:
81 msg = f"Unexpected error loading DOCX file {source}: {e}"
82 raise RAGError(msg) from e
85# ---------------------------------------------------------------------------
86# ExcelLoader
87# ---------------------------------------------------------------------------
90class ExcelLoader:
91 """Load Microsoft Excel (.xlsx / .xls) documents.
93 Produces one chunk per worksheet row (or per sheet when
94 ``chunk_per_sheet=True``).
96 Requires: openpyxl
97 Install: pip install openpyxl
98 """
100 def __init__(self, *, chunk_per_sheet: bool = False) -> None:
101 """Initialize Excel loader.
103 Args:
104 chunk_per_sheet: When True, produce one chunk per sheet instead
105 of one chunk per row.
106 """
107 self.chunk_per_sheet = chunk_per_sheet
109 async def load(self, source: str | Path) -> list[Chunk]:
110 """Load an Excel file.
112 Args:
113 source: Path to the .xlsx or .xls file.
115 Returns:
116 List of chunks.
118 Raises:
119 ImportError: If openpyxl is not installed.
120 RAGError: If the file cannot be read or parsed.
121 """
122 try:
123 try:
124 import openpyxl # type: ignore[import-untyped]
125 except ImportError as e:
126 msg = "ExcelLoader requires 'openpyxl'. Install with: pip install openpyxl"
127 raise ImportError(msg) from e
129 path = Path(source)
130 raw = await read_file_bytes(path)
132 def _parse() -> list[Chunk]:
133 import io as _io
135 wb = openpyxl.load_workbook(_io.BytesIO(raw), data_only=True)
136 result: list[Chunk] = []
138 for sheet in wb.worksheets:
139 if self.chunk_per_sheet:
140 rows_text = []
141 for row in sheet.iter_rows(values_only=True):
142 cells = [str(c) if c is not None else "" for c in row]
143 rows_text.append("\t".join(cells))
144 text = "\n".join(rows_text)
145 if text.strip():
146 result.append(
147 Chunk(
148 text=text,
149 source=str(path),
150 chunk_index=len(result),
151 metadata={
152 "source": str(path),
153 "type": "excel",
154 "sheet": sheet.title,
155 },
156 )
157 )
158 else:
159 headers: list[str] = []
160 for row_idx, row in enumerate(
161 sheet.iter_rows(values_only=True)
162 ):
163 if row_idx == 0:
164 headers = [
165 str(c) if c is not None else f"col{i}"
166 for i, c in enumerate(row)
167 ]
168 continue
169 cells = [str(c) if c is not None else "" for c in row]
170 text = " ".join(
171 f"{h}: {v}"
172 for h, v in zip(headers, cells, strict=False)
173 )
174 if text.strip():
175 result.append(
176 Chunk(
177 text=text,
178 source=str(path),
179 chunk_index=len(result),
180 metadata={
181 "source": str(path),
182 "type": "excel",
183 "sheet": sheet.title,
184 "row": row_idx,
185 },
186 )
187 )
188 return result
190 return await asyncio.to_thread(_parse)
192 except (ImportError, RAGError):
193 raise
194 except (OSError, FileNotFoundError) as e:
195 msg = f"Failed to read Excel file {source}: {e}"
196 raise RAGError(msg) from e
197 except Exception as e:
198 msg = f"Unexpected error loading Excel file {source}: {e}"
199 raise RAGError(msg) from e
202# ---------------------------------------------------------------------------
203# EmailLoader
204# ---------------------------------------------------------------------------
207class EmailLoader:
208 """Load email messages (.eml files).
210 Extracts subject, from, to, date headers and the plain-text body.
211 Uses only stdlib ``email`` package — no additional dependencies.
212 """
214 async def load(self, source: str | Path) -> list[Chunk]:
215 """Load an .eml file.
217 Args:
218 source: Path to the .eml file.
220 Returns:
221 Single chunk whose text is the email body (plain-text part).
223 Raises:
224 RAGError: If the file cannot be read or parsed.
225 """
226 try:
227 path = Path(source)
228 raw = await read_file_text(path, encoding="utf-8")
230 def _parse() -> Chunk:
231 msg = email.message_from_string(raw, policy=email.policy.default)
232 subject = str(msg.get("subject", ""))
233 from_addr = str(msg.get("from", ""))
234 to_addr = str(msg.get("to", ""))
235 date = str(msg.get("date", ""))
237 body = ""
238 if msg.is_multipart():
239 for part in msg.walk():
240 if part.get_content_type() == "text/plain":
241 payload = part.get_payload(decode=True)
242 if isinstance(payload, bytes):
243 body = payload.decode(
244 part.get_content_charset() or "utf-8",
245 errors="replace",
246 )
247 break
248 else:
249 payload = msg.get_payload(decode=True)
250 if isinstance(payload, bytes):
251 body = payload.decode(
252 msg.get_content_charset() or "utf-8",
253 errors="replace",
254 )
256 text = f"Subject: {subject}\nFrom: {from_addr}\nTo: {to_addr}\nDate: {date}\n\n{body}"
257 return Chunk(
258 text=text.strip(),
259 source=str(path),
260 chunk_index=0,
261 metadata={
262 "source": str(path),
263 "type": "email",
264 "subject": subject,
265 "from": from_addr,
266 "to": to_addr,
267 "date": date,
268 },
269 )
271 chunk = await asyncio.to_thread(_parse)
272 return [chunk]
274 except (OSError, UnicodeDecodeError) as e:
275 msg = f"Failed to read email file {source}: {e}"
276 raise RAGError(msg) from e
277 except Exception as e:
278 msg = f"Unexpected error loading email file {source}: {e}"
279 raise RAGError(msg) from e
282__all__ = ["DocxLoader", "EmailLoader", "ExcelLoader"]