"""Service for parsing Excel documents via ks_xlsx_parser. This service provides comprehensive Excel parsing capabilities using: - ks_xlsx_parser for form-layout-aware Excel→HTML→Markdown conversion - Image extraction with AI-generated descriptions - Shape extraction and cell annotations - Mermaid flowchart detection (shapes→mermaid cells) """ import os import tempfile from functools import lru_cache from typing import Any, Dict, List, Optional from uuid import UUID import structlog from src.core.config import settings from src.dtos.document_dtos import ChunkedElementDTO, ElementMetadataDTO, TextElementDTO from src.extraction_v2.azure_chat_service import AzureChatService from src.extraction_v2.excel_image_extraction_service import ExcelImageExtractionService from src.extraction_v2.ks_parser import ksparser_to_sheets from src.extraction_v2.shape_extractor import ShapeExtractor logger = structlog.get_logger(__name__) class ExcelParserService: """Service for parsing Excel files using ks_xlsx_parser with image and shape preprocessing.""" def __init__( self, image_extraction_service: ExcelImageExtractionService, shape_extractor: ShapeExtractor, ): """Initialize the Excel parser service. Args: image_extraction_service: Service for extracting images from Excel files shape_extractor: Service for extracting shapes from Excel files """ self.image_extraction_service = image_extraction_service self.shape_extractor = shape_extractor async def parse_excel_file( self, file_content: bytes, file_id: UUID, original_filename: str, max_characters: int = 4000, ) -> List[ChunkedElementDTO]: """Parse an Excel file and extract chunked elements via ks_xlsx_parser. This method implements a hybrid approach: 1. Pre-process images: Extract and generate AI descriptions, replace in Excel 2. Pre-process shapes: Extract and apply as cell annotations 3. Flowchart detection: shapes→mermaid cells (STEP 1.5) 4. Parse with ks_xlsx_parser, one DTO per sheet (no splitting) Args: file_content: Binary content of the Excel file file_id: PostgreSQL UUID of the file record original_filename: Original filename max_characters: Maximum characters per chunk (default: 4000) Returns: List of ChunkedElementDTO objects (text + image chunks, NO shape chunks) """ temp_file_path = None modified_file_path = None try: # Determine file extension _, ext = os.path.splitext(original_filename) suffix = ext if ext in ['.xlsx', '.xls', '.xlsm'] else '.xlsx' # Create temporary file temp_fd, temp_file_path = tempfile.mkstemp(suffix=suffix) # Write content (sync - no need for async file I/O here) with os.fdopen(temp_fd, 'wb') as temp_file: temp_file.write(file_content) logger.info(f"Parsing Excel file: {original_filename}") # File to parse (will be modified during pre-processing) file_to_parse = temp_file_path # File content to use for image extraction (may be converted) file_content_for_extraction = file_content # Convert .xls/.xlsm to .xlsx if needed (ks_xlsx_parser only supports .xlsx) if suffix.lower() in ['.xls', '.xlsm']: logger.info(f"Converting {suffix} to .xlsx format using LibreOffice...") converted_file_path = self._convert_xls_to_xlsx(temp_file_path) file_to_parse = converted_file_path # Track for cleanup if modified_file_path is None: modified_file_path = converted_file_path logger.info(f"Conversion complete: {converted_file_path}") # Read converted file content for image extraction with open(converted_file_path, 'rb') as f: file_content_for_extraction = f.read() # STEP 1: Image pre-processing (ALWAYS done) logger.info("Extracting images and generating descriptions...") image_extraction_result = await self._extract_images_with_descriptions( file_content_for_extraction, file_id, original_filename ) if image_extraction_result and image_extraction_result.get('images'): # Create position descriptions map 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 ) file_to_parse = modified_file_path logger.info(f"Image replacement complete, file to parse: {file_to_parse}") # STEP 1.5: Flowchart detection (before shape extraction) # Converts drawing shapes (xdr:sp/xdr:cxnSp) → mermaid syntax → written into cells logger.info("Detecting flowcharts...") flowchart_result = self._detect_and_extract_flowchart(file_to_parse) if flowchart_result: flowchart_file_path, sheets_with_flowcharts = flowchart_result file_to_parse = flowchart_file_path if modified_file_path is None: modified_file_path = flowchart_file_path logger.info( "Flowchart mermaid embedded for %d sheet(s): %s", len(sheets_with_flowcharts), sheets_with_flowcharts, ) # STEP 2: Shape pre-processing (extract and apply annotations) logger.info("Extracting shapes and applying annotations...") # Create temp directory for shape processing temp_dir = tempfile.mkdtemp() shape_temp_dir_to_cleanup = None # Extract shapes and apply annotations to cells annotated_file = self.shape_extractor.extract_and_apply_annotations( file_to_parse, temp_dir ) if annotated_file: logger.info("Shape annotations applied successfully") # Track temp dir for cleanup AFTER parsing shape_temp_dir_to_cleanup = temp_dir # Update file to parse if modified_file_path is None: modified_file_path = annotated_file file_to_parse = annotated_file else: logger.info("No shape annotations to apply") import shutil try: shutil.rmtree(temp_dir) except Exception as e: logger.debug(f"Failed to remove temp directory: {e}") # STEP 4/5: Parse with ks_xlsx_parser; one DTO per sheet, no splitting logger.info("Parsing Excel with ks_xlsx_parser...") try: sheets = ksparser_to_sheets(file_to_parse) except Exception as e: logger.error(f"ks_xlsx_parser failed: {e}") import traceback logger.debug(traceback.format_exc()) sheets = [] # Clean up shape temp directory now that parsing is done if shape_temp_dir_to_cleanup: import shutil try: shutil.rmtree(shape_temp_dir_to_cleanup) logger.debug(f"Cleaned up shape temp directory: {shape_temp_dir_to_cleanup}") except Exception as e: logger.debug(f"Failed to remove shape temp directory: {e}") # Extract file type _, ext_lower = os.path.splitext(original_filename) file_type = ext_lower.lstrip('.').lower() if ext_lower else 'xlsx' # Build one ChunkedElementDTO per sheet chunked_dtos: List[ChunkedElementDTO] = [] for sheet_entry in sheets: sheet_name = sheet_entry['sheet_name'] markdown = sheet_entry['markdown'] if not markdown or not markdown.strip(): continue metadata: Dict[str, Any] = { 'source_filename': original_filename, 'unique_filename': original_filename, 'sheet_name': sheet_name, } dto = ChunkedElementDTO( file_id=file_id, element_type="Text", file_type=file_type, text=markdown, metadata=metadata, ) chunked_dtos.append(dto) logger.info(f"Created {len(chunked_dtos)} text chunks from {len(sheets)} sheets") # STEP 6: Create image chunks (if images were extracted) image_count = 0 if image_extraction_result and image_extraction_result.get('images'): image_dtos = self._convert_images_to_chunked_dtos( image_extraction_result, file_id, original_filename ) chunked_dtos.extend(image_dtos) image_count = len(image_dtos) # Log summary if image_count > 0: text_count = len(chunked_dtos) - image_count logger.info( f"Total chunks: {len(chunked_dtos)} " f"(text: {text_count}, images: {image_count})" ) # IMPORTANT: Shape chunks are NOT returned (shapes only used for annotations) 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}") async def parse_to_markdown( self, file_content: bytes, file_id: UUID, original_filename: str, ) -> Dict[str, Any]: """Parse Excel file and return per-sheet markdown without chunking. Runs the same pre-processing pipeline as parse_excel_file (images, shapes, flowcharts) but stops at markdown export. Chunking is deferred to the downstream hierarchical chunking task. Args: file_content: Binary content of the Excel file file_id: PostgreSQL UUID of the file record original_filename: Original filename Returns: Dict with: - sheets: List of dicts, each with 'sheet_name' and 'markdown' - image_chunks: List of ChunkedElementDTO for images (with base64) - flowchart_chunks: List of ChunkedElementDTO for flowcharts """ temp_file_path = None modified_file_path = None try: # Determine file extension _, ext = os.path.splitext(original_filename) suffix = ext if ext in ['.xlsx', '.xls', '.xlsm'] else '.xlsx' # Create temporary file temp_fd, temp_file_path = tempfile.mkstemp(suffix=suffix) with os.fdopen(temp_fd, 'wb') as temp_file: temp_file.write(file_content) logger.info(f"Parsing Excel file to markdown: {original_filename}") file_to_parse = temp_file_path file_content_for_extraction = file_content # Convert .xls/.xlsm to .xlsx if needed if suffix.lower() in ['.xls', '.xlsm']: logger.info(f"Converting {suffix} to .xlsx format...") converted_file_path = self._convert_xls_to_xlsx(temp_file_path) file_to_parse = converted_file_path if modified_file_path is None: modified_file_path = converted_file_path with open(converted_file_path, 'rb') as f: file_content_for_extraction = f.read() # STEP 1: Image pre-processing logger.info("Extracting images and generating descriptions...") image_extraction_result = await self._extract_images_with_descriptions( file_content_for_extraction, file_id, original_filename ) if image_extraction_result and image_extraction_result.get('images'): 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 ) file_to_parse = modified_file_path # STEP 1.5: Flowchart detection # Converts drawing shapes (xdr:sp/xdr:cxnSp) → mermaid syntax → written into cells. # ks_xlsx_parser then reads those cells via collect_mermaid_blocks/restore_mermaid_blocks. logger.info("Detecting flowcharts...") flowchart_result = self._detect_and_extract_flowchart(file_to_parse) if flowchart_result: flowchart_file_path, sheets_with_flowcharts = flowchart_result file_to_parse = flowchart_file_path if modified_file_path is None: modified_file_path = flowchart_file_path logger.info( "Flowchart mermaid embedded for %d sheet(s): %s", len(sheets_with_flowcharts), sheets_with_flowcharts, ) # STEP 2: Shape pre-processing logger.info("Extracting shapes and applying annotations...") temp_dir = tempfile.mkdtemp() shape_temp_dir_to_cleanup = None annotated_file = self.shape_extractor.extract_and_apply_annotations( file_to_parse, temp_dir ) if annotated_file: shape_temp_dir_to_cleanup = temp_dir if modified_file_path is None: modified_file_path = annotated_file file_to_parse = annotated_file else: import shutil try: shutil.rmtree(temp_dir) except Exception: pass # STEP 4/5: Parse with ks_xlsx_parser (replaces Docling + sentinel injection) logger.info("Parsing with ks_xlsx_parser...") try: sheets = ksparser_to_sheets(file_to_parse) except Exception as e: logger.error(f"ks_xlsx_parser failed: {e}") import traceback logger.debug(traceback.format_exc()) return {'sheets': [], 'image_chunks': [], 'flowchart_chunks': []} # Clean up shape temp directory if shape_temp_dir_to_cleanup: import shutil try: shutil.rmtree(shape_temp_dir_to_cleanup) except Exception: pass logger.info( f"Parse to markdown complete: {len(sheets)} sheets, " f"{sum(len(s['markdown']) for s in sheets)} total chars" ) # STEP 6: Create image chunks (with base64 for SeaweedFS upload) image_chunks = [] if image_extraction_result and image_extraction_result.get('images'): image_chunks = self._convert_images_to_chunked_dtos( image_extraction_result, file_id, original_filename ) # Flowcharts are embedded as mermaid cells in the workbook before ks parses, # so they show up in `sheets` already. The `flowchart_chunks` key is preserved # (empty) for downstream callers that may still read it. flowchart_chunks: List[ChunkedElementDTO] = [] logger.info( f"Parse to markdown complete: {len(sheets)} sheets, " f"{len(image_chunks)} images" ) return { 'sheets': sheets, 'image_chunks': image_chunks, 'flowchart_chunks': flowchart_chunks, } finally: if temp_file_path: try: os.remove(temp_file_path) except Exception: pass if modified_file_path and modified_file_path != temp_file_path: try: os.remove(modified_file_path) except Exception: pass async def _extract_images_with_descriptions( self, file_content: bytes, file_id: UUID, original_filename: str, ) -> 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 Returns: Dictionary containing image extraction results """ try: # Extract images with descriptions result = await self.image_extraction_service.extract_images_from_file( file_content=file_content, file_id=str(file_id), original_filename=original_filename, generate_descriptions=True, description_detail="auto", ) 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 _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") try: from openpyxl import load_workbook # Load the workbook workbook = load_workbook(file_path) replacements_made = 0 # Process each position description for pos_key, description in position_descriptions.items(): try: # Parse position key if "_" not in pos_key: continue parts = pos_key.split("_", 1) sheet_name = parts[0] cell_ref = parts[1] if len(parts) > 1 else "" if sheet_name not in workbook.sheetnames: logger.warning(f"Sheet '{sheet_name}' not found") continue worksheet = workbook[sheet_name] # Validate cell reference import re if cell_ref and re.match(r"^[A-Z]+\d+$", cell_ref): # Image has a specific cell position - add description to that cell cell = worksheet[cell_ref] current_value = str(cell.value) if cell.value else "" # Prepend description 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 else: # Image doesn't have a cell position (e.g., floating image) # Add description to first empty cell in the sheet (A1, A2, A3, etc.) logger.debug(f"Image without cell position: {pos_key}, adding to first empty cell") # Find first empty cell in column A row_num = 1 while worksheet[f'A{row_num}'].value is not None: row_num += 1 description_text = f"[Image Description]: {description}" worksheet[f'A{row_num}'].value = description_text replacements_made += 1 logger.debug(f"Added image description to A{row_num} in sheet '{sheet_name}'") except Exception as e: logger.debug(f"Could not process position {pos_key}: {e}") # Save modified workbook modified_path = file_path.replace(".xlsx", "_with_descriptions.xlsx") workbook.save(modified_path) workbook.close() logger.info(f"Replaced {replacements_made} images with descriptions") return modified_path except Exception as e: logger.error(f"Error replacing images: {e}") return file_path def _detect_and_extract_flowchart( self, file_path: str, ) -> Optional[tuple[str, List[str]]]: """Embed per-graph mermaid blocks and strip drawing shapes in-place. For each worksheet: 1. Detect every connected shape+connector component via ``extract_mermaid_graphs``. 2. Convert each component to a fenced ```mermaid``` block whose top-left cell is the embed target. 3. Clear any worksheet cells whose text was mined as edge-label content so the same text doesn't leak into the downstream markdown alongside the embedded mermaid. 4. Delete all xdr:sp / xdr:cxnSp anchors from drawing XML. Args: file_path: Path to .xlsx file Returns: Tuple of (modified_file_path, sheets_with_flowcharts) when at least one sheet produced a mermaid block; None otherwise. """ try: from src.extraction_v2.mermaid_flowchart_extractor import extract_mermaid_graphs per_sheet = extract_mermaid_graphs(file_path) embeddings_by_sheet: Dict[str, List[tuple[str, str]]] = {} consumed_by_sheet: Dict[str, List[str]] = {} for sheet_name, (graphs, consumed) in per_sheet.items(): if not graphs: continue pairs: List[tuple[str, str]] = [] for g in graphs: cell_ref = g.location.split(":", 1)[0] if g.location else "A1" pairs.append((cell_ref, f"```mermaid\n{g.mermaid_diagram}\n```")) embeddings_by_sheet[sheet_name] = pairs if consumed: consumed_by_sheet[sheet_name] = consumed if not embeddings_by_sheet: logger.info("No flowcharts detected") return None sheets_with_flowcharts = list(embeddings_by_sheet.keys()) logger.info( "Flowcharts detected on %d sheet(s): %s", len(sheets_with_flowcharts), sheets_with_flowcharts, ) base_name = os.path.splitext(os.path.basename(file_path))[0] dir_name = os.path.dirname(file_path) or "." output_path = os.path.join(dir_name, f"{base_name}_flowchart.xlsx") self._embed_mermaid_and_delete_shapes( file_path, output_path, embeddings_by_sheet, consumed_by_sheet, ) return output_path, sheets_with_flowcharts except Exception as e: logger.warning(f"Flowchart detection failed: {e}") import traceback logger.debug(f"Flowchart detection error: {traceback.format_exc()}") return None def _embed_mermaid_and_delete_shapes( self, input_path: str, output_path: str, embeddings_by_sheet: Dict[str, List[tuple[str, str]]], consumed_by_sheet: Dict[str, List[str]], ) -> None: """Single ZIP pass: strip shapes, clear consumed cells, inject mermaid cells. Cell clearing runs BEFORE the embedding loop so a freshly cleared cell is reusable as an embed target without colliding with leftover content. Args: input_path: Path to input .xlsx file. output_path: Path to output .xlsx file. embeddings_by_sheet: Dict[sheet_name, List[(cell_ref, mermaid_text)]]. Mermaid text should already be wrapped in a ```mermaid``` fence. consumed_by_sheet: Dict[sheet_name, List[cell_coord]] of cells whose text was mined as edge-label content; these are cleared in the sheet XML (strip / + drop t attribute, keep + style). """ import xml.etree.ElementTree as ET import zipfile import re import shutil import tempfile NS_XDR = "http://schemas.openxmlformats.org/drawingml/2006/spreadsheetDrawing" NS_A = "http://schemas.openxmlformats.org/drawingml/2006/main" NS_MAIN = "http://schemas.openxmlformats.org/spreadsheetml/2006/main" NS_R = "http://schemas.openxmlformats.org/officeDocument/2006/relationships" NS_PR = "http://schemas.openxmlformats.org/package/2006/relationships" NS_LOOKUP = {"xdr": NS_XDR, "a": NS_A, "main": NS_MAIN, "r": NS_R} ET.register_namespace("", NS_MAIN) ET.register_namespace("xdr", NS_XDR) ET.register_namespace("a", NS_A) # Resolve sheet name -> sheet_xml_path via workbook.xml + rels sheet_xml_by_name: Dict[str, str] = {} with zipfile.ZipFile(input_path, "r") as zf: wb_xml = ET.fromstring(zf.read("xl/workbook.xml")) sheets_meta = [] for s in wb_xml.findall(f"{{{NS_MAIN}}}sheets/{{{NS_MAIN}}}sheet"): sheets_meta.append((s.get("name", ""), s.get(f"{{{NS_R}}}id", ""))) wb_rels_root = ET.fromstring(zf.read("xl/_rels/workbook.xml.rels")) wb_rels = { r.get("Id", ""): r.get("Target", "") for r in wb_rels_root.findall(f"{{{NS_PR}}}Relationship") } for name, rid in sheets_meta: target = wb_rels.get(rid, "") if not target: continue sheet_xml = ("xl/" + target.lstrip("/").removeprefix("/")).replace("xl/xl/", "xl/") sheet_xml_by_name[name] = sheet_xml embeddings_by_xml: Dict[str, List[tuple[str, str]]] = {} consumed_by_xml: Dict[str, set] = {} for name, pairs in embeddings_by_sheet.items(): xml_path = sheet_xml_by_name.get(name) if xml_path: embeddings_by_xml[xml_path] = pairs for name, refs in consumed_by_sheet.items(): xml_path = sheet_xml_by_name.get(name) if xml_path and refs: consumed_by_xml[xml_path] = set(refs) touched_sheets = set(embeddings_by_xml.keys()) | set(consumed_by_xml.keys()) def _col_letters_to_idx(letters: str) -> int: n = 0 for ch in letters: n = n * 26 + (ord(ch) - ord("A") + 1) return n def _idx_to_col_letters(idx: int) -> str: s = "" while idx > 0: idx, r = divmod(idx - 1, 26) s = chr(ord("A") + r) + s return s def _cell_has_value(cell: ET.Element) -> bool: return ( cell.find(f"{{{NS_MAIN}}}v") is not None or cell.find(f"{{{NS_MAIN}}}is") is not None ) deleted_shapes = 0 deleted_connectors = 0 cleared_cells = 0 inserted_cells = 0 temp_dir = tempfile.mkdtemp() temp_xlsx = os.path.join(temp_dir, "temp.xlsx") try: with zipfile.ZipFile(input_path, "r") as zin, zipfile.ZipFile( temp_xlsx, "w", zipfile.ZIP_DEFLATED ) as zout: for item in zin.infolist(): data = zin.read(item.filename) if item.filename.startswith("xl/drawings/") and item.filename.endswith(".xml"): try: root = ET.fromstring(data) for anchor in list(root): if anchor.find(".//xdr:sp", NS_LOOKUP) is not None: root.remove(anchor) deleted_shapes += 1 elif anchor.find(".//xdr:cxnSp", NS_LOOKUP) is not None: root.remove(anchor) deleted_connectors += 1 data = ET.tostring(root, encoding="utf-8", xml_declaration=True) except Exception as e: logger.warning(f"drawing parse failed for {item.filename}: {e}") if item.filename in touched_sheets: try: root = ET.fromstring(data) sheet_data = root.find(f"{{{NS_MAIN}}}sheetData") if sheet_data is None: logger.warning(f"sheet {item.filename} has no sheetData") else: consumed_refs = consumed_by_xml.get(item.filename, set()) if consumed_refs: for row_elem in sheet_data.findall(f"{{{NS_MAIN}}}row"): for cell in row_elem.findall(f"{{{NS_MAIN}}}c"): if cell.get("r") in consumed_refs: v = cell.find(f"{{{NS_MAIN}}}v") if v is not None: cell.remove(v) is_el = cell.find(f"{{{NS_MAIN}}}is") if is_el is not None: cell.remove(is_el) if cell.get("t") in {"inlineStr", "s", "str"}: cell.attrib.pop("t", None) cleared_cells += 1 for cell_ref, mermaid_text in embeddings_by_xml.get(item.filename, []): m = re.match(r"^([A-Z]+)([0-9]+)$", cell_ref) if not m: logger.warning(f"bad cell ref {cell_ref!r}, skipping") continue col_letter = m.group(1) row_num = int(m.group(2)) row_elem = None for r in sheet_data.findall(f"{{{NS_MAIN}}}row"): if int(r.get("r", "0")) == row_num: row_elem = r break if row_elem is None: row_elem = ET.SubElement(sheet_data, f"{{{NS_MAIN}}}row") row_elem.set("r", str(row_num)) existing_by_ref = { c.get("r"): c for c in row_elem.findall(f"{{{NS_MAIN}}}c") } target_ref = cell_ref existing = existing_by_ref.get(target_ref) if existing is not None and _cell_has_value(existing): start_col_idx = _col_letters_to_idx(col_letter) row_col_indices = sorted( _col_letters_to_idx(re.match(r"^([A-Z]+)", ref).group(1)) for ref in existing_by_ref.keys() ) max_col_idx = row_col_indices[-1] if row_col_indices else start_col_idx chosen_idx = None for idx in range(start_col_idx + 1, max_col_idx + 2): ref = f"{_idx_to_col_letters(idx)}{row_num}" cand = existing_by_ref.get(ref) if cand is None or not _cell_has_value(cand): chosen_idx = idx break if chosen_idx is None: chosen_idx = max_col_idx + 1 target_ref = f"{_idx_to_col_letters(chosen_idx)}{row_num}" logger.info( " %s already has content; relocating mermaid to %s", cell_ref, target_ref, ) existing = existing_by_ref.get(target_ref) if existing is not None: row_elem.remove(existing) cell = ET.SubElement(row_elem, f"{{{NS_MAIN}}}c") cell.set("r", target_ref) cell.set("t", "inlineStr") is_elem = ET.SubElement(cell, f"{{{NS_MAIN}}}is") t_elem = ET.SubElement(is_elem, f"{{{NS_MAIN}}}t") t_elem.set("{http://www.w3.org/XML/1998/namespace}space", "preserve") t_elem.text = mermaid_text inserted_cells += 1 data = ET.tostring(root, encoding="utf-8", xml_declaration=True) except Exception as e: logger.warning(f"sheet patch failed for {item.filename}: {e}") zout.writestr(item, data) shutil.move(temp_xlsx, output_path) finally: shutil.rmtree(temp_dir, ignore_errors=True) logger.info( "embed+strip: deleted %d shapes / %d connectors; cleared %d edge-label cell(s); inserted %d cell(s)", deleted_shapes, deleted_connectors, cleared_cells, inserted_cells, ) def _convert_images_to_chunked_dtos( self, image_extraction_result: Dict[str, Any], file_id: UUID, original_filename: str, ) -> 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 Returns: List of ChunkedElementDTO objects for images """ 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 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 = { "document_id": str(file_id), "source_filename": original_filename, "unique_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 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") # Base64 data for SeaweedFS/MinIO storage (will be filtered out when storing to Milvus) if "base64_data" in img_meta: metadata["image_base64"] = img_meta["base64_data"] if "mime_type" in img_meta: metadata["mime_type"] = img_meta["mime_type"] # 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 _convert_xls_to_xlsx(self, xls_path: str) -> str: """Convert .xls or .xlsm file to .xlsx format using LibreOffice. ks_xlsx_parser only supports .xlsx files, so we need to convert older formats. Uses LibreOffice which preserves images, shapes, and formatting. Args: xls_path: Path to the .xls or .xlsm 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 and Docker) possible_paths = [ 'libreoffice', 'soffice', '/Applications/LibreOffice.app/Contents/MacOS/soffice', # macOS '/usr/bin/libreoffice', # Linux/Docker '/usr/bin/soffice', # Linux/Docker ] 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 .xls/.xlsm conversion." ) logger.info(f"Converting 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=300 ) 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 to .xlsx: {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 @lru_cache() def get_excel_parser_service() -> ExcelParserService: """Get a cached Excel parser service instance. Returns: ExcelParserService instance """ from src.extraction_v2.excel_image_extraction_service import ( get_excel_image_extraction_service, ) from src.extraction_v2.shape_extractor import ShapeExtractor return ExcelParserService( image_extraction_service=get_excel_image_extraction_service(), shape_extractor=ShapeExtractor(), )