"""Service for parsing Excel documents""" import os import tempfile from docling.datamodel.base_models import InputFormat from docling.document_converter import DocumentConverter, ExcelFormatOption from docling_core.transforms.chunker.hierarchical_chunker import ( ChunkingDocSerializer, ChunkingSerializerProvider, ) from docling_core.transforms.serializer.markdown import MarkdownTableSerializer from docling_core.types.doc.document import DoclingDocument from openpyxl import load_workbook from src.core.config import settings from src.core.logging import get_logger logger = get_logger(__name__) class ExcelTableAnnotationSerializerProvider(ChunkingSerializerProvider): """Custom serializer provider for chunking Excel with table support.""" def get_serializer(self, doc: DoclingDocument) -> ChunkingDocSerializer: """Get serializer with custom table serializer.""" return ChunkingDocSerializer( doc=doc, table_serializer=MarkdownTableSerializer(), ) class ExcelParserService: """Service for parsing Excel files,""" def parse_excel_file( self, file_content: bytes, original_filename: str, ) -> str: """Parse an Excel file and extract chunked elements. Args: file_content: Binary content of the Excel file file_id: PostgreSQL UUID of the file record original_filename: Original filename Returns: Markdown content representing the parsed Excel file """ # Use settings defaults if not specified temp_file_path = None modified_file_path = None try: # Determine file extension _, ext = os.path.splitext(original_filename) if not ext: raise ValueError("File must have an extension to determine Excel format") # Create temporary file temp_fd, temp_file_path = tempfile.mkstemp(suffix=ext) os.close(temp_fd) # Write content with open(temp_file_path, mode="wb") as temp_file: temp_file.write(file_content) logger.info(f"Parsing Excel file: {original_filename}") # Determine which file to parse file_to_parse = temp_file_path # Convert .xls or .xlsm to .xlsx if needed is_xls_file = ext.lower() == ".xls" is_xlsm_file = ext.lower() == ".xlsm" if is_xls_file: logger.info("Detected .xls file, converting to .xlsx format using LibreOffice...") converted_bytes = self._convert_xls_to_xlsx(file_content) # Write converted bytes to a new temp file temp_fd, converted_file_path = tempfile.mkstemp(suffix=".xlsx") os.close(temp_fd) with open(converted_file_path, mode="wb") as f: f.write(converted_bytes) file_to_parse = converted_file_path modified_file_path = converted_file_path elif is_xlsm_file: logger.info("Detected .xlsm file, converting to .xlsx format (removing macros)...") converted_bytes = self._convert_xlsm_to_xlsx(file_content) # Write converted bytes to a new temp file temp_fd, converted_file_path = tempfile.mkstemp(suffix=".xlsx") os.close(temp_fd) with open(converted_file_path, mode="wb") as f: f.write(converted_bytes) file_to_parse = converted_file_path modified_file_path = converted_file_path md_content = self._parse(file_to_parse) return md_content finally: # Clean up temporary files if temp_file_path: try: os.remove(temp_file_path) except Exception as e: logger.debug(f"Failed to remove temp file: {e}") if modified_file_path and modified_file_path != temp_file_path: try: os.remove(modified_file_path) except Exception as e: logger.debug(f"Failed to remove modified file: {e}") def _convert_xlsm_to_xlsx(self, xlsm_content: bytes) -> bytes: """Convert .xlsm to .xlsx by removing macros and saving as standard Excel format. Args: xlsm_content: Binary content of the .xlsm file Returns: Binary content of the converted .xlsx file """ logger.debug("Converting .xlsm to .xlsx") try: # Create temporary file for the xlsm content import io # Load the .xlsm file from bytes with openpyxl # data_only=False to preserve formulas, keep_vba=False to remove macros workbook = load_workbook(io.BytesIO(xlsm_content), data_only=False, keep_vba=False) # Save to BytesIO as .xlsx (macros will be removed automatically) output = io.BytesIO() workbook.save(output) workbook.close() # Get the bytes output.seek(0) xlsx_bytes = output.read() logger.debug(f"Successfully converted .xlsm to .xlsx ({len(xlsx_bytes)} bytes)") return xlsx_bytes except Exception as e: logger.error(f"Error converting .xlsm to .xlsx: {e}") raise def _convert_xls_to_xlsx_libreoffice(self, xls_content: bytes) -> bytes: """Convert .xls to .xlsx using LibreOffice (preserves images/shapes). This method requires LibreOffice to be installed on the system. It preserves images, shapes, charts, and formatting. Args: xls_content: Binary content of the .xls file Returns: Binary content of the converted .xlsx file Raises: RuntimeError: If LibreOffice is not available or conversion fails """ import shutil import subprocess # Check if LibreOffice is available libreoffice_cmd = None # Check common locations (including macOS app bundle) possible_paths = [ "libreoffice", "soffice", "/Applications/LibreOffice.app/Contents/MacOS/soffice", # macOS "/usr/bin/libreoffice", # Linux "/usr/bin/soffice", # Linux ] for cmd in possible_paths: if os.path.exists(cmd) or shutil.which(cmd): libreoffice_cmd = cmd break if not libreoffice_cmd: raise RuntimeError( "LibreOffice not found. Please install LibreOffice for full .xls conversion with images/shapes." ) logger.debug(f"Converting .xls to .xlsx using LibreOffice") # Create temporary directory for conversion temp_dir = tempfile.mkdtemp() temp_xls_path = None try: # Write xls content to temporary file temp_fd, temp_xls_path = tempfile.mkstemp(suffix=".xls", dir=temp_dir) os.write(temp_fd, xls_content) os.close(temp_fd) # Run LibreOffice in headless mode to convert result = subprocess.run( [ libreoffice_cmd, "--headless", "--convert-to", "xlsx", "--outdir", temp_dir, temp_xls_path, ], capture_output=True, text=True, timeout=60, ) if result.returncode != 0: raise RuntimeError(f"LibreOffice conversion failed: {result.stderr}") # Calculate output path xlsx_path = temp_xls_path.rsplit(".", 1)[0] + ".xlsx" if not os.path.exists(xlsx_path): raise RuntimeError("Conversion appeared to succeed but output file not found") # Read the converted file with open(xlsx_path, "rb") as f: xlsx_bytes = f.read() logger.debug( f"Successfully converted .xls to .xlsx with LibreOffice ({len(xlsx_bytes)} bytes)" ) return xlsx_bytes except subprocess.TimeoutExpired as e: raise RuntimeError("LibreOffice conversion timed out") from e except Exception as e: logger.error(f"Error converting with LibreOffice: {e}") raise finally: # Clean up temporary directory import shutil try: shutil.rmtree(temp_dir) except Exception as e: logger.debug(f"Failed to remove temp dir: {e}") def _convert_xls_to_xlsx(self, xls_content: bytes) -> bytes: """Convert .xls file to .xlsx format using LibreOffice. Always uses LibreOffice which preserves everything including images, shapes, and formatting. Falls back to xlrd (text-only) if LibreOffice is not available. Args: xls_content: Binary content of the .xls file Returns: Binary content of the converted .xlsx file """ # Try LibreOffice first (always) try: return self._convert_xls_to_xlsx_libreoffice(xls_content) except RuntimeError as e: raise RuntimeError(f"Failed to convert .xls to .xlsx: {e}") from e def _parse(self, file_path: str) -> str: """Parse Excel file using Docling with HybridChunker for efficient chunking. Args: file_path: Path to Excel file Returns: Markdown content """ try: logger.info(f"Parsing Excel file with Docling: {file_path}") # Initialize Docling converter for Excel processing # Configure with ExcelFormatOption converter = DocumentConverter(format_options={InputFormat.XLSX: ExcelFormatOption()}) # Convert Excel document logger.info("Processing Excel with Docling...") result = converter.convert(file_path) doc = result.document return doc.export_to_html() except Exception as e: logger.error(f"Error parsing with Docling: {e}") raise def get_excel_parser_service() -> ExcelParserService: """Factory function to get an instance of ExcelParserService.""" return ExcelParserService()