"""Service for parsing Excel documents with image extraction and AI descriptions. This service provides comprehensive Excel parsing capabilities including: - Structured content extraction (tables, text, elements) - Image extraction with AI-generated descriptions - Multiple output formats (JSON, Markdown, Text) - Configurable chunking strategies for large documents - Automatic optimization for large files """ import os import tempfile from functools import lru_cache from typing import Any, Dict, List, Tuple, Optional from uuid import UUID import aiofiles import tiktoken import xlrd from openpyxl import load_workbook, Workbook from langchain.text_splitter import RecursiveCharacterTextSplitter from docling.datamodel.base_models import InputFormat from docling.document_converter import DocumentConverter, ExcelFormatOption from docling_core.transforms.chunker.base import BaseChunk from docling_core.transforms.chunker.hierarchical_chunker import ( ChunkingDocSerializer, ChunkingSerializerProvider, DocChunk, ) from docling_core.transforms.chunker.hybrid_chunker import HybridChunker from docling_core.transforms.chunker.tokenizer.openai import OpenAITokenizer from docling_core.transforms.serializer.markdown import MarkdownTableSerializer from src.commons.configs.settings import get_settings from src.commons.handlers.log_handler import get_logger from src.dtos.document_dtos import ChunkedElementDTO, ElementMetadataDTO, TextElementDTO from src.services.excel_image_extraction_service import ExcelImageExtractionService from src.services.excel_shape_extraction_service import ExcelShapeExtractionService logger = get_logger(__name__) # Get settings settings = get_settings() class ExcelTableAnnotationSerializerProvider(ChunkingSerializerProvider): """Custom serializer provider for chunking Excel with table support.""" def get_serializer(self, doc): """Get serializer with custom table serializer.""" return ChunkingDocSerializer( doc=doc, table_serializer=MarkdownTableSerializer(), ) class ExcelParserService: """Service for parsing Excel files, including image extraction and descriptions.""" def __init__( self, image_extraction_service: ExcelImageExtractionService, shape_extraction_service: ExcelShapeExtractionService, ): """Initialize the Excel parser service. Args: image_extraction_service: Service for extracting images from Excel files shape_extraction_service: Service for extracting shapes from Excel files """ self.image_extraction_service = image_extraction_service self.shape_extraction_service = shape_extraction_service # Initialize tokenizer for chunking (similar to PDF parser) self.tokenizer = OpenAITokenizer( tokenizer=tiktoken.encoding_for_model("text-embedding-3-large"), max_tokens=8000, ) async def parse_excel_file( self, file_content: bytes, file_id: UUID, original_filename: str, replace_images_with_descriptions: bool = False, max_characters: int | None = None, large_file_threshold: int | None = None, description_detail: str = "auto", unique_filename: Optional[str] = None, use_pipe_separators: bool = False, ) -> List[ChunkedElementDTO]: """Parse an Excel file and extract chunked elements with image and shape processing. Args: file_content: Binary content of the Excel file file_id: PostgreSQL UUID of the file record original_filename: Original filename replace_images_with_descriptions: Whether to replace images with descriptions in Excel before parsing max_characters: Maximum characters per chunk (uses settings default if None) large_file_threshold: Cell count threshold (deprecated, kept for compatibility) description_detail: Detail level for AI image descriptions ("auto", "low", "high") unique_filename: Optional unique filename (defaults to original_filename if not provided) use_pipe_separators: Whether to use pipe characters (|) as cell separators in markdown tables (True = markdown tables with pipes, False = space-separated format) Returns: List of ChunkedElementDTO objects representing the parsed and chunked content (includes image and shape chunks) """ # Use settings defaults if not specified if max_characters is None: max_characters = settings.excel_max_characters if large_file_threshold is None: large_file_threshold = settings.excel_large_file_threshold temp_file_path = None modified_file_path = None try: # Determine file extension _, ext = os.path.splitext(original_filename) suffix = ext if ext in settings.excel_supported_extensions else settings.excel_supported_extensions[0] # Create temporary file temp_fd, temp_file_path = tempfile.mkstemp(suffix=suffix) os.close(temp_fd) # Write content async with aiofiles.open(temp_file_path, mode="wb") as temp_file: await 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 converted_file_content = file_content # Default to original content is_xls_file = suffix.lower() == '.xls' is_xlsm_file = suffix.lower() == '.xlsm' needs_conversion = is_xls_file or is_xlsm_file if is_xls_file: logger.info("Detected .xls file, converting to .xlsx format using LibreOffice...") converted_file_path = self._convert_xls_to_xlsx(temp_file_path) file_to_parse = converted_file_path # Update modified_file_path to track the converted file if modified_file_path is None: modified_file_path = converted_file_path # Read the converted file content for image/shape extraction async with aiofiles.open(converted_file_path, mode="rb") as f: converted_file_content = await f.read() elif is_xlsm_file: logger.info("Detected .xlsm file, converting to .xlsx format (removing macros)...") converted_file_path = self._convert_xlsm_to_xlsx(temp_file_path) file_to_parse = converted_file_path # Update modified_file_path to track the converted file if modified_file_path is None: modified_file_path = converted_file_path # Read the converted file content for image/shape extraction async with aiofiles.open(converted_file_path, mode="rb") as f: converted_file_content = await f.read() # Extract images if needed (for either replacement or separate chunks) image_extraction_result = None if replace_images_with_descriptions: logger.info("Extracting images and generating descriptions...") image_extraction_result = await self._extract_images_with_descriptions( converted_file_content, file_id, original_filename, description_detail, is_converted_from_xls=needs_conversion ) # Replace images with descriptions if requested if replace_images_with_descriptions and image_extraction_result: position_descriptions = self._create_position_descriptions_map( image_extraction_result ) if position_descriptions: logger.info( f"Replacing {len(position_descriptions)} images with descriptions..." ) modified_file_path = self._replace_images_with_descriptions( file_to_parse, position_descriptions # Use file_to_parse (already converted if .xls) ) file_to_parse = modified_file_path # Extract shapes BEFORE parsing (so we can apply annotations first) shape_count = 0 logger.info("Extracting shapes with text content...") shape_dtos = await self.shape_extraction_service.extract_shapes_from_file( file_content=converted_file_content, file_id=file_id, original_filename=original_filename, unique_filename=unique_filename, ) # Apply shape annotations to cells BEFORE parsing if shape_dtos and not replace_images_with_descriptions: # Only apply annotations if we're not already modifying the file for images logger.info(f"Applying {len(shape_dtos)} shape annotations to cells...") file_to_parse = self._apply_shape_annotations_to_cells( file_to_parse, shape_dtos ) logger.info(f"Successfully applied shape annotations") logger.info(f"File to parse after applying shape annotations: {file_to_parse}") # Extract and apply comments BEFORE parsing logger.info("Extracting comments from Excel file...") comments_by_sheet = self._extract_comments_from_file(file_to_parse) comment_id_mapping = {} # Will store {comment_id: comment_metadata} if comments_by_sheet: total_comments = sum(len(comments) for comments in comments_by_sheet.values()) logger.info(f"Found {total_comments} comments, applying ID markers to cells...") # Apply comment ID markers to cells and get mapping commented_file_path, comment_id_mapping = self._apply_comments_to_cells( file_to_parse, comments_by_sheet ) # Update file_to_parse only if a new file was created if commented_file_path != file_to_parse: # If we had a previous modified file, we need to track it for cleanup if modified_file_path is None: modified_file_path = commented_file_path file_to_parse = commented_file_path logger.info(f"Successfully applied comment annotations") else: logger.info("No comments found in Excel file") # Check file size and determine parsing method _, total_cells = self._check_file_size(file_to_parse) use_openpyxl = total_cells > large_file_threshold if use_openpyxl: logger.warning( f"File has {total_cells:,} cells (threshold: {large_file_threshold:,})" ) logger.warning("Using openpyxl parsing for better performance") # Parse with LangChain-based method (NOW ONLY ONCE, with annotations already applied) elements = self._parse_with_langchain( file_to_parse, max_characters, use_pipe_separators ) # Convert TextElementDTO objects to ChunkedElementDTO chunked_dtos = self._convert_documents_to_chunked_dtos( elements, file_id, original_filename, unique_filename ) # Create image chunks if requested image_count = 0 if image_extraction_result: image_dtos = self._convert_images_to_chunked_dtos( image_extraction_result, file_id, original_filename, unique_filename ) # Combine text chunks and image chunks chunked_dtos.extend(image_dtos) image_count = len(image_dtos) # Add shape chunks if shape_dtos: chunked_dtos.extend(shape_dtos) shape_count = len(shape_dtos) logger.info(f"Added {shape_count} shape chunks") # Create and add comment chunks comment_count = 0 if comment_id_mapping: comment_dtos = self._convert_comments_to_chunked_dtos( comment_id_mapping, file_id, original_filename, unique_filename ) chunked_dtos.extend(comment_dtos) comment_count = len(comment_dtos) logger.info(f"Added {comment_count} comment chunks") # Log summary if we added extra chunks if image_count > 0 or shape_count > 0 or comment_count > 0: text_count = len(chunked_dtos) - image_count - shape_count - comment_count summary_parts = [f"text: {text_count}"] if image_count > 0: summary_parts.append(f"images: {image_count}") if shape_count > 0: summary_parts.append(f"shapes: {shape_count}") if comment_count > 0: summary_parts.append(f"comments: {comment_count}") logger.info( f"Total chunks: {len(chunked_dtos)} ({', '.join(summary_parts)})" ) return chunked_dtos 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_path: str) -> str: """Convert .xlsm to .xlsx by removing macros and saving as standard Excel format. Args: xlsm_path: Path to the .xlsm file Returns: Path to the converted .xlsx file """ logger.info(f"Converting .xlsm to .xlsx: {xlsm_path}") try: # Load the .xlsm file with openpyxl # data_only=False to preserve formulas, keep_vba=False to remove macros workbook = load_workbook(xlsm_path, data_only=False, keep_vba=False) # Create output path with .xlsx extension xlsx_path = xlsm_path.replace('.xlsm', '.xlsx') # Save as .xlsx (macros will be removed automatically) workbook.save(xlsx_path) workbook.close() logger.info(f"Successfully converted .xlsm to .xlsx: {xlsx_path}") return xlsx_path except Exception as e: logger.error(f"Error converting .xlsm to .xlsx: {e}") raise def _convert_xls_to_xlsx_libreoffice(self, xls_path: str) -> str: """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_path: Path to the .xls file Returns: Path to the converted .xlsx file Raises: RuntimeError: If LibreOffice is not available or conversion fails """ import subprocess import shutil # 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.info(f"Converting .xls to .xlsx using LibreOffice: {xls_path}") try: # Get directory and filename dir_path = os.path.dirname(xls_path) # Run LibreOffice in headless mode to convert result = subprocess.run( [ libreoffice_cmd, '--headless', '--convert-to', 'xlsx', '--outdir', dir_path, 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 = xls_path.rsplit('.', 1)[0] + '.xlsx' if not os.path.exists(xlsx_path): raise RuntimeError("Conversion appeared to succeed but output file not found") logger.info(f"Successfully converted .xls to .xlsx with LibreOffice: {xlsx_path}") return xlsx_path except subprocess.TimeoutExpired: raise RuntimeError("LibreOffice conversion timed out") except Exception as e: logger.error(f"Error converting with LibreOffice: {e}") raise def _convert_xls_to_xlsx(self, xls_path: str) -> str: """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_path: Path to the .xls file Returns: Path to the converted .xlsx file """ # Try LibreOffice first (always) try: return self._convert_xls_to_xlsx_libreoffice(xls_path) except RuntimeError as e: logger.warning(f"LibreOffice conversion not available: {e}") logger.info("Falling back to basic xlrd conversion (no images/shapes)") # Fallback to xlrd if LibreOffice not available logger.info(f"Converting .xls file to .xlsx format using xlrd: {xls_path}") logger.warning("Note: Images, shapes, and formatting will not be preserved in conversion") try: # Read the .xls file using xlrd xls_workbook = xlrd.open_workbook(xls_path, formatting_info=False) # Create a new .xlsx workbook using openpyxl xlsx_workbook = Workbook() # Remove default sheet safely if xlsx_workbook.active: xlsx_workbook.remove(xlsx_workbook.active) # Convert each sheet for sheet_idx in range(xls_workbook.nsheets): xls_sheet = xls_workbook.sheet_by_index(sheet_idx) xlsx_sheet = xlsx_workbook.create_sheet(title=xls_sheet.name) # Copy all cell values for row_idx in range(xls_sheet.nrows): for col_idx in range(xls_sheet.ncols): cell_value = xls_sheet.cell_value(row_idx, col_idx) cell_type = xls_sheet.cell_type(row_idx, col_idx) # Convert xlrd cell types to appropriate Python types if cell_type == xlrd.XL_CELL_EMPTY: cell_value = None elif cell_type == xlrd.XL_CELL_TEXT: cell_value = str(cell_value) elif cell_type == xlrd.XL_CELL_NUMBER: cell_value = float(cell_value) elif cell_type == xlrd.XL_CELL_DATE: # Convert date to appropriate format try: cell_value = xlrd.xldate_as_datetime(float(cell_value), xls_workbook.datemode) except (ValueError, TypeError): # If conversion fails, keep as string cell_value = str(cell_value) elif cell_type == xlrd.XL_CELL_BOOLEAN: cell_value = bool(cell_value) elif cell_type == xlrd.XL_CELL_ERROR: cell_value = None # Write to xlsx (openpyxl uses 1-based indexing) try: xlsx_sheet.cell(row=row_idx + 1, column=col_idx + 1, value=cell_value) except Exception: # Skip cells that can't be written (e.g., merged cells) pass # Save to a temporary .xlsx file xlsx_path = xls_path.replace('.xls', '.xlsx') xlsx_workbook.save(xlsx_path) xlsx_workbook.close() logger.info(f"Successfully converted .xls to .xlsx: {xlsx_path}") return xlsx_path except Exception as e: logger.error(f"Error converting .xls to .xlsx: {e}") raise def _check_file_size(self, file_path: str) -> Tuple[int, int]: """Check the size of the Excel file. Args: file_path: Path to Excel file Returns: Tuple of (max_rows, total_cells) """ try: workbook = load_workbook(file_path, read_only=True, data_only=True) max_rows = 0 total_cells = 0 for sheet_name in workbook.sheetnames: sheet = workbook[sheet_name] rows = sheet.max_row or 0 cols = sheet.max_column or 0 max_rows = max(max_rows, rows) total_cells += rows * cols workbook.close() logger.info( f"File size check: max {max_rows:,} rows, total {total_cells:,} cells" ) return max_rows, total_cells except Exception as e: logger.warning(f"Could not check file size: {e}") return 0, 0 def _parse_with_langchain( self, file_path: str, max_characters: int, use_pipe_separators: bool = False ) -> List[TextElementDTO]: """Parse Excel file using Docling with HybridChunker for efficient chunking. Args: file_path: Path to Excel file max_characters: Maximum characters per chunk (deprecated, using tokenizer max_tokens instead) use_pipe_separators: Whether to use pipe characters as cell separators (deprecated, docling handles this) Returns: List of TextElementDTO elements """ 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 logger.info("Document conversion complete") # Apply chunking with HybridChunker logger.info("Chunking document with HybridChunker...") chunker = HybridChunker( tokenizer=self.tokenizer, serializer_provider=ExcelTableAnnotationSerializerProvider(), ) chunk_iter = chunker.chunk(dl_doc=doc) chunks = list(chunk_iter) logger.info(f"Created {len(chunks)} chunks from Docling") # Convert chunks to TextElementDTO objects all_elements = [] for chunk in chunks: # Get contextualized text from chunk text = chunker.contextualize(chunk=chunk) # Skip empty chunks if not text.strip(): continue # Validate chunk as DocChunk to access metadata doc_chunk = DocChunk.model_validate(chunk) # Extract metadata from chunk - extract sheet name and number sheet_name = None sheet_number = None # Try to extract sheet info from doc items for item in doc_chunk.meta.doc_items: if hasattr(item, 'prov') and item.prov: for prov_item in item.prov: # For Excel files, page_no corresponds to sheet index if hasattr(prov_item, 'page_no') and prov_item.page_no is not None: sheet_number = prov_item.page_no break if sheet_number is not None: break # Try to get sheet name from document metadata or use sheet number as fallback if sheet_number is not None and hasattr(doc, '_md_table_of_contents') and doc._md_table_of_contents: # Try to map sheet number to sheet name if available # For now, use generic naming sheet_name = f"Sheet{sheet_number}" # Create metadata using ElementMetadataDTO metadata = ElementMetadataDTO( sheet_name=sheet_name, sheet_number=sheet_number, ) # Create Text element text_element = TextElementDTO( text=text, metadata=metadata, ) all_elements.append(text_element) logger.info(f"Converted {len(all_elements)} chunks to TextElementDTO objects") return all_elements except Exception as e: logger.error(f"Error parsing with Docling: {e}") raise def _convert_documents_to_chunked_dtos( self, elements: List[TextElementDTO], file_id: UUID, original_filename: str, unique_filename: Optional[str] = None, ) -> List[ChunkedElementDTO]: """Convert TextElementDTO objects to ChunkedElementDTO objects. Args: elements: List of TextElementDTO elements file_id: PostgreSQL UUID of the file record original_filename: Original filename unique_filename: Optional unique filename (defaults to original_filename if not provided) Returns: List of ChunkedElementDTO objects """ if not elements: return [] # Extract file type from filename _, ext = os.path.splitext(original_filename) file_type = ext.lstrip('.').lower() if ext else 'unknown' dtos = [] for element in elements: # Get text content text = str(element) # Extract metadata from element metadata: Dict[str, Any] = {} if hasattr(element, 'metadata'): meta_obj = element.metadata metadata['sheet_name'] = getattr(meta_obj, 'page_name', None) or getattr(meta_obj, 'sheet_name', None) metadata['sheet_number'] = getattr(meta_obj, 'page_number', None) or getattr(meta_obj, 'sheet_number', None) # Add file context to metadata metadata['source_filename'] = original_filename # Use unique_filename if provided, otherwise fall back to original_filename metadata['unique_filename'] = unique_filename if unique_filename else original_filename # Create DTO dto = ChunkedElementDTO( file_id=file_id, element_type="Text", file_type=file_type, text=text, metadata=metadata, ) dtos.append(dto) return dtos async def _extract_images_with_descriptions( self, file_content: bytes, file_id: UUID, original_filename: str, description_detail: str = "auto", is_converted_from_xls: bool = False, ) -> Dict[str, Any]: """Extract images from Excel file and generate AI descriptions. Args: file_content: Binary content of the Excel file file_id: PostgreSQL UUID of the file record original_filename: Original filename description_detail: Detail level for AI descriptions ("auto", "low", "high") is_converted_from_xls: If True, file_content is .xlsx converted from .xls Returns: Dictionary containing image extraction results from the image service """ try: # If file was converted from .xls, use .xlsx extension for temp file filename_for_extraction = original_filename if is_converted_from_xls: # Change extension to .xlsx so openpyxl can read it filename_for_extraction = os.path.splitext(original_filename)[0] + '.xlsx' logger.debug(f"Using .xlsx extension for converted file: {filename_for_extraction}") # Extract images with descriptions result = await self.image_extraction_service.extract_images_from_file( file_content=file_content, file_id=str(file_id), # Convert UUID to string for image service original_filename=filename_for_extraction, generate_descriptions=True, description_detail=description_detail, ) if result and "images" in result: logger.info(f"Extracted {len(result['images'])} images with descriptions") else: logger.info("No images found in Excel file") return result except Exception as e: logger.error(f"Error extracting images and descriptions: {e}") return {} def _create_position_descriptions_map( self, image_extraction_result: Dict[str, Any] ) -> Dict[str, str]: """Create a mapping of position keys to image descriptions. Args: image_extraction_result: Result from image extraction service Returns: Dictionary mapping position keys to descriptions """ position_descriptions = {} for img_meta in image_extraction_result.get("images", []): # Create position key from sheet and image info sheet_name = img_meta.get("sheet_name", "Unknown") position = img_meta.get("position", {}) # Try to create a meaningful position key if position.get("from_cell"): pos_key = f"{sheet_name}_{position['from_cell']}" else: img_num = img_meta.get("image_number", 1) pos_key = f"{sheet_name}_IMG{img_num}" # Get description description = img_meta.get("description", "[No description available]") position_descriptions[pos_key] = description return position_descriptions def _convert_images_to_chunked_dtos( self, image_extraction_result: Dict[str, Any], file_id: UUID, original_filename: str, unique_filename: Optional[str] = None, ) -> List[ChunkedElementDTO]: """Convert image extraction results to ChunkedElementDTO objects. Args: image_extraction_result: Result from image extraction service file_id: PostgreSQL UUID of the file record original_filename: Original filename unique_filename: Optional unique filename (defaults to original_filename if not provided) Returns: List of ChunkedElementDTO objects, one per image """ if not image_extraction_result or "images" not in image_extraction_result: return [] # Extract file type _, ext = os.path.splitext(original_filename) file_type = ext.lstrip(".").lower() if ext else "xlsx" # Create ChunkedElementDTO for each image image_chunks = [] for idx, img_meta in enumerate(image_extraction_result.get("images", [])): # Extract image metadata sheet_name = img_meta.get("sheet_name", "Unknown") sheet_number = img_meta.get("sheet_number") position = img_meta.get("position", {}) description = img_meta.get("description") or "[No description available]" image_number = img_meta.get("image_number", idx + 1) # Create position key for reference if position.get("from_cell"): pos_key = f"{sheet_name}_{position['from_cell']}" cell_ref = position["from_cell"] else: pos_key = f"{sheet_name}_IMG{image_number}" cell_ref = None # Build metadata metadata = { "source_filename": original_filename, "sheet_name": sheet_name, "sheet_number": sheet_number, "image_number": image_number, "position_key": pos_key, "image_format": img_meta.get("format", "unknown"), "width": img_meta.get("width"), "height": img_meta.get("height"), } # Add position details if available if position: metadata["position"] = position if cell_ref: metadata["cell_reference"] = cell_ref metadata["from_cell"] = position.get("from_cell") metadata["to_cell"] = position.get("to_cell") else: metadata["position"] = None # Base64 data will only be used for storing images in MinIO if "base64_data" in img_meta: metadata["image_base64"] = img_meta["base64_data"] # Add MIME type if available if "mime_type" in img_meta: metadata["mime_type"] = img_meta["mime_type"] else: metadata["mime_type"] = "application/octet-stream" # Use unique_filename if provided, otherwise fall back to original_filename metadata["unique_filename"] = unique_filename or original_filename # Create ChunkedElementDTO chunk = ChunkedElementDTO( file_id=file_id, element_type="Image", file_type=file_type, text=description, metadata=metadata, ) image_chunks.append(chunk) logger.info(f"Created {len(image_chunks)} image chunks") return image_chunks def _replace_images_with_descriptions( self, file_path: str, position_descriptions: Dict[str, str] ) -> str: """Create a new Excel file with images replaced by their descriptions. Args: file_path: Path to original Excel file position_descriptions: Dictionary of position to description mappings Returns: Path to modified Excel file with descriptions instead of images """ if not position_descriptions: logger.info("No image descriptions to replace") return file_path logger.info( f"Replacing {len(position_descriptions)} images with descriptions in Excel file" ) try: # Load the workbook workbook = load_workbook(file_path) replacements_made = 0 unknown_images = [] # Process each position description for pos_key, description in position_descriptions.items(): try: # Parse position key to get sheet name and coordinates sheet_name, cell_ref = self._parse_position_key(pos_key) # Handle images without proper sheet/cell information if sheet_name == "Unknown" or not cell_ref: logger.warning( f"Image {pos_key} has no sheet/cell information - will skip" ) continue if sheet_name not in workbook.sheetnames: logger.warning( f"Sheet '{sheet_name}' not found in workbook. " f"Available sheets: {workbook.sheetnames}" ) unknown_images.append((pos_key, description)) continue worksheet = workbook[sheet_name] # Add description text to the cell cell = worksheet[cell_ref] current_value = str(cell.value) if cell.value else "" # Prepend description to existing content description_text = f"[Image Description]: {description}" if current_value and current_value != "None": cell.value = f"{description_text}\n{current_value}" else: cell.value = description_text replacements_made += 1 logger.info( f"✓ Successfully added description to cell {cell_ref} in sheet '{sheet_name}'" ) except Exception as e: logger.debug(f"Could not process position {pos_key}: {e}") unknown_images.append((pos_key, description)) # Handle unknown images by adding them to a summary cell in the first sheet if unknown_images and workbook.sheetnames: first_sheet = workbook[workbook.sheetnames[0]] # Find an empty cell to add the unknown image descriptions summary_cell = first_sheet.cell( row=1, column=50 ) # Column AT (arbitrary empty location) summary_text = "Images without specific location:\n\n" for pos_key, description in unknown_images: summary_text += f"[{pos_key}]: {description}\n\n" replacements_made += 1 summary_cell.value = summary_text logger.info( f"✓ Added {len(unknown_images)} unknown images to summary cell in '{workbook.sheetnames[0]}'" ) # Save the modified workbook to a new temporary file modified_excel_path = file_path.replace(".xlsx", "_with_descriptions.xlsx") workbook.save(modified_excel_path) workbook.close() logger.info( f"Successfully replaced {replacements_made} images with descriptions" ) return modified_excel_path except Exception as e: logger.error(f"Error replacing images with descriptions: {e}") return file_path def _parse_position_key(self, pos_key: str) -> Tuple[str, str]: """Parse position key to extract sheet name and cell reference. Args: pos_key: Position key in format "SheetName_CellRef" (e.g., "Sheet1_VY141") Returns: Tuple of (sheet_name, cell_reference) """ try: if "_" not in pos_key: return pos_key, "" # Fallback to sheet name only parts = pos_key.split("_", 1) # Split only on first underscore sheet_name = parts[0] cell_ref = parts[1] if len(parts) > 1 else "" # Validate cell reference format (letters followed by numbers) import re if cell_ref and re.match(r"^[A-Z]+\d+$", cell_ref): return sheet_name, cell_ref else: return sheet_name, "" except Exception as e: logger.debug(f"Could not parse position key {pos_key}: {e}") return "Unknown", "" def _apply_shape_annotations_to_cells( self, file_path: str, shape_dtos: List[ChunkedElementDTO] ) -> str: """Apply shape annotations to cells in the Excel file. Logic: 1. If shape has connected_target: Add annotation to target cell 2. If shape has no target: Check bounding box cells - If all cells empty: Skip - If any cells have values: Add annotation to each non-empty cell Args: file_path: Path to Excel file shape_dtos: List of shape DTOs with metadata Returns: Path to modified Excel file with annotations """ try: logger.info(f"Applying shape annotations to {file_path}") # Load with data_only=True to get calculated values instead of formulas # This prevents loss of data when saving workbook = load_workbook(file_path, data_only=True) annotations_applied = 0 for shape_dto in shape_dtos: try: metadata = shape_dto.metadata sheet_name = metadata.get("sheet_name") shape_text = shape_dto.text if not sheet_name or sheet_name not in workbook.sheetnames: logger.warning(f"Sheet '{sheet_name}' not found, skipping shape") continue worksheet = workbook[sheet_name] # Check if shape has a connector target if "connected_target" in metadata and metadata["connected_target"]: target = metadata["connected_target"] cell_ref = target.get("cell_ref") if cell_ref: # Apply annotation to target cell cell = worksheet[cell_ref] # IMPORTANT: Check for None explicitly to preserve 0 and other falsy values current_value = str(cell.value) if cell.value is not None else "" # Create HTML-like annotation tag # Format: cell_value if current_value and current_value != "None": cell.value = f'{current_value}' else: cell.value = f'' annotations_applied += 1 logger.debug(f"Added annotation to target cell {cell_ref} in '{sheet_name}'") else: # No connector - check bounding box position = metadata.get("position", {}) from_cell = position.get("from_cell") to_cell = position.get("to_cell") if not from_cell or not to_cell: logger.debug(f"Shape has no position info, skipping") continue # Get cell range from openpyxl.utils import range_boundaries min_col, min_row, max_col, max_row = range_boundaries(f"{from_cell}:{to_cell}") # Check if any cells in range have values cells_with_values = [] for row in range(min_row, max_row + 1): for col in range(min_col, max_col + 1): cell = worksheet.cell(row=row, column=col) # IMPORTANT: Check for None explicitly to include 0, False, etc. if cell.value is not None and str(cell.value).strip(): cells_with_values.append(cell) # If any cells have values, annotate them if cells_with_values: for cell in cells_with_values: # IMPORTANT: Use 'is not None' to preserve 0 and other falsy values current_value = str(cell.value) if cell.value is not None else "" # Create HTML-like annotation tag cell.value = f'{current_value}' annotations_applied += 1 logger.debug( f"Added annotation to {len(cells_with_values)} cells in range " f"{from_cell}:{to_cell} in '{sheet_name}'" ) else: # All cells empty - merge the range and put shape text as value logger.debug( f"All cells empty in range {from_cell}:{to_cell}, merging cells and adding shape text" ) try: # Check if any cells in the range are already part of a merged cell from openpyxl.worksheet.cell_range import CellRange range_to_merge = CellRange(f"{from_cell}:{to_cell}") # Check for conflicts with existing merged cells is_conflicting = False for merged_range in worksheet.merged_cells.ranges: # Check if ranges overlap (but not identical to themselves) try: intersection = range_to_merge.intersection(merged_range) # intersection returns None if no overlap, or a CellRange if there is overlap if intersection is not None: is_conflicting = True logger.debug( f"Range {from_cell}:{to_cell} conflicts with existing merged range {merged_range}, skipping merge" ) break except ValueError: # No intersection - ranges don't overlap continue if not is_conflicting: # Merge cells worksheet.merge_cells(f"{from_cell}:{to_cell}") # Set the shape text in the top-left cell of the merged range top_left_cell = worksheet[from_cell] top_left_cell.value = shape_text # Optional: Center align the text in merged cell from openpyxl.styles import Alignment top_left_cell.alignment = Alignment(horizontal='center', vertical='center', wrap_text=True) annotations_applied += 1 logger.debug(f"Merged {from_cell}:{to_cell} and added shape text in '{sheet_name}'") else: # Can't merge, just put text in top-left cell top_left_cell = worksheet[from_cell] if not isinstance(top_left_cell, type(worksheet['A1']).__bases__[0]): # Check if not MergedCell top_left_cell.value = shape_text annotations_applied += 1 logger.debug(f"Added shape text to {from_cell} (merge skipped due to conflict) in '{sheet_name}'") except Exception as merge_error: logger.warning(f"Error merging cells {from_cell}:{to_cell}: {merge_error}") # Fallback: just try to set value in top-left cell try: top_left_cell = worksheet[from_cell] top_left_cell.value = shape_text annotations_applied += 1 except Exception as fallback_error: logger.warning(f"Failed to set value in {from_cell}: {fallback_error}") except Exception as e: logger.warning(f"Error applying annotation for shape: {e}") continue if annotations_applied > 0: # Save modified workbook annotated_path = file_path.replace(".xlsx", "_with_annotations.xlsx") workbook.save(annotated_path) workbook.close() logger.info(f"Applied {annotations_applied} shape annotations, saved to {annotated_path}") return annotated_path else: workbook.close() logger.info("No annotations applied") return file_path except Exception as e: logger.error(f"Error applying shape annotations: {e}") return file_path def _extract_comments_from_file(self, file_path: str) -> Dict[str, List[Dict[str, Any]]]: """Extract all comments from an Excel file with full metadata. Args: file_path: Path to Excel file Returns: Dictionary mapping {sheet_name: [list of comment dictionaries]} Each comment dict contains: cell_address, row, column, column_index, cell_value, comment_text, author """ try: logger.info(f"Extracting comments from {file_path}") workbook = load_workbook(file_path, data_only=False) comments_by_sheet = {} total_comments = 0 for sheet_name in workbook.sheetnames: sheet = workbook[sheet_name] sheet_comments = [] # Iterate through all cells to find comments for row in sheet.iter_rows(): for cell in row: if cell.comment: # Extract comment text comment_text = cell.comment.text if hasattr(comment_text, "text"): # If it's a RichText object comment_text = str(comment_text) # Clean the comment text comment_text = comment_text.strip() if comment_text: # Get author author = None if hasattr(cell.comment, "author"): author = cell.comment.author # Get cell value cell_value = str(cell.value) if cell.value is not None else "" comment_data = { "cell_address": cell.coordinate, "row": cell.row, "column": cell.column_letter, "column_index": cell.column, "cell_value": cell_value, "comment_text": comment_text, "author": author, } sheet_comments.append(comment_data) total_comments += 1 logger.debug( f"Found comment at {sheet_name}!{cell.coordinate}: " f"{comment_text[:50]}..." ) if sheet_comments: comments_by_sheet[sheet_name] = sheet_comments workbook.close() logger.info(f"Extracted {total_comments} comments from {len(comments_by_sheet)} sheets") return comments_by_sheet except Exception as e: logger.error(f"Error extracting comments: {e}") return {} def _apply_comments_to_cells( self, file_path: str, comments_by_sheet: Dict[str, List[Dict[str, Any]]] ) -> Tuple[str, Dict[str, Dict[str, Any]]]: """Apply comment ID markers to cells in the Excel file. Args: file_path: Path to Excel file comments_by_sheet: Dictionary mapping {sheet_name: [list of comment dicts]} Returns: Tuple of (path to modified Excel file, comment_id_mapping) comment_id_mapping: {comment_id: comment_metadata} """ try: logger.info(f"Applying comment ID markers to {file_path}") # Load with data_only=True to get calculated values instead of formulas workbook = load_workbook(file_path, data_only=True) comments_applied = 0 comment_id_mapping = {} # {comment_id: full_comment_metadata} for sheet_name, sheet_comments in comments_by_sheet.items(): if sheet_name not in workbook.sheetnames: logger.warning(f"Sheet '{sheet_name}' not found, skipping comments") continue worksheet = workbook[sheet_name] for comment_data in sheet_comments: try: cell_ref = comment_data["cell_address"] cell = worksheet[cell_ref] # Generate unique comment ID import uuid comment_id = f"comment_{uuid.uuid4().hex[:8]}" # Get current cell value # IMPORTANT: Check for None explicitly to preserve 0 and other falsy values current_value = str(cell.value) if cell.value is not None else "" # Create HTML-like comment tag # Format: cell_value if current_value and current_value != "None": cell.value = f'{current_value}' else: cell.value = f'' # Store comment metadata with ID comment_id_mapping[comment_id] = { "comment_id": comment_id, "sheet_name": sheet_name, "cell_address": cell_ref, "row": comment_data["row"], "column": comment_data["column"], "column_index": comment_data["column_index"], "cell_value": comment_data["cell_value"], "comment_text": comment_data["comment_text"], "author": comment_data.get("author"), } comments_applied += 1 logger.debug( f"Added comment marker {comment_id} to cell {cell_ref} in '{sheet_name}'" ) except Exception as e: logger.warning( f"Failed to apply comment to {sheet_name}!{cell_ref}: {e}" ) continue logger.info(f"Successfully applied {comments_applied} comment markers") # Save to a new temporary file import tempfile temp_fd, temp_path = tempfile.mkstemp(suffix=".xlsx") os.close(temp_fd) workbook.save(temp_path) workbook.close() logger.info(f"Saved modified file with comment markers to {temp_path}") return temp_path, comment_id_mapping except Exception as e: logger.error(f"Error applying comment markers: {e}") return file_path, {} def _convert_comments_to_chunked_dtos( self, comment_id_mapping: Dict[str, Dict[str, Any]], file_id: UUID, original_filename: str, unique_filename: Optional[str] = None, ) -> List[ChunkedElementDTO]: """Convert comment metadata to ChunkedElementDTO objects. Args: comment_id_mapping: Dictionary mapping {comment_id: comment_metadata} file_id: PostgreSQL UUID of the file record original_filename: Original filename unique_filename: Optional unique filename Returns: List of ChunkedElementDTO objects for comments """ comment_dtos = [] for comment_id, comment_meta in comment_id_mapping.items(): try: # Determine file extension _, ext = os.path.splitext(original_filename) file_type = ext.lstrip(".").lower() if ext else "xlsx" # Create metadata metadata = { "comment_id": comment_id, "sheet_name": comment_meta["sheet_name"], "sheet_number": None, # We don't track sheet number during extraction "cell_address": comment_meta["cell_address"], "row": comment_meta["row"], "column": comment_meta["column"], "column_index": comment_meta["column_index"], "cell_value": comment_meta["cell_value"], "author": comment_meta.get("author"), "source_filename": original_filename, "unique_filename": unique_filename or original_filename, } # Create ChunkedElementDTO with comment text comment_dto = ChunkedElementDTO( file_id=file_id, element_type="Comment", file_type=file_type, text=comment_meta["comment_text"], metadata=metadata, ) comment_dtos.append(comment_dto) logger.debug( f"Created comment chunk {comment_id} for cell " f"{comment_meta['sheet_name']}!{comment_meta['cell_address']}" ) except Exception as e: logger.warning(f"Failed to create comment DTO for {comment_id}: {e}") continue logger.info(f"Created {len(comment_dtos)} comment chunks") return comment_dtos @lru_cache() def get_excel_parser_service() -> ExcelParserService: from src.services.excel_image_extraction_service import ( get_excel_image_extraction_service, ) from src.services.excel_shape_extraction_service import ( get_excel_shape_extraction_service, ) return ExcelParserService( image_extraction_service=get_excel_image_extraction_service(), shape_extraction_service=get_excel_shape_extraction_service(), )