"""Table boundary detection for Excel sheets.""" from typing import Optional import json import structlog from openpyxl import Workbook from openpyxl.worksheet.worksheet import Worksheet from openpyxl.cell.cell import Cell, MergedCell from openpyxl.utils import get_column_letter, column_index_from_string from openai import AzureOpenAI from src.extraction.models import TableBoundary from src.core.config import settings from src.llm import get_chat_client, get_chat_model_name, strip_thinking logger = structlog.get_logger(__name__) # LLM detection prompt template DETECTION_PROMPT_TEMPLATE = """You are analyzing the first 20 rows of an Excel sheet to identify table boundaries. Sheet data: {sheet_sample} Identify: 1. Header row number (the row with column names) 2. Data start row number (first row with actual data) 3. Estimated end row (based on pattern, or say "continue to end") 4. Start column letter 5. End column letter If multiple tables exist, identify each one separately. Respond in JSON format: {{ "tables": [ {{ "header_row": 3, "data_start_row": 4, "estimated_end": "continue to end", "start_col": "A", "end_col": "F" }} ] }} """ class TableDetector: """Detects table boundaries in Excel sheets using heuristics and optional LLM assistance.""" def detect_tables(self, workbook: Workbook) -> list[TableBoundary]: """Detect all tables in a workbook. Args: workbook: openpyxl Workbook object to analyze Returns: List of TableBoundary objects for all detected tables Raises: ValueError: If workbook is invalid """ if not workbook or not workbook.worksheets: logger.warning("detect_tables_called_with_empty_workbook") return [] all_tables = [] for sheet in workbook.worksheets: logger.debug( "detecting_tables_in_sheet", sheet_name=sheet.title, max_row=sheet.max_row, max_col=sheet.max_column, ) # Detect boundaries in this sheet boundaries = self._detect_boundaries_in_sheet(sheet) # Convert to TableBoundary objects for idx, boundary in enumerate(boundaries, start=1): table = TableBoundary( sheet=sheet.title, table_id=idx, start_row=boundary["start_row"], end_row=boundary["end_row"], start_col=boundary["start_col"], end_col=boundary["end_col"], title=boundary.get("title"), ) all_tables.append(table) logger.info( "tables_detected_in_sheet", sheet_name=sheet.title, table_count=len(boundaries), ) return all_tables def _detect_boundaries_in_sheet(self, sheet: Worksheet) -> list[dict]: """Detect table boundaries in a single sheet using heuristics. Args: sheet: Worksheet to analyze Returns: List of boundary dictionaries with start_row, end_row, start_col, end_col """ # Handle empty sheets if sheet.max_row is None or sheet.max_row == 0: logger.debug("empty_sheet_detected", sheet_name=sheet.title) return [] # Handle single-row sheets (header-only tables) if sheet.max_row == 1: logger.debug("single_row_sheet_detected", sheet_name=sheet.title) first_row = list(sheet.iter_rows(min_row=1, max_row=1))[0] if self._is_blank_row(first_row): return [] return [ { "start_row": 1, "end_row": 1, "start_col": self._get_first_populated_col(first_row), "end_col": self._get_last_populated_col(first_row), } ] boundaries = [] current_table = None blank_row_count = 0 pending_title = None # Title from merged row to assign to next table for row_idx, row in enumerate(sheet.iter_rows(), start=1): # Check if row is blank if self._is_blank_row(row): blank_row_count += 1 # If we have a current table and 2+ blank rows, end it if current_table and blank_row_count >= 2: current_table["end_row"] = row_idx - blank_row_count boundaries.append(current_table) current_table = None pending_title = None # Reset pending title after table ends logger.debug( "table_ended_by_blank_rows", sheet_name=sheet.title, end_row=current_table["end_row"] if current_table else row_idx, ) continue # Reset blank row counter blank_row_count = 0 # Check for merged title rows - capture the title text if self._is_merged_title_row(row, sheet): title_text = self._get_merged_title_text(row, sheet) if title_text: pending_title = title_text logger.debug( "captured_table_title", sheet_name=sheet.title, row=row_idx, title=title_text, ) continue # Check if this is a potential header row if self._is_header_row(row) and not current_table: current_table = { "start_row": row_idx, "start_col": self._get_first_populated_col(row), "end_col": self._get_last_populated_col(row), "title": pending_title, # Assign pending title to this table } pending_title = None # Reset pending title logger.debug( "table_started", sheet_name=sheet.title, start_row=row_idx, start_col=current_table["start_col"], end_col=current_table["end_col"], title=current_table["title"], ) # Close last table at end of sheet if current_table: current_table["end_row"] = sheet.max_row boundaries.append(current_table) logger.debug( "table_closed_at_sheet_end", sheet_name=sheet.title, end_row=sheet.max_row, ) # Handle sheets with no clear table structure if not boundaries and sheet.max_row > 0: logger.warning( "no_clear_table_structure", sheet_name=sheet.title, max_row=sheet.max_row, attempting_best_effort=True, ) # Best-effort: treat entire sheet as one table first_row = list(sheet.iter_rows(min_row=1, max_row=1))[0] if not self._is_blank_row(first_row): boundaries.append( { "start_row": 1, "end_row": sheet.max_row, "start_col": self._get_first_populated_col(first_row), "end_col": self._get_last_populated_col(first_row), } ) # Expand column ranges based on merged cells in each boundary for boundary in boundaries: start_col, end_col = self._expand_column_range_from_merges( sheet, boundary["start_row"], boundary["end_row"], boundary["start_col"], boundary["end_col"], ) if start_col != boundary["start_col"] or end_col != boundary["end_col"]: logger.debug( "column_range_expanded", sheet_name=sheet.title, original_range=f"{boundary['start_col']}:{boundary['end_col']}", expanded_range=f"{start_col}:{end_col}", ) boundary["start_col"] = start_col boundary["end_col"] = end_col return boundaries def _is_blank_row(self, row: tuple[Cell, ...]) -> bool: """Check if a row is completely blank. Args: row: Tuple of Cell objects Returns: True if row is blank, False otherwise """ for cell in row: if isinstance(cell, MergedCell): continue if cell.value is not None and str(cell.value).strip(): return False return True def _is_header_row(self, row: tuple[Cell, ...]) -> bool: """Check if a row is likely a header row. Heuristics: - Most cells (>50%) contain text (not numbers, not blank) - At least 2 cells are populated (or merged cells are present) Args: row: Tuple of Cell objects Returns: True if row is likely a header, False otherwise """ text_count = 0 populated_count = 0 merged_cell_count = 0 for cell in row: if isinstance(cell, MergedCell): # Merged cells indicate structured headers - count them as "populated text" merged_cell_count += 1 continue if cell.value is not None and str(cell.value).strip(): populated_count += 1 # Check if value is text (not purely numeric) value_str = str(cell.value).strip() try: float(value_str) # It's a number, not text except (ValueError, AttributeError): # It's text text_count += 1 # Merged cells count as populated text cells for header detection populated_count += merged_cell_count text_count += merged_cell_count # Need at least 2 populated cells if populated_count < 2: return False # More than 50% should be text text_ratio = text_count / populated_count if populated_count > 0 else 0 return text_ratio > 0.5 def _is_merged_title_row(self, row: tuple[Cell, ...], sheet: Worksheet) -> bool: """Check if a row is a document title row (single cell merged across full width). A true "title row" is one where: - There's a single merged range spanning most/all columns - Contains just one value (the document title) Multi-level headers with partial merges (e.g., "Products" spanning B:C) should NOT be considered title rows. Args: row: Tuple of Cell objects sheet: Worksheet object for merged cell lookup Returns: True if row is a document title (full-width merge), False otherwise """ if not row: return False row_idx = row[0].row row_col_count = len(row) # Find merged ranges that include this row merged_ranges_in_row = [] for merged_range in sheet.merged_cells.ranges: if merged_range.min_row <= row_idx <= merged_range.max_row: merged_ranges_in_row.append(merged_range) if not merged_ranges_in_row: return False # Check if there's a single merge spanning most of the row (>= 70%) for merged_range in merged_ranges_in_row: col_span = merged_range.max_col - merged_range.min_col + 1 # If a single merge spans 70%+ of the row's columns, it's likely a title if col_span >= row_col_count * 0.7: # Also check if it's a single-row merge (not a multi-row header structure) row_span = merged_range.max_row - merged_range.min_row + 1 if row_span == 1: logger.debug( "detected_title_row", row=row_idx, merged_range=str(merged_range), col_span=col_span, ) return True # Multiple partial merges = multi-level header structure, not a title return False def _get_merged_title_text(self, row: tuple[Cell, ...], sheet: Worksheet) -> Optional[str]: """Extract the text from a merged title row. Args: row: Tuple of Cell objects sheet: Worksheet object for merged cell lookup Returns: Title text or None if no text found """ if not row: return None row_idx = row[0].row # Find the merged cell that spans most of the row and get its value for merged_range in sheet.merged_cells.ranges: if merged_range.min_row <= row_idx <= merged_range.max_row: # Get the top-left cell of the merge (which contains the value) top_left_cell = sheet.cell(merged_range.min_row, merged_range.min_col) if top_left_cell.value is not None: return str(top_left_cell.value).strip() # Fallback: check regular cells in row for cell in row: if not isinstance(cell, MergedCell) and cell.value is not None: value = str(cell.value).strip() if value: return value return None def _get_first_populated_col(self, row: tuple[Cell, ...]) -> str: """Get the letter of the first populated column in a row. Handles both regular cells and merged cells by checking if any cell (including merged ones) represents populated data. Args: row: Tuple of Cell objects Returns: Column letter (e.g., "A") or "A" if no populated cells found """ for cell in row: # For merged cells, they still represent a column position if isinstance(cell, MergedCell): # MergedCell has 'column' (int) but not 'column_letter' return get_column_letter(cell.column) if cell.value is not None and str(cell.value).strip(): return cell.column_letter return "A" def _get_last_populated_col(self, row: tuple[Cell, ...]) -> str: """Get the letter of the last populated column in a row. Handles both regular cells and merged cells by checking if any cell (including merged ones) represents populated data. Args: row: Tuple of Cell objects Returns: Column letter (e.g., "F") or "A" if no populated cells found """ last_col = "A" for cell in row: # For merged cells, they still represent a column position if isinstance(cell, MergedCell): # MergedCell has 'column' (int) but not 'column_letter' last_col = get_column_letter(cell.column) elif cell.value is not None and str(cell.value).strip(): last_col = cell.column_letter return last_col def _expand_column_range_from_merges( self, sheet: Worksheet, start_row: int, end_row: int, start_col: str, end_col: str ) -> tuple[str, str]: """Expand column range based on merged cell ranges in the table area. If there are merged cells that extend beyond the detected column range, expand the range to include them. Args: sheet: Worksheet to analyze start_row: Table start row end_row: Table end row start_col: Initially detected start column end_col: Initially detected end column Returns: Tuple of (expanded_start_col, expanded_end_col) """ start_col_idx = column_index_from_string(start_col) end_col_idx = column_index_from_string(end_col) for merged_range in sheet.merged_cells.ranges: # Check if this merged range intersects with our table rows if merged_range.min_row <= end_row and merged_range.max_row >= start_row: # Expand column range if needed if merged_range.min_col < start_col_idx: start_col_idx = merged_range.min_col if merged_range.max_col > end_col_idx: end_col_idx = merged_range.max_col return get_column_letter(start_col_idx), get_column_letter(end_col_idx) def _llm_detect_boundaries( self, sheet: Worksheet, document_id: Optional[str] = None ) -> Optional[list[dict]]: """Use Azure OpenAI to detect table boundaries in ambiguous cases. Args: sheet: Worksheet to analyze document_id: Optional document ID for logging Returns: List of boundary dictionaries or None if LLM call fails """ # Check if Azure OpenAI is configured if not settings.azure_openai_api_key or not settings.azure_openai_endpoint: logger.info( "llm_detection_skipped_no_config", sheet_name=sheet.title, document_id=document_id, ) return None try: # Extract first 20 rows as text sample sheet_sample = self._extract_sheet_sample(sheet, max_rows=20) # Create LLM client via factory client = get_chat_client() # Format prompt prompt = DETECTION_PROMPT_TEMPLATE.format(sheet_sample=sheet_sample) # Call LLM response = client.chat.completions.create( model=get_chat_model_name(), messages=[ { "role": "system", "content": "You are an Excel table structure analyzer. Respond only with valid JSON.", }, {"role": "user", "content": prompt}, ], temperature=0.1, # Low temperature for consistent output max_tokens=500, response_format={"type": "json_object"}, ) # Parse response content = strip_thinking(response.choices[0].message.content) if not content: return [] result = json.loads(content) # Log token usage for cost tracking logger.info( "llm_detection_completed", sheet_name=sheet.title, document_id=document_id, prompt_tokens=response.usage.prompt_tokens, completion_tokens=response.usage.completion_tokens, total_tokens=response.usage.total_tokens, ) # Convert LLM response to boundary format boundaries = [] for table in result.get("tables", []): boundary = { "start_row": table.get("header_row", 1), "end_row": ( sheet.max_row if table.get("estimated_end") == "continue to end" else table.get("estimated_end", sheet.max_row) ), "start_col": table.get("start_col", "A"), "end_col": table.get("end_col", "A"), } boundaries.append(boundary) return boundaries except json.JSONDecodeError as e: logger.warning( "llm_detection_json_parse_error", sheet_name=sheet.title, document_id=document_id, error=str(e), ) return None except TimeoutError as e: logger.warning( "llm_detection_timeout", sheet_name=sheet.title, document_id=document_id, error=str(e), ) return None except ConnectionError as e: logger.warning( "llm_detection_connection_error", sheet_name=sheet.title, document_id=document_id, error=str(e), ) return None except Exception as e: logger.warning( "llm_detection_failed", sheet_name=sheet.title, document_id=document_id, error=str(e), error_type=type(e).__name__, ) return None def _extract_sheet_sample(self, sheet: Worksheet, max_rows: int = 20) -> str: """Extract first N rows of sheet as formatted text for LLM. Args: sheet: Worksheet to extract from max_rows: Maximum number of rows to extract Returns: Formatted text representation of sheet rows """ lines = [] row_limit = min(max_rows, sheet.max_row or 0) for row_idx, row in enumerate(sheet.iter_rows(max_row=row_limit), start=1): cells = [] for cell in row: if isinstance(cell, MergedCell): cells.append("[MERGED]") else: value = cell.value if cell.value is not None else "" cells.append(str(value)) lines.append(f"Row {row_idx}: {' | '.join(cells)}") return "\n".join(lines)