"""Multi-level header extraction from Excel sheets.""" import structlog 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 src.extraction.models import TableBoundary, HeaderInfo from src.core.exceptions import ExtractionError logger = structlog.get_logger(__name__) class HeaderExtractor: """Extracts multi-level column headers from Excel tables. Supports 1-4 levels of hierarchical headers with merged cell handling. Generates flattened column names with " > " separator. """ def extract_headers( self, sheet: Worksheet, table_boundary: TableBoundary ) -> dict[str, HeaderInfo]: """Extract headers from a detected table. Args: sheet: openpyxl Worksheet object table_boundary: TableBoundary from table detection Returns: Dictionary mapping column_letter -> HeaderInfo Raises: ExtractionError: If header extraction fails """ try: # Detect header depth depth = self._detect_header_depth(sheet, table_boundary) logger.info( "header_depth_detected", sheet_name=sheet.title, table_id=table_boundary.table_id, depth=depth, ) # Extract header rows header_rows = self._extract_header_rows(sheet, table_boundary, depth) # Resolve merged cells resolved_headers = self._resolve_merged_headers(sheet, header_rows, table_boundary, depth) # Build HeaderInfo for each column headers = self._build_header_info(resolved_headers, table_boundary) logger.info( "headers_extracted", sheet_name=sheet.title, table_id=table_boundary.table_id, column_count=len(headers), depth=depth, ) return headers except Exception as e: logger.error( "header_extraction_failed", sheet_name=sheet.title, table_id=table_boundary.table_id, error=str(e), ) raise ExtractionError( code="HEADER_EXTRACTION_FAILED", message=f"Failed to extract headers from table {table_boundary.table_id}", details={"sheet": sheet.title, "error": str(e)}, ) def _detect_header_depth( self, sheet: Worksheet, table_boundary: TableBoundary ) -> int: """Detect how many rows comprise the header. Strategy (based on merged cells and row structure): 1. Rows with horizontal merged cells = definitely header rows 2. The first row after merged rows with no merging = column name row (last header) 3. When two consecutive non-merged rows have the same structure, data has started 4. Max depth: 4 rows Args: sheet: openpyxl Worksheet object table_boundary: TableBoundary from table detection Returns: Header depth (1-4) """ start_col_idx = column_index_from_string(table_boundary.start_col) end_col_idx = column_index_from_string(table_boundary.end_col) # Collect info about each row (up to 5 rows to detect pattern) row_info = [] for row_offset in range(min(5, table_boundary.end_row - table_boundary.start_row + 1)): row_idx = table_boundary.start_row + row_offset # Get row cells and check for merges cells = [] has_merge = False for col_idx in range(start_col_idx, end_col_idx + 1): cell = sheet.cell(row=row_idx, column=col_idx) cells.append(cell) if isinstance(cell, MergedCell): has_merge = True # Also check if this row starts a horizontal merge has_horizontal_merge = self._row_has_horizontal_merge( sheet, row_idx, start_col_idx, end_col_idx ) row_info.append({ "row_idx": row_idx, "cells": cells, "has_merge": has_merge or has_horizontal_merge, "structure": self._get_row_structure(cells), }) # Determine header depth using the new algorithm depth = self._analyze_header_depth(row_info) # Minimum 1 row, maximum 4 rows return max(1, min(depth, 4)) def _row_has_horizontal_merge( self, sheet: Worksheet, row_idx: int, start_col: int, end_col: int ) -> bool: """Check if a row contains any horizontal merged cells. Args: sheet: openpyxl Worksheet object row_idx: Row index to check start_col: Start column index end_col: End column index Returns: True if row has horizontal merges """ for merged_range in sheet.merged_cells.ranges: # Check if merge is in this row and spans multiple columns if merged_range.min_row <= row_idx <= merged_range.max_row: if merged_range.min_col >= start_col and merged_range.max_col <= end_col: # Horizontal merge (spans columns) if merged_range.max_col > merged_range.min_col: return True return False def _get_row_structure(self, cells: list[Cell]) -> tuple: """Get a structural signature of a row for comparison. Returns a tuple representing the pattern of the row: - Which columns are populated - Approximate length categories of values Args: cells: List of Cell objects Returns: Tuple representing row structure """ structure = [] for cell in cells: if isinstance(cell, MergedCell): structure.append("M") # Merged elif cell.value is None or str(cell.value).strip() == "": structure.append("E") # Empty else: value_str = str(cell.value).strip() # Categorize by length: S(hort), M(edium), L(ong) if len(value_str) <= 5: structure.append("S") elif len(value_str) <= 20: structure.append("M") else: structure.append("L") return tuple(structure) def _analyze_header_depth(self, row_info: list[dict]) -> int: """Analyze row information to determine header depth. Rules: 1. Rows with merged cells are definitely headers 2. Rows with symbols (○×△) are definitely data 3. Rows with small integers immediately after merged rows may be sub-headers 4. Rows with real data patterns (IDs, larger numbers, dates) are data 5. Text-only rows before the first data row are headers 6. After merged rows, include up to 2 more non-merged rows as column names Args: row_info: List of dictionaries with row analysis info Returns: Header depth """ if not row_info: return 1 depth = 0 last_merged_row_idx = -1 non_merged_after_merge_count = 0 for i, info in enumerate(row_info): # Check for symbols in this row (definitely data) if self._row_has_symbols(info["cells"]): break # Check if row is empty if self._row_is_empty(info["cells"]): if i == 0: depth = 1 break # Check for numeric values - but be smart about sub-headers if self._row_has_numeric_values(info["cells"]): # If previous row had merges, this might be sub-header row with indices if last_merged_row_idx == i - 1: # Check if these are likely column indices (small sequential integers) if self._is_column_index_row(info["cells"]): depth = i + 1 # This is the last header row (column indices) break # Otherwise it's a data row break if info["has_merge"]: depth = i + 1 last_merged_row_idx = i non_merged_after_merge_count = 0 else: depth = i + 1 if last_merged_row_idx >= 0: non_merged_after_merge_count += 1 # Allow only 1 non-merged row after merges if non_merged_after_merge_count >= 1: break return max(depth, 1) def _is_column_index_row(self, cells: list[Cell]) -> bool: """Check if a row contains column index values (like 1, 2, 3... or A, B, C...). These are typically sub-headers for merged column groups. Must have multiple consecutive small integers to be considered column indices. Args: cells: List of Cell objects Returns: True if row appears to be column indices """ values = [] for cell in cells: if isinstance(cell, MergedCell): continue if cell.value is None or str(cell.value).strip() == "": continue values.append(str(cell.value).strip()) if not values: return False # Need at least 3 values to be considered a column index row if len(values) < 3: return False # Count small sequential integers small_int_values = [] for v in values: try: num = int(v) # Small integers (typically column indices 1-20) if 1 <= num <= 20: small_int_values.append(num) except ValueError: pass # Need at least 3 small integers that look like column indices if len(small_int_values) < 3: return False # Check if they are mostly sequential (column indices like 1,2,3,4,5...) if len(small_int_values) >= 3: sorted_ints = sorted(small_int_values) # Check if they form a consecutive sequence is_sequential = all( sorted_ints[i+1] - sorted_ints[i] == 1 for i in range(len(sorted_ints) - 1) ) if is_sequential and len(small_int_values) / len(values) >= 0.5: return True return False def _row_is_empty(self, cells: list[Cell]) -> bool: """Check if a row is empty (all cells empty or whitespace). Args: cells: List of Cell objects Returns: True if row is empty """ for cell in cells: if isinstance(cell, MergedCell): continue if cell.value is not None and str(cell.value).strip(): return False return True def _row_has_numeric_values(self, cells: list[Cell]) -> bool: """Check if a row contains numeric values or ID-like patterns (indicating data). Args: cells: List of Cell objects Returns: True if row has numeric values or ID-like patterns """ import re numeric_count = 0 id_like_count = 0 total_count = 0 # Pattern for ID-like values: letter(s) followed by numbers (E001, P002, etc) id_pattern = re.compile(r'^[A-Za-z]{1,4}\d{2,}$') # Pattern for date-like values (YYYY-MM-DD or MM/DD/YYYY) date_pattern = re.compile(r'^(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{2,4})$') # Pattern for currency values ($1,234.56 or $1234) currency_pattern = re.compile(r'^[\$\u00A3\u20AC\u00A5][\d,]+\.?\d*$') # Pattern for percentage values (75%, 12.5%) percent_pattern = re.compile(r'^\d+\.?\d*%$') for cell in cells: if isinstance(cell, MergedCell): continue if cell.value is None or str(cell.value).strip() == "": continue total_count += 1 value_str = str(cell.value).strip() # Check for numeric values (including negative) try: float(value_str.replace(',', '')) numeric_count += 1 continue except (ValueError, TypeError): pass # Check for ID-like patterns (E001, P002, etc) if id_pattern.match(value_str): id_like_count += 1 continue # Check for date-like patterns if date_pattern.match(value_str): numeric_count += 1 continue # Check for currency patterns if currency_pattern.match(value_str): numeric_count += 1 continue # Check for percentage patterns if percent_pattern.match(value_str): numeric_count += 1 continue # If any ID-like values found, it's likely a data row if id_like_count > 0: return True # If more than 30% of cells are numeric/currency/date, it's likely a data row if total_count > 0 and numeric_count / total_count > 0.3: return True return False def _structures_are_similar(self, struct1: tuple, struct2: tuple) -> bool: """Check if two row structures are similar (indicating data rows). Two rows are similar if they have the same pattern of populated columns and similar value length distributions. Args: struct1: First row structure tuple struct2: Second row structure tuple Returns: True if structures are similar """ if len(struct1) != len(struct2): return False # Count matches matches = 0 for s1, s2 in zip(struct1, struct2): # Empty matches empty, or both are populated (regardless of length) if s1 == s2: matches += 1 elif s1 != "E" and s1 != "M" and s2 != "E" and s2 != "M": # Both are populated (S, M, or L), consider as match matches += 1 # If more than 70% match, consider similar return matches / len(struct1) >= 0.7 if struct1 else False def _row_has_symbols(self, cells: list[Cell]) -> bool: """Check if a row contains symbol characters indicating data. Args: cells: List of Cell objects Returns: True if row contains symbols """ symbol_chars = set("○×△◎★☆→※●◇□■▲▼►◄♦♠♣♥✓✗✔✘") for cell in cells: if isinstance(cell, MergedCell): continue if cell.value is None: continue value_str = str(cell.value).strip() if len(value_str) <= 2 and any(c in symbol_chars for c in value_str): return True return False def _is_header_like(self, row: list[Cell]) -> bool: """Check if a row is likely a header row (text-heavy). Heuristics: - Most cells (>60%) contain text (not numbers) - Symbol characters (Unicode symbols like ○×△) indicate data rows - Empty cells are ignored Args: row: List of Cell objects Returns: True if row is header-like """ text_count = 0 numeric_count = 0 symbol_count = 0 # Common symbol characters that indicate data, not headers # These are typically used in Japanese/Asian documents to mark status symbol_chars = set("○×△◎★☆→※●◇□■▲▼►◄♦♠♣♥✓✗✔✘") for cell in row: # Skip merged cells and empty cells if isinstance(cell, MergedCell): continue if cell.value is None or not str(cell.value).strip(): continue # Check if value is numeric or text value_str = str(cell.value).strip() # Check if value is primarily a symbol character # A cell is "symbol-like" if it's short (1-2 chars) AND contains symbol chars if len(value_str) <= 2 and any(c in symbol_chars for c in value_str): symbol_count += 1 continue try: float(value_str) numeric_count += 1 except (ValueError, AttributeError): text_count += 1 total_populated = text_count + numeric_count + symbol_count # If no populated cells, not a header if total_populated == 0: return False # If row has symbol characters, likely a data row # Symbol dictionaries have symbols in every row of data if symbol_count > 0: # If any cell contains a symbol, this is probably a data row # (headers rarely contain ○, ×, △, etc.) return False # If mostly text, it's a header text_ratio = text_count / total_populated return text_ratio > 0.6 def _extract_header_rows( self, sheet: Worksheet, table_boundary: TableBoundary, depth: int ) -> list[list[Cell]]: """Extract header rows from sheet. Args: sheet: openpyxl Worksheet object table_boundary: TableBoundary from table detection depth: Number of header rows Returns: List of header rows (each row is list of Cell objects) """ header_rows = [] start_col_idx = column_index_from_string(table_boundary.start_col) end_col_idx = column_index_from_string(table_boundary.end_col) for row_offset in range(depth): row_idx = table_boundary.start_row + row_offset row = [] for col_idx in range(start_col_idx, end_col_idx + 1): cell = sheet.cell(row=row_idx, column=col_idx) row.append(cell) header_rows.append(row) return header_rows def _resolve_merged_headers( self, sheet: Worksheet, header_rows: list[list[Cell]], table_boundary: TableBoundary, depth: int, ) -> list[list[str]]: """Resolve merged cells in header rows. For each merged range in headers: - Propagate top-left value to all cells in range Args: sheet: openpyxl Worksheet object header_rows: List of header rows table_boundary: TableBoundary from table detection depth: Number of header rows Returns: 2D list of resolved header values (rows x columns) """ resolved = [] start_col_idx = column_index_from_string(table_boundary.start_col) end_col_idx = column_index_from_string(table_boundary.end_col) for row_offset in range(depth): row_idx = table_boundary.start_row + row_offset row_values = [] for col_idx in range(start_col_idx, end_col_idx + 1): cell = sheet.cell(row=row_idx, column=col_idx) if isinstance(cell, MergedCell): # Find the merged range containing this cell value = self._get_merged_cell_value(sheet, cell) row_values.append(value) else: # Regular cell value = cell.value if cell.value is not None else "" row_values.append(str(value).strip() if value else "") resolved.append(row_values) return resolved def _get_merged_cell_value(self, sheet: Worksheet, merged_cell: MergedCell) -> str: """Get the value of a merged cell by finding its master cell. Args: sheet: openpyxl Worksheet object merged_cell: MergedCell object Returns: String value from the master cell """ for merged_range in sheet.merged_cells.ranges: if merged_cell.coordinate in merged_range: # Get the top-left cell (master) of the merged range master_cell = sheet.cell( row=merged_range.min_row, column=merged_range.min_col ) value = master_cell.value if master_cell.value is not None else "" return str(value).strip() if value else "" # If not found in any merged range, return empty return "" def _build_header_info( self, resolved_headers: list[list[str]], table_boundary: TableBoundary ) -> dict[str, HeaderInfo]: """Build HeaderInfo objects for each column. Args: resolved_headers: 2D list of resolved header values table_boundary: TableBoundary from table detection Returns: Dictionary mapping column_letter -> HeaderInfo """ headers = {} start_col_idx = column_index_from_string(table_boundary.start_col) end_col_idx = column_index_from_string(table_boundary.end_col) num_cols = end_col_idx - start_col_idx + 1 for col_offset in range(num_cols): col_idx = start_col_idx + col_offset col_letter = get_column_letter(col_idx) # Extract raw levels for this column (one value per header row) raw_levels = [] last_non_empty = "" for row in resolved_headers: value = row[col_offset] if col_offset < len(row) else "" value = value.strip() if value else "" if value: last_non_empty = value raw_levels.append(value) else: # Inherit parent value for empty cells (don't treat as new level) raw_levels.append(last_non_empty) # Remove empty values and compress consecutive duplicates # This fixes the "項番 > 項番 > 項番 > 項番" issue levels = [] for v in raw_levels: if not v: continue # Only add if different from the last level (compress duplicates) if not levels or levels[-1] != v: levels.append(v) # Skip columns with no header values if not levels: logger.debug( "skipping_empty_header_column", sheet_name=table_boundary.sheet, column=col_letter, ) continue # Create flattened display name display = " > ".join(levels) # Create HeaderInfo headers[col_letter] = HeaderInfo( display=display, levels=levels, depth=len(levels), column_letter=col_letter, ) return headers