"""
Document Converter using Docling for Excel, Word, PDF, and PowerPoint documents.
This module provides document conversion with support for multiple formats:
- Excel (.xlsx, .xls, .xlsm) → HTML
- Word (.docx, .doc, .docm) → Markdown
- PDF (.pdf) → Markdown
- PowerPoint (.pptx, .ppt, .pptm) → Markdown
It includes LibreOffice CLI conversion for legacy formats (.xls, .xlsm, .doc, .docm, .ppt, .pptm)
and preprocessing to improve extraction accuracy.
Excel preprocessing includes:
- Shape/annotation extraction and embedding into cells
- Comment extraction and ID marker application
- Table border detection and separator row insertion
PowerPoint preprocessing includes:
- Speaker notes extraction and inline embedding
Word preprocessing (future):
- Image extraction with AI captions
- Shape and annotation extraction
- Comment extraction
"""
import asyncio
import fcntl
import os
import subprocess
import shutil
import tempfile
import time
from pathlib import Path
from typing import Optional, List, Dict, Any
import structlog
from openpyxl import load_workbook
from docling.datamodel.base_models import InputFormat
from docling.document_converter import DocumentConverter, ExcelFormatOption, WordFormatOption, PdfFormatOption
from src.extraction_v2.excel_loader import load_excel_file, convert_xls_to_xlsx
logger = structlog.get_logger(__name__)
class _ThreadSafeEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
"""
Custom event loop policy that automatically creates event loops for threads.
The default asyncio policy raises RuntimeError when get_event_loop() is called
from a thread without an event loop. This policy instead creates a new loop,
which is needed for libraries like Docling that use internal threading.
"""
def get_event_loop(self):
"""Get or create an event loop for the current thread."""
try:
return super().get_event_loop()
except RuntimeError:
# No event loop in this thread, create one
loop = self.new_event_loop()
self.set_event_loop(loop)
return loop
def _ensure_event_loop() -> None:
"""
Ensure an event loop exists and set a thread-safe event loop policy.
This is needed because Docling uses asyncio internally (including in spawned
threads), and in forked Celery worker processes, there may not be a valid
event loop. The Celery signal handler should set one up, but this provides
a fallback safety net.
We also install a custom event loop policy that auto-creates loops for any
thread that needs one, handling Docling's internal threading.
"""
# Install thread-safe policy if not already installed
current_policy = asyncio.get_event_loop_policy()
if not isinstance(current_policy, _ThreadSafeEventLoopPolicy):
asyncio.set_event_loop_policy(_ThreadSafeEventLoopPolicy())
logger.debug("Installed thread-safe event loop policy")
try:
asyncio.get_event_loop()
except RuntimeError:
# No event loop exists, create one
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
logger.debug("Created new event loop for current thread")
# Global lock file for LibreOffice to prevent concurrent access
LIBREOFFICE_LOCK_FILE = "/tmp/dsol_libreoffice.lock"
class DocumentToHtmlConverter:
"""Converts Excel, Word, PDF, and PowerPoint documents using Docling.
Supports:
- Excel (.xlsx, .xls, .xlsm) → HTML
- Word (.docx, .doc, .docm) → Markdown
- PDF (.pdf) → Markdown
- PowerPoint (.pptx, .ppt, .pptm) → Markdown
"""
def __init__(self):
"""Initialize the converter with Docling configuration for Excel, Word, PDF, and PowerPoint."""
# Configure PDF options with OCR disabled for performance
from docling.datamodel.pipeline_options import PdfPipelineOptions
pdf_pipeline_options = PdfPipelineOptions(do_ocr=False)
self.converter = DocumentConverter(
format_options={
InputFormat.XLSX: ExcelFormatOption(),
InputFormat.DOCX: WordFormatOption(),
InputFormat.PDF: PdfFormatOption(pipeline_options=pdf_pipeline_options),
# PowerPoint: No format option needed, Docling handles it automatically
}
)
self._last_doc = None
self._last_sheet_names = []
def convert_to_html(self, file_path: str) -> Optional[str]:
"""
Convert document to HTML (Excel) or Markdown (Word/PDF/PowerPoint).
Routes to appropriate conversion method based on file extension:
- Excel (.xlsx, .xls, .xlsm) → _convert_excel_to_html() → HTML
- Word (.docx, .doc, .docm) → _convert_word_to_markdown() → Markdown
- PDF (.pdf) → _convert_pdf_to_markdown() → Markdown
- PowerPoint (.pptx, .ppt, .pptm) → _convert_pptx_to_markdown() → Markdown
Args:
file_path: Path to Excel, Word, PDF, or PowerPoint file
Returns:
HTML string (for Excel) or Markdown string (for Word/PDF/PowerPoint), None on failure
"""
file_path_obj = Path(file_path)
file_ext = file_path_obj.suffix.lower()
# Route to appropriate converter
if file_ext in ['.xlsx', '.xls', '.xlsm']:
return self._convert_excel_to_html(file_path)
elif file_ext in ['.docx', '.doc', '.docm']:
return self._convert_word_to_markdown(file_path)
elif file_ext == '.pdf':
return self._convert_pdf_to_markdown(file_path)
elif file_ext in ['.pptx', '.ppt', '.pptm']:
return self._convert_pptx_to_markdown(file_path)
else:
logger.error(f"Unsupported file format: {file_ext}")
return None
def _convert_excel_to_html(self, file_path: str) -> Optional[str]:
"""
Convert Excel to HTML.
Processing Pipeline:
1. Convert .xls/.xlsm to .xlsx if needed
2. Preprocess: Detect table borders and insert separator rows
3. Convert preprocessed .xlsx to HTML (via Docling)
Handles .xlsx, .xls, and .xlsm files:
- .xlsx: Preprocess → Docling
- .xls: pandas+xlrd → Preprocess → Docling
- .xlsm: LibreOffice → Preprocess → Docling
Args:
file_path: Path to Excel file (.xlsx, .xls, or .xlsm)
Returns:
Tuple of (html_string, color_map_json_path) or (None, None) on failure
color_map_json_path is a temp file that must be cleaned up by caller
"""
temp_xlsx_path = None
temp_dir = None
preprocessed_path = None
color_map_path = None
try:
# Step 1: Convert .xls or .xlsm to .xlsx first
_, ext = os.path.splitext(file_path)
ext_lower = ext.lower()
if ext_lower == '.xls':
# Use LibreOffice for .xls files to preserve shapes/drawings/flowcharts
logger.info(f"Detected .xls file, converting to .xlsx via LibreOffice...")
temp_xlsx_path = self._convert_to_xlsx_via_libreoffice(file_path)
if temp_xlsx_path is None:
logger.error("LibreOffice conversion failed")
return None
file_to_preprocess = temp_xlsx_path
elif ext_lower == '.xlsm':
# Use LibreOffice for .xlsm files (macro-enabled)
logger.info(f"Detected .xlsm file, converting to .xlsx via LibreOffice...")
temp_xlsx_path = self._convert_to_xlsx_via_libreoffice(file_path)
if temp_xlsx_path is None:
logger.error("LibreOffice conversion failed")
return None
file_to_preprocess = temp_xlsx_path
else:
# Already .xlsx
file_to_preprocess = file_path
# Step 2: Preprocess - Detect borders, insert separator rows, extract colors
logger.info("Preprocessing: Detecting table borders and inserting separators...")
preprocessed_path, color_map_path = self._preprocess_with_border_detection(file_to_preprocess)
if preprocessed_path is None:
logger.warning("Border detection preprocessing failed, using original file")
file_to_convert = file_to_preprocess
else:
logger.info("Preprocessing complete, using separated file")
file_to_convert = preprocessed_path
# Step 3: Convert to HTML via Docling
logger.info(f"Converting Excel to HTML via Docling: {file_to_convert}")
_ensure_event_loop() # Ensure event loop exists for Docling
result = self.converter.convert(file_to_convert)
doc = result.document
# Get sheet names from original Excel file
sheet_names = self._get_sheet_names(file_to_convert)
logger.debug(f"Sheet names from Excel: {sheet_names}")
# Export to HTML format
html = doc.export_to_html()
# Store sheet names and document for later use
self._last_doc = doc
self._last_sheet_names = sheet_names
logger.info(f"Successfully converted Excel to HTML ({len(html)} chars)")
return html, color_map_path
except Exception as e:
logger.error(f"Docling conversion failed: {e}")
return None, None
finally:
# Clean up temporary files (color_map_path NOT cleaned here - caller's responsibility)
for temp_file in [temp_xlsx_path, preprocessed_path]:
if temp_file and temp_file != file_path and os.path.exists(temp_file):
try:
os.remove(temp_file)
logger.debug(f"Cleaned up temporary file: {temp_file}")
except Exception as e:
logger.warning(f"Failed to remove temp file {temp_file}: {e}")
# Clean up temporary directory (for .xls conversion)
if temp_dir and os.path.exists(temp_dir):
try:
shutil.rmtree(temp_dir)
logger.debug(f"Cleaned up temporary directory: {temp_dir}")
except Exception as e:
logger.warning(f"Failed to remove temp directory {temp_dir}: {e}")
def _preprocess_with_border_detection(self, file_path: str) -> tuple[Optional[str], Optional[str]]:
"""
Preprocess Excel file with comprehensive extraction and enhancement.
Processing steps:
0. Extract cell colors (non-modifying, creates separate JSON)
1. Extract images and apply descriptions to cells
2. Extract and apply shape annotations to cells
3. Extract and apply comment markers to cells
4. Detect table borders and insert separator rows
Flowcharts are handled separately in ExcelParserService (mermaid
embedding) before this preprocessing path runs.
This enriches the Excel file before Docling conversion, ensuring
images, annotations, comments, and table structures are preserved in HTML output.
Args:
file_path: Path to .xlsx file
Returns:
Tuple of (preprocessed_file_path, color_map_json_path)
Both can be None on failure
"""
color_map_path = None
try:
from src.extraction_v2.detect_border import BorderTableDetector
from src.extraction_v2.color_extractor import CellColorExtractor
import json
# Create temporary directory for intermediate files
temp_dir = tempfile.mkdtemp()
current_file = file_path
# Step 0: Extract cell colors (non-modifying, creates JSON sidecar)
logger.info("Step 0/5: Extracting cell colors...")
try:
color_extractor = CellColorExtractor()
color_result = color_extractor.extract_colors(file_path)
if color_result.status in ["success", "partial_failure"]:
# Create temp JSON file for color map
temp_fd, color_map_path = tempfile.mkstemp(suffix='_colors.json')
os.close(temp_fd)
with open(color_map_path, 'w') as f:
json.dump(color_result.color_map, f)
logger.info(
f"Color extraction {color_result.status}: "
f"{color_result.cells_with_colors} colored cells found "
f"(processed {color_result.cells_processed} cells)"
)
if color_result.warnings:
for warning in color_result.warnings[:5]: # Log first 5 warnings
logger.warning(f"Color extraction warning: {warning}")
elif color_result.status == "no_colors":
logger.info("No non-default cell colors found")
else:
logger.warning(f"Color extraction failed: {color_result.error_message}")
except Exception as e:
logger.warning(f"Color extraction failed: {e}")
# Continue without colors - not a blocking error
# Step 1: Apply image descriptions
logger.info("Step 1/5: Extracting images and applying descriptions...")
images_file = self._apply_image_descriptions(current_file, temp_dir)
if images_file:
current_file = images_file
logger.info("Image descriptions applied successfully")
else:
logger.info("No images found or image processing skipped")
# Step 2: Apply shape annotations
logger.info("Step 2/5: Extracting and applying shape annotations...")
shapes_file = self._apply_shape_annotations(current_file, temp_dir)
if shapes_file:
current_file = shapes_file
logger.info("Shape annotations applied successfully")
else:
logger.info("No shapes found or shape processing skipped")
# Step 3: Apply comment markers
logger.info("Step 3/5: Extracting and applying comment markers...")
comments_file = self._apply_comment_markers(current_file, temp_dir)
if comments_file:
current_file = comments_file
logger.info("Comment markers applied successfully")
else:
logger.info("No comments found or comment processing skipped")
# Step 5: Border detection and separator insertion
logger.info("Step 4/5: Detecting borders and inserting separators...")
detector = BorderTableDetector()
output_path = os.path.join(
temp_dir,
f"{os.path.splitext(os.path.basename(file_path))[0]}_preprocessed.xlsx"
)
stats = detector.insert_separator_rows(current_file, output_path)
logger.info(
f"Border detection complete: {stats['tables_processed']} tables processed, "
f"{stats['rows_inserted_above'] + stats['rows_inserted_below']} separator rows inserted"
)
logger.info("Preprocessing complete: All enhancements applied")
return output_path, color_map_path
except Exception as e:
logger.warning(f"Preprocessing failed: {e}")
# Clean up color map if it was created
if color_map_path and os.path.exists(color_map_path):
try:
os.remove(color_map_path)
except Exception:
pass
return None, None
def _apply_image_descriptions(self, file_path: str, temp_dir: str) -> Optional[str]:
"""
Extract images and apply AI-generated descriptions to cells.
Uses the image extraction service to get images with descriptions,
then embeds those descriptions into nearby cells.
Args:
file_path: Path to Excel file
temp_dir: Temporary directory for output file
Returns:
Path to file with image descriptions applied, or None if no images found
"""
try:
# Check if file has images by inspecting the archive
if not self._has_images(file_path):
logger.debug("No images found in file")
return None
# Import image extraction service
import asyncio
from uuid import uuid4
from src.extraction_v2.image_extractor import ExcelImageExtractionService
from src.services.azure_chat_service import AzureChatService
logger.info("Extracting images with AI descriptions...")
# Read file content as bytes for the service
with open(file_path, 'rb') as f:
file_content = f.read()
# Create Azure chat service for AI descriptions
azure_chat_service = AzureChatService()
# Extract images using the service
service = ExcelImageExtractionService(azure_chat_service=azure_chat_service)
file_id = str(uuid4())
filename = os.path.basename(file_path)
# Run async extraction in sync context
image_result = asyncio.run(
service.extract_images_from_file(
file_content=file_content,
file_id=file_id,
original_filename=filename,
generate_descriptions=True,
description_detail="auto"
)
)
if not image_result or "images" not in image_result or not image_result["images"]:
logger.info("No images found in Excel file")
return None
images = image_result["images"]
logger.info(f"Found {len(images)} images, applying descriptions to cells...")
# Create position-to-description mapping
position_descriptions = self._create_position_descriptions_map(images)
if not position_descriptions:
logger.warning("Images found but no position information available")
return None
# Apply image descriptions to cells
output_path = os.path.join(temp_dir, f"{os.path.splitext(filename)[0]}_with_images.xlsx")
self._replace_images_with_descriptions(file_path, position_descriptions, output_path)
return output_path
except Exception as e:
logger.warning(f"Image processing failed: {e}")
import traceback
logger.debug(f"Image processing error details: {traceback.format_exc()}")
return None
def _has_images(self, file_path: str) -> bool:
"""Check if Excel file contains any images."""
try:
import zipfile
with zipfile.ZipFile(file_path, 'r') as zip_ref:
# Check for media files in the archive
media_files = [name for name in zip_ref.namelist() if 'xl/media/' in name]
return len(media_files) > 0
except Exception:
return False
def _create_position_descriptions_map(self, images: List[Dict[str, Any]]) -> Dict[str, str]:
"""
Create a mapping of position keys to image descriptions.
Args:
images: List of image metadata dictionaries
Returns:
Dictionary mapping position keys to descriptions
"""
position_descriptions = {}
for img_meta in 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
logger.debug(f"Mapped image position {pos_key} to description: {description[:50]}...")
return position_descriptions
def _replace_images_with_descriptions(self, file_path: str, position_descriptions: Dict[str, str], output_path: 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
output_path: Path to save modified file
"""
if not position_descriptions:
logger.info("No image descriptions to apply")
return
logger.info(f"Applying {len(position_descriptions)} image descriptions to cells")
try:
workbook = load_excel_file(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")
unknown_images.append((pos_key, description))
continue
if sheet_name not in workbook.sheetnames:
logger.warning(f"Sheet '{sheet_name}' not found in workbook")
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.debug(f"Applied image 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
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")
# Save the modified workbook
workbook.save(output_path)
workbook.close()
logger.info(f"Successfully applied {replacements_made} image descriptions to {output_path}")
except Exception as e:
logger.error(f"Error applying image descriptions: {e}")
def _parse_position_key(self, pos_key: str) -> tuple:
"""
Parse position key to extract sheet name and cell reference.
Args:
pos_key: Position key in format "SheetName_CellRef" (e.g., "Sheet1_A5")
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(self, file_path: str, temp_dir: str) -> Optional[str]:
"""
Apply shape annotations to cells in the Excel file.
Extracts shapes with text content and embeds them into nearby cells
as annotations, making them visible in the HTML output.
Args:
file_path: Path to Excel file
temp_dir: Temporary directory for output file
Returns:
Path to file with shape annotations applied, or None if no shapes found
"""
try:
# Check if file has shapes by inspecting the archive
if not self._has_shapes(file_path):
logger.debug("No shapes found in file")
return None
# Use the shape extractor from extraction_v2
from src.extraction_v2.shape_extractor import ShapeExtractor
logger.info("Extracting shapes with text content...")
# Extract shapes and apply annotations
extractor = ShapeExtractor()
annotated_path = extractor.extract_and_apply_annotations(file_path, temp_dir)
if annotated_path:
logger.info(f"Successfully applied shape annotations to {annotated_path}")
else:
logger.info("No shapes with text content found")
return annotated_path
except Exception as e:
logger.warning(f"Shape annotation processing failed: {e}")
import traceback
logger.debug(f"Shape annotation error details: {traceback.format_exc()}")
return None
def _apply_comment_markers(self, file_path: str, temp_dir: str) -> Optional[str]:
"""
Extract comments and apply ID markers to cells.
This makes comments visible in the HTML output by embedding comment
markers in cell values.
Args:
file_path: Path to Excel file
temp_dir: Temporary directory for output file
Returns:
Path to file with comment markers applied, or None if no comments found
"""
try:
workbook = load_excel_file(file_path, data_only=False)
# Extract all comments
comments_by_sheet = self._extract_comments_from_file(workbook)
if not comments_by_sheet:
logger.debug("No comments found in Excel file")
return None
total_comments = sum(len(comments) for comments in comments_by_sheet.values())
logger.info(f"Found {total_comments} comments across {len(comments_by_sheet)} sheets")
# Apply comment markers to cells
output_path = os.path.join(temp_dir, f"{os.path.splitext(os.path.basename(file_path))[0]}_with_comments.xlsx")
self._apply_comments_to_cells(workbook, comments_by_sheet, output_path)
workbook.close()
logger.info(f"Applied {total_comments} comment markers to cells")
return output_path
except Exception as e:
logger.warning(f"Comment processing failed: {e}")
return None
def _has_shapes(self, file_path: str) -> bool:
"""Check if Excel file contains any shapes with text."""
try:
import zipfile
with zipfile.ZipFile(file_path, 'r') as zip_ref:
# Check for drawing files in the archive
drawing_files = [name for name in zip_ref.namelist() if 'xl/drawings/' in name]
return len(drawing_files) > 0
except Exception:
return False
def _extract_comments_from_file(self, workbook) -> Dict[str, List[Dict[str, Any]]]:
"""
Extract all comments from an Excel workbook with full metadata.
Args:
workbook: Openpyxl workbook object
Returns:
Dictionary mapping {sheet_name: [list of comment dictionaries]}
"""
comments_by_sheet = {}
for sheet in workbook.worksheets:
sheet_comments = []
# Use openpyxl's built-in comment detection (much faster)
# The sheet._comments attribute contains all comments
if not hasattr(sheet, '_comments') or not sheet._comments:
continue
for cell_ref, comment in sheet._comments.items():
try:
# Get the cell
cell = sheet[cell_ref]
# Extract comment text
comment_text = comment.text
if hasattr(comment_text, "text"):
comment_text = str(comment_text)
comment_text = comment_text.strip()
if comment_text:
# Get author
author = None
if hasattr(comment, "author"):
author = 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)
logger.debug(f"Found comment at {sheet.title}!{cell.coordinate}: {comment_text[:50]}...")
except Exception as e:
logger.warning(f"Failed to extract comment from {sheet.title}!{cell_ref}: {e}")
if sheet_comments:
comments_by_sheet[sheet.title] = sheet_comments
return comments_by_sheet
def _apply_comments_to_cells(self, workbook, comments_by_sheet: Dict[str, List[Dict[str, Any]]], output_path: str):
"""
Apply comment markers to cells in the workbook.
Embeds comment content directly into cell values with a marker format
that will be preserved in HTML conversion.
Args:
workbook: Openpyxl workbook object
comments_by_sheet: Dictionary of comments by sheet name
output_path: Path to save the modified workbook
"""
import uuid
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
comment_id = f"comment_{uuid.uuid4().hex[:8]}"
# Get current cell value
current_value = str(cell.value) if cell.value is not None else ""
# Create HTML-like comment tag with the comment text
# Format: [COMMENT: text] original_value
comment_marker = f"[COMMENT: {comment_data['comment_text']}]"
if current_value and current_value != "None":
cell.value = f"{comment_marker} {current_value}"
else:
cell.value = comment_marker
logger.debug(f"Applied comment marker to {sheet_name}!{cell_ref}")
except Exception as e:
logger.warning(f"Failed to apply comment to {sheet_name}!{cell_ref}: {e}")
# Save the modified workbook
workbook.save(output_path)
logger.info(f"Saved workbook with comment markers to {output_path}")
def _apply_shape_annotations_to_cells(self, file_path: str, shape_dtos: List, output_path: str) -> 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: Merge range and add shape text
- 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
output_path: Path to save annotated file
Returns:
Path to modified Excel file with annotations
"""
try:
logger.info(f"Applying {len(shape_dtos)} shape annotations to cells...")
# Load with data_only=True to get calculated values instead of formulas
workbook = load_excel_file(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]
current_value = str(cell.value) if cell.value is not None else ""
# Create HTML-like annotation tag
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)
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:
current_value = str(cell.value) if cell.value is not None else ""
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:
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:
try:
intersection = range_to_merge.intersection(merged_range)
if intersection is not None:
is_conflicting = True
logger.debug(
f"Range {from_cell}:{to_cell} conflicts with existing merged range, skipping merge"
)
break
except ValueError:
continue
if not is_conflicting:
# Merge cells
worksheet.merge_cells(f"{from_cell}:{to_cell}")
# Set the shape text in the top-left cell
top_left_cell = worksheet[from_cell]
top_left_cell.value = shape_text
# 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]
top_left_cell.value = shape_text
annotations_applied += 1
logger.debug(f"Added shape text to {from_cell} 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
# Save modified workbook
workbook.save(output_path)
workbook.close()
logger.info(f"Applied {annotations_applied} shape annotations to {output_path}")
return output_path
except Exception as e:
logger.error(f"Error applying shape annotations: {e}")
return file_path
def _convert_word_to_markdown(self, file_path: str) -> Optional[str]:
"""
Convert Word document to Markdown.
Processing Pipeline:
1. Validate file
2. Convert .doc/.docm to .docx if needed (via LibreOffice)
3. Preprocess .docx (extract images/shapes/comments, embed inline)
4. Convert preprocessed .docx to Markdown (via Docling)
Args:
file_path: Path to Word file (.docx, .doc, or .docm)
Returns:
Markdown string or None on failure
"""
file_path_obj = Path(file_path)
temp_docx_path = None
preprocessed_path = None
try:
# Step 1: Validate file
is_valid, error_msg = self._validate_word_file(file_path_obj)
if not is_valid:
logger.error(f"Word file validation failed: {error_msg}")
return None
# Step 2: Format normalization (.doc/.docm → .docx)
if file_path_obj.suffix.lower() in ['.doc', '.docm']:
logger.info(f"Converting {file_path_obj.suffix} to .docx via LibreOffice...")
temp_docx_path = self._convert_doc_to_docx(file_path_obj)
if temp_docx_path is None:
logger.error("LibreOffice conversion failed")
return None
file_to_preprocess = temp_docx_path
else:
# Already .docx
file_to_preprocess = str(file_path)
# Step 3: Convert to Markdown via Docling
# Try without preprocessing first (faster and avoids LibreOffice conflicts)
# Preprocessing can cause issues when Docling tries to convert embedded
# drawings to images via LibreOffice
logger.info(f"Converting Word to Markdown via Docling: {file_to_preprocess}")
# Validate the converted file before passing to docling
file_to_convert_path = Path(file_to_preprocess)
if not file_to_convert_path.exists():
raise FileNotFoundError(f"Converted file not found: {file_to_preprocess}")
file_size = file_to_convert_path.stat().st_size
logger.info(f"Converting file size: {file_size:,} bytes")
if file_size == 0:
raise ValueError(f"Converted file is empty: {file_to_preprocess}")
try:
_ensure_event_loop() # Ensure event loop exists for Docling
result = self.converter.convert(str(file_to_preprocess))
if not result or not result.document:
raise RuntimeError("Docling conversion returned empty result")
doc = result.document
markdown = doc.export_to_markdown()
logger.info(f"Successfully converted Word to Markdown ({len(markdown)} chars)")
return markdown
except Exception as docling_error:
# Log detailed error information
import traceback
error_details = traceback.format_exc()
logger.error(
f"Docling conversion failed for {file_path_obj.name}: {docling_error}",
extra={
"original_file": str(file_path_obj),
"converted_file": str(file_to_preprocess),
"file_size": file_size,
"error_type": type(docling_error).__name__,
"error_details": error_details
}
)
logger.warning("Trying with preprocessing...")
# Step 4: Retry with preprocessing if direct conversion failed
try:
from src.extraction_v2.word_preprocessor import WordPreprocessor
preprocessor = WordPreprocessor()
prep_path = preprocessor.preprocess(Path(file_to_preprocess))
if prep_path:
# Update for cleanup
preprocessed_path = str(prep_path)
logger.info("Preprocessing successful, retrying conversion")
_ensure_event_loop() # Ensure event loop exists for Docling
result = self.converter.convert(preprocessed_path)
if result and result.document:
doc = result.document
markdown = doc.export_to_markdown()
logger.info(
f"Successfully converted Word to Markdown (with preprocessing) "
f"({len(markdown)} chars)"
)
return markdown
except Exception as preprocess_error:
logger.warning(f"Preprocessing retry also failed: {preprocess_error}")
# Re-raise original error if all attempts failed
raise docling_error
except Exception as e:
logger.error(f"Docling Word conversion failed: {e}")
import traceback
logger.debug(f"Conversion error details: {traceback.format_exc()}")
return None
finally:
# Clean up temporary files
for temp_file in [temp_docx_path, preprocessed_path]:
if temp_file and os.path.exists(temp_file):
try:
# Also clean up the temp directory
temp_dir = os.path.dirname(temp_file)
if temp_dir and os.path.exists(temp_dir) and 'tmp' in temp_dir:
shutil.rmtree(temp_dir)
logger.debug(f"Cleaned up temporary directory: {temp_dir}")
except Exception as e:
logger.warning(f"Failed to remove temp file/directory {temp_file}: {e}")
def _convert_doc_to_docx(self, file_path: Path) -> Optional[str]:
"""
Convert .doc or .docm to .docx.
Handles:
- .doc (Word 97-2003) → .docx via LibreOffice
- .docm (macro-enabled) → .docx via direct XML extraction
Args:
file_path: Path to .doc or .docm file
Returns:
Path to converted .docx file, or None on failure
"""
# For .docm files, use direct XML extraction (LibreOffice often loses content)
if file_path.suffix.lower() == '.docm':
return self._convert_docm_to_docx_direct(file_path)
# For .doc files, use LibreOffice
# Find LibreOffice executable
libreoffice_cmd = self._find_libreoffice_command()
if not libreoffice_cmd:
logger.error(
"LibreOffice not found. Required for .doc/.docm conversion. "
"Install: apt-get install libreoffice (Ubuntu) or brew install libreoffice (macOS)"
)
return None
try:
# Ensure file_path is absolute (LibreOffice needs absolute path)
abs_file_path = file_path.resolve()
if not abs_file_path.exists():
logger.error(f"File not found: {abs_file_path}")
return None
# Create temp directory for output
temp_dir = tempfile.mkdtemp(prefix="docx_convert_")
logger.info(
"Converting to .docx",
input=abs_file_path.name,
format=abs_file_path.suffix,
path=str(abs_file_path)
)
# Use file lock to prevent concurrent LibreOffice access
# LibreOffice can only run one instance at a time
lock_fd = None
# Create a unique user profile directory for this conversion
# This prevents LibreOffice profile lock conflicts when multiple
# processes try to convert files concurrently
profile_dir = tempfile.mkdtemp(prefix="lo_profile_")
try:
lock_fd = open(LIBREOFFICE_LOCK_FILE, 'w')
# Acquire exclusive lock (blocks until available)
logger.debug("Waiting for LibreOffice lock...")
fcntl.flock(lock_fd, fcntl.LOCK_EX) # Blocking lock
logger.debug("LibreOffice lock acquired")
# Run LibreOffice conversion with unique profile
# -env:UserInstallation prevents profile lock conflicts
result = subprocess.run(
[
libreoffice_cmd,
"--headless",
f"-env:UserInstallation=file://{profile_dir}",
"--convert-to", "docx",
"--outdir", temp_dir,
str(abs_file_path)
],
check=False, # Don't raise on non-zero, check output file instead
capture_output=True,
timeout=60,
text=True
)
# Log LibreOffice output for debugging
if result.stdout:
logger.debug(f"LibreOffice stdout: {result.stdout}")
if result.stderr:
logger.warning(f"LibreOffice stderr: {result.stderr}")
if result.returncode != 0:
logger.warning(f"LibreOffice returned non-zero exit code: {result.returncode}")
finally:
if lock_fd:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
# Clean up the temporary profile directory
try:
shutil.rmtree(profile_dir)
except Exception:
pass # Ignore cleanup errors
# Find the converted file (use the resolved stem for consistency)
converted_file = Path(temp_dir) / f"{abs_file_path.stem}.docx"
if not converted_file.exists():
logger.error(
"LibreOffice conversion failed: output file not created",
input_file=abs_file_path.name,
expected_output=str(converted_file),
stdout=result.stdout,
stderr=result.stderr,
returncode=result.returncode
)
return None
# Validate converted file
if converted_file.stat().st_size == 0:
logger.error(f"Conversion produced empty file: {abs_file_path.name}")
return None
logger.info(
"Conversion successful",
input=abs_file_path.name,
output=converted_file.name,
size_mb=converted_file.stat().st_size / (1024 * 1024)
)
return str(converted_file)
except subprocess.TimeoutExpired:
logger.error("Conversion timeout", file=abs_file_path.name if 'abs_file_path' in locals() else file_path.name, timeout=60)
return None
except Exception as e:
logger.error("Unexpected conversion error", file=file_path.name, error=str(e))
return None
def _convert_docm_to_docx_direct(self, file_path: Path) -> Optional[str]:
"""
Convert .docm to .docx by direct XML manipulation.
LibreOffice often loses content when converting .docm files.
This method directly copies and transforms the archive structure.
Process:
1. Extract .docm archive (which is a ZIP file)
2. Update [Content_Types].xml to change docm to docx type
3. Remove VBA macros (vbaProject.bin)
4. Repackage as .docx
Args:
file_path: Path to .docm file
Returns:
Path to converted .docx file, or None on failure
"""
import zipfile
from lxml import etree
try:
# Ensure file_path is absolute
abs_file_path = file_path.resolve()
if not abs_file_path.exists():
logger.error(f"File not found: {abs_file_path}")
return None
# Create temp directory for output
temp_dir = tempfile.mkdtemp(prefix="docm_convert_")
logger.info(
"Converting .docm to .docx via direct XML extraction",
input=abs_file_path.name,
path=str(abs_file_path)
)
# Output file path
output_file = Path(temp_dir) / f"{abs_file_path.stem}.docx"
# Copy the .docm as .docx with modifications
with zipfile.ZipFile(abs_file_path, 'r') as src_zip:
with zipfile.ZipFile(output_file, 'w', zipfile.ZIP_DEFLATED) as dest_zip:
for item in src_zip.infolist():
# Skip VBA macros
if 'vbaProject' in item.filename:
logger.debug(f"Skipping VBA file: {item.filename}")
continue
# Skip [trash] folder
if '[trash]' in item.filename:
logger.debug(f"Skipping trash: {item.filename}")
continue
# Read the file content
data = src_zip.read(item.filename)
# Modify [Content_Types].xml
if item.filename == '[Content_Types].xml':
# Parse and modify content types
root = etree.fromstring(data)
ns = {'ct': 'http://schemas.openxmlformats.org/package/2006/content-types'}
# Change macro-enabled content type to standard docx
for override in root.findall('.//ct:Override', ns):
content_type = override.get('ContentType', '')
if 'macroEnabled' in content_type:
new_type = content_type.replace(
'vnd.ms-word.document.macroEnabled.main+xml',
'vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml'
)
override.set('ContentType', new_type)
logger.debug(f"Changed content type: {content_type} -> {new_type}")
# Remove Override entries for vbaProject
for override in root.findall('.//ct:Override', ns):
part_name = override.get('PartName', '')
if 'vbaProject' in part_name:
root.remove(override)
logger.debug(f"Removed vbaProject Override: {part_name}")
# Remove Default entries for vbaProject
for default in root.findall('.//ct:Default', ns):
if default.get('Extension') == 'bin':
content_type = default.get('ContentType', '')
if 'vbaProject' in content_type:
root.remove(default)
logger.debug("Removed vbaProject content type")
data = etree.tostring(root, xml_declaration=True, encoding='UTF-8')
# Remove vbaProject relationships from .rels files
if item.filename.endswith('.rels'):
root = etree.fromstring(data)
ns = {'r': 'http://schemas.openxmlformats.org/package/2006/relationships'}
# Find and remove vbaProject relationships
for rel in root.findall('.//r:Relationship', ns):
target = rel.get('Target', '')
rel_type = rel.get('Type', '')
if 'vbaProject' in target or 'vbaProject' in rel_type:
root.remove(rel)
logger.debug(f"Removed vbaProject relationship: {target}")
data = etree.tostring(root, xml_declaration=True, encoding='UTF-8')
# Write to output
dest_zip.writestr(item, data)
# Validate converted file
if not output_file.exists():
logger.error("Conversion failed: output file not created")
return None
if output_file.stat().st_size == 0:
logger.error("Conversion produced empty file")
return None
logger.info(
"Direct .docm conversion successful",
input=abs_file_path.name,
output=output_file.name,
size_mb=output_file.stat().st_size / (1024 * 1024)
)
return str(output_file)
except Exception as e:
logger.error(
"Direct .docm conversion failed",
file=file_path.name,
error=str(e)
)
import traceback
logger.debug(f"Conversion error details: {traceback.format_exc()}")
return None
def _validate_word_file(self, file_path: Path) -> tuple[bool, Optional[str]]:
"""
Validate Word file before conversion.
Args:
file_path: Path to Word file
Returns:
Tuple of (is_valid, error_message)
"""
# Check exists
if not file_path.exists():
return False, "File not found"
# Check size (<100MB)
size_mb = file_path.stat().st_size / (1024 * 1024)
if size_mb > 100:
return False, f"File too large: {size_mb:.1f}MB (max 100MB)"
# Check extension
ext = file_path.suffix.lower()
if ext not in ['.doc', '.docx', '.docm']:
return False, f"Unsupported format: {ext}"
# For .docx/.docm: verify ZIP structure (they are ZIP archives)
if ext in ['.docx', '.docm']:
try:
import zipfile
with zipfile.ZipFile(file_path, 'r') as zf:
# Check for required Word document structure
if 'word/document.xml' not in zf.namelist():
return False, "Invalid Word document structure (missing word/document.xml)"
except zipfile.BadZipFile:
return False, "Corrupted Word file (not a valid ZIP archive)"
except Exception as e:
return False, f"File validation error: {str(e)}"
return True, None
def _convert_pdf_to_markdown(self, file_path: str) -> Optional[str]:
"""
Convert PDF to Markdown using Docling.
Pipeline:
1. Validate file
2. Convert PDF to Markdown (via Docling)
Images and tables are handled by custom serializers during chunking phase.
Args:
file_path: Path to PDF file
Returns:
Markdown string or None on failure
"""
file_path_obj = Path(file_path)
try:
# Step 1: Validate file
is_valid, error_msg = self._validate_pdf_file(file_path_obj)
if not is_valid:
logger.error(f"PDF validation failed: {error_msg}")
return None
# Step 2: Docling conversion (no preprocessing needed)
logger.info(f"Converting PDF to Markdown via Docling: {file_path_obj}")
_ensure_event_loop()
result = self.converter.convert(str(file_path_obj))
if not result or not result.document:
raise RuntimeError("Docling conversion returned empty result")
markdown = result.document.export_to_markdown()
logger.info(f"Successfully converted PDF to Markdown ({len(markdown)} chars)")
return markdown
except Exception as e:
logger.error(f"Docling PDF conversion failed: {e}")
import traceback
logger.debug(f"Conversion error details: {traceback.format_exc()}")
return None
def _validate_pdf_file(self, file_path: Path) -> tuple[bool, Optional[str]]:
"""
Validate PDF file before conversion.
Args:
file_path: Path to PDF file
Returns:
Tuple of (is_valid, error_message)
"""
# Check exists
if not file_path.exists():
return False, "File not found"
# Check size (<100MB)
size_mb = file_path.stat().st_size / (1024 * 1024)
if size_mb > 100:
return False, f"File too large: {size_mb:.1f}MB (max 100MB)"
# Check extension
ext = file_path.suffix.lower()
if ext != '.pdf':
return False, f"Not a PDF file: {ext}"
# Verify PDF structure (basic check)
try:
with open(file_path, 'rb') as f:
header = f.read(5)
if not header.startswith(b'%PDF-'):
return False, "Invalid PDF file (missing PDF header)"
except Exception as e:
return False, f"File validation error: {str(e)}"
return True, None
def _convert_pptx_to_markdown(self, file_path: str) -> Optional[str]:
"""
Convert PowerPoint to Markdown using Docling.
Pipeline:
1. Convert .ppt/.pptm to .pptx if needed (via LibreOffice)
2. Preprocess: Extract and embed speaker notes inline
3. Convert to Markdown (via Docling)
Images and tables are handled by custom serializers during chunking phase.
Args:
file_path: Path to PowerPoint file (.pptx, .ppt, or .pptm)
Returns:
Markdown string or None on failure
"""
file_path_obj = Path(file_path)
ext = file_path_obj.suffix.lower()
temp_pptx_path = None
preprocessed_path = None
try:
# Step 1: Convert .ppt or .pptm to .pptx if needed
if ext in ['.ppt', '.pptm']:
logger.info(f"Converting {ext} to .pptx via LibreOffice...")
temp_pptx_path = self._convert_ppt_to_pptx(file_path_obj)
if temp_pptx_path is None:
logger.error(f"LibreOffice conversion failed for {file_path_obj}")
return None
file_to_preprocess = Path(temp_pptx_path)
else:
# Already .pptx
file_to_preprocess = file_path_obj
# Step 2: Check if there are speaker notes to preprocess
from pptx import Presentation
prs = Presentation(str(file_to_preprocess))
has_notes = any(slide.has_notes_slide and slide.notes_slide.notes_text_frame.text.strip()
for slide in prs.slides)
if has_notes:
# Preprocess - Extract and embed speaker notes
logger.info(f"Preprocessing PowerPoint (speaker notes): {file_to_preprocess}")
from src.extraction_v2.pptx_preprocessor import PptxPreprocessor
preprocessor = PptxPreprocessor()
preprocessed_path = preprocessor.preprocess(file_to_preprocess)
file_for_docling = preprocessed_path
else:
# No speaker notes, skip preprocessing
logger.info("No speaker notes found, skipping preprocessing")
preprocessed_path = None
file_for_docling = file_to_preprocess
# Step 3: Docling conversion
logger.info(f"Converting PowerPoint to Markdown via Docling: {file_for_docling}")
_ensure_event_loop()
result = self.converter.convert(str(file_for_docling))
if not result or not result.document:
raise RuntimeError("Docling conversion returned empty result")
markdown = result.document.export_to_markdown()
logger.info(f"Successfully converted PowerPoint to Markdown ({len(markdown)} chars)")
return markdown
except Exception as e:
logger.error(f"PowerPoint conversion failed: {e}")
import traceback
logger.debug(f"Conversion error details: {traceback.format_exc()}")
return None
finally:
# Clean up temporary files
if temp_pptx_path and Path(temp_pptx_path).exists():
try:
# Remove temp directory
temp_dir = Path(temp_pptx_path).parent
shutil.rmtree(temp_dir, ignore_errors=True)
except Exception as e:
logger.warning(f"Failed to clean up temp pptx file: {e}")
if preprocessed_path and preprocessed_path.exists():
try:
preprocessed_path.unlink()
except Exception as e:
logger.warning(f"Failed to clean up preprocessed file: {e}")
def _convert_ppt_to_pptx(self, file_path: Path) -> Optional[str]:
"""
Convert .ppt or .pptm to .pptx using LibreOffice CLI.
Preserves:
- Slides and content
- Speaker notes
- Images (handled by Docling + serializers)
Strips:
- Macros (.pptm)
Args:
file_path: Path to .ppt or .pptm file
Returns:
Path to converted .pptx file, or None on failure
"""
libreoffice_cmd = self._find_libreoffice_command()
if not libreoffice_cmd:
logger.error(
"LibreOffice not found. Required for .ppt/.pptm conversion. "
"Install: apt-get install libreoffice (Ubuntu) or brew install libreoffice (macOS)"
)
return None
try:
temp_dir = tempfile.mkdtemp(prefix="pptx_convert_")
profile_dir = tempfile.mkdtemp(prefix="lo_profile_")
# File locking to prevent concurrent LibreOffice access
lock_fd = open(LIBREOFFICE_LOCK_FILE, 'w')
fcntl.flock(lock_fd, fcntl.LOCK_EX) # Blocking lock
logger.debug("LibreOffice lock acquired")
try:
result = subprocess.run(
[
libreoffice_cmd,
'--headless',
f'-env:UserInstallation=file://{profile_dir}',
'--convert-to', 'pptx',
'--outdir', temp_dir,
str(file_path)
],
capture_output=True,
text=True,
timeout=60 # 60 second timeout
)
finally:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
# Clean up profile directory
shutil.rmtree(profile_dir, ignore_errors=True)
if result.returncode != 0:
logger.error(f"LibreOffice conversion failed: {result.stderr}")
return None
output_file = Path(temp_dir) / f"{file_path.stem}.pptx"
if not output_file.exists():
logger.error(f"Conversion succeeded but output file not found: {output_file}")
return None
logger.info(f"Successfully converted to .pptx: {output_file}")
return str(output_file)
except subprocess.TimeoutExpired:
logger.error("LibreOffice conversion timed out (>60s)")
return None
except Exception as e:
logger.error(f"Error during LibreOffice conversion: {e}")
return None
def _convert_to_xlsx_via_libreoffice(self, input_file: str) -> Optional[str]:
"""
Convert .xls or .xlsm to .xlsx using LibreOffice CLI.
This preserves images, shapes, charts, and formatting.
Args:
input_file: Path to .xls or .xlsm file
Returns:
Path to converted .xlsx file, or None on failure
"""
# Find LibreOffice executable
libreoffice_cmd = self._find_libreoffice_command()
if not libreoffice_cmd:
logger.error(
"LibreOffice not found. Please install LibreOffice for .xls/.xlsm support. "
"Install: https://www.libreoffice.org/download/"
)
return None
try:
# Create temp directory for output
temp_dir = tempfile.mkdtemp()
# Create a unique user profile directory for this conversion
# This prevents LibreOffice profile lock conflicts when multiple
# processes try to convert files concurrently
profile_dir = tempfile.mkdtemp(prefix="lo_profile_")
# Use file lock to prevent concurrent LibreOffice access
lock_fd = None
try:
lock_fd = open(LIBREOFFICE_LOCK_FILE, 'w')
# Acquire exclusive lock (blocks until available)
logger.debug("Waiting for LibreOffice lock...")
fcntl.flock(lock_fd, fcntl.LOCK_EX) # Blocking lock
logger.debug("LibreOffice lock acquired")
# Run LibreOffice conversion with unique profile
# -env:UserInstallation prevents profile lock conflicts
logger.info(f"Running LibreOffice conversion: {libreoffice_cmd}")
result = subprocess.run(
[
libreoffice_cmd,
'--headless', # No GUI
f'-env:UserInstallation=file://{profile_dir}',
'--convert-to', 'xlsx', # Target format
'--outdir', temp_dir, # Output directory
input_file # Input file
],
capture_output=True,
text=True,
timeout=30 # 30 second timeout
)
finally:
if lock_fd:
fcntl.flock(lock_fd, fcntl.LOCK_UN)
lock_fd.close()
# Clean up the temporary profile directory
try:
shutil.rmtree(profile_dir)
except Exception:
pass # Ignore cleanup errors
if result.returncode != 0:
logger.error(f"LibreOffice conversion failed: {result.stderr}")
return None
# Find the output file
input_basename = os.path.splitext(os.path.basename(input_file))[0]
output_file = os.path.join(temp_dir, f"{input_basename}.xlsx")
if not os.path.exists(output_file):
logger.error(f"Conversion succeeded but output file not found: {output_file}")
return None
logger.info(f"Successfully converted to .xlsx: {output_file}")
return output_file
except subprocess.TimeoutExpired:
logger.error("LibreOffice conversion timed out (>30s)")
return None
except Exception as e:
logger.error(f"Error during LibreOffice conversion: {e}")
return None
def _find_libreoffice_command(self) -> Optional[str]:
"""
Find LibreOffice executable on the system.
Returns:
Path to LibreOffice command, or None if not found
"""
# Common locations for LibreOffice
possible_commands = [
'libreoffice',
'soffice',
'/Applications/LibreOffice.app/Contents/MacOS/soffice', # macOS
'/usr/bin/libreoffice', # Linux
'/usr/bin/soffice', # Linux alternative
'C:\\Program Files\\LibreOffice\\program\\soffice.exe', # Windows
]
for cmd in possible_commands:
# Check if it's a full path that exists
if os.path.exists(cmd):
return cmd
# Check if it's in PATH
if shutil.which(cmd):
return cmd
return None
def _get_sheet_names(self, file_path: str) -> list[str]:
"""Get sheet names from Excel file.
Args:
file_path: Path to Excel file
Returns:
List of sheet names in order
"""
try:
wb = load_excel_file(file_path, read_only=True, data_only=True)
sheet_names = [sheet.title for sheet in wb.worksheets]
wb.close()
return sheet_names
except Exception as e:
logger.warning(f"Failed to get sheet names: {e}")
return []
def get_sheet_name_for_table(self, table_index: int) -> str | None:
"""Get the sheet name for a table based on its index.
Uses the last conversion's document structure to map table to page to sheet.
Args:
table_index: Index of the table (0-based, from HTML extraction order)
Returns:
Sheet name or None if not available
"""
if not self._last_doc or not self._last_sheet_names:
return None
try:
# Get the table from Docling document
if table_index >= len(self._last_doc.tables):
return None
table = self._last_doc.tables[table_index]
# Get page number from table's provenance
if hasattr(table, 'prov') and table.prov:
# Get first provenance entry
first_prov = table.prov[0]
if hasattr(first_prov, 'page_no'):
page_no = first_prov.page_no
# Docling uses 1-indexed page_no, convert to 0-indexed for sheet name lookup
sheet_index = page_no - 1
# Map page number to sheet name
if 0 <= sheet_index < len(self._last_sheet_names):
return self._last_sheet_names[sheet_index]
elif sheet_index >= len(self._last_sheet_names) and self._last_sheet_names:
# Docling sometimes returns page_no beyond sheet count
# Use the last sheet as fallback
logger.debug(
f"Table {table_index} has page_no={page_no} (sheet_index={sheet_index}) "
f"but only {len(self._last_sheet_names)} sheets. "
f"Using last sheet as fallback."
)
return self._last_sheet_names[-1]
return None
except Exception as e:
logger.debug(f"Failed to get sheet name for table {table_index}: {e}")
return None
# Convenience functions for easy imports
def convert_excel_to_html(file_path: str) -> Optional[str]:
"""
Convenience function to convert Excel to HTML.
Args:
file_path: Path to Excel file (.xlsx, .xls, or .xlsm)
Returns:
HTML string or None on failure
"""
converter = DocumentToHtmlConverter()
return converter.convert_to_html(file_path)
def convert_word_to_markdown(file_path: str) -> Optional[str]:
"""
Convenience function to convert Word to Markdown.
Args:
file_path: Path to Word file (.docx, .doc, or .docm)
Returns:
Markdown string or None on failure
"""
converter = DocumentToHtmlConverter()
return converter.convert_to_html(file_path)