"""Table extraction service for parsing markdown tables. This service extracts tables from markdown content and provides structured access to table data including headers and rows. """ import re from dataclasses import dataclass from typing import Optional from src.core.logging import get_logger logger = get_logger(__name__) @dataclass class TableCell: """Represents a single cell in a table.""" content: str row_index: int col_index: int @dataclass class TableRow: """Represents a single row in a table.""" cells: list[str] row_index: int def __getitem__(self, index: int) -> str: """Get cell value by column index.""" return self.cells[index] if 0 <= index < len(self.cells) else "" def __len__(self) -> int: """Get number of cells in row.""" return len(self.cells) @dataclass class ExtractedTable: """Represents an extracted table with headers and data rows.""" headers: list[str] rows: list[TableRow] source_start_line: int source_end_line: int raw_markdown: str @property def num_columns(self) -> int: """Get number of columns in table.""" return len(self.headers) @property def num_rows(self) -> int: """Get number of data rows (excluding header).""" return len(self.rows) def get_cell(self, row_index: int, col_index: int) -> str: """Get cell value by row and column index. Args: row_index: Row index (0-based, excluding header) col_index: Column index (0-based) Returns: Cell content or empty string if out of bounds """ if 0 <= row_index < len(self.rows): return self.rows[row_index][col_index] return "" def get_column(self, col_index: int) -> list[str]: """Get all values in a column. Args: col_index: Column index (0-based) Returns: List of cell values in the column """ return [row[col_index] for row in self.rows] def to_dict_list(self) -> list[dict[str, str]]: """Convert table to list of dictionaries. Each row becomes a dictionary mapping header names to cell values. Returns: List of row dictionaries """ result = [] for row in self.rows: row_dict = {} for i, header in enumerate(self.headers): row_dict[header] = row[i] if i < len(row.cells) else "" result.append(row_dict) return result class MarkdownTableExtractor: """Extracts tables from markdown content. This class parses markdown content and extracts all tables, providing structured access to headers and data rows. """ # Regex pattern to match markdown table rows # Matches: | cell1 | cell2 | cell3 | TABLE_ROW_PATTERN = re.compile(r'^\s*\|(.+)\|\s*$') # Regex pattern to match separator rows # Matches: |-------|-------|-------| SEPARATOR_PATTERN = re.compile(r'^\s*\|[\s\-:|]+\|\s*$') def __init__(self): """Initialize the table extractor.""" pass def extract_tables(self, markdown_content: str) -> list[ExtractedTable]: """Extract all tables from markdown content. Args: markdown_content: Markdown text content Returns: List of ExtractedTable objects """ if not markdown_content: return [] lines = markdown_content.split('\n') tables = [] current_table_lines = [] table_start_line = -1 in_table = False for line_num, line in enumerate(lines): is_table_row = self.TABLE_ROW_PATTERN.match(line) is_separator = self.SEPARATOR_PATTERN.match(line) if is_table_row or is_separator: if not in_table: # Start of new table in_table = True table_start_line = line_num current_table_lines = [line] else: # Continue current table current_table_lines.append(line) else: # Not a table row if in_table: # End of current table table = self._parse_table_lines( current_table_lines, table_start_line, line_num - 1 ) if table: tables.append(table) # Reset state in_table = False current_table_lines = [] table_start_line = -1 # Handle table at end of content if in_table and current_table_lines: table = self._parse_table_lines( current_table_lines, table_start_line, len(lines) - 1 ) if table: tables.append(table) logger.info(f"Extracted {len(tables)} tables from markdown") return tables def _parse_table_lines( self, lines: list[str], start_line: int, end_line: int ) -> Optional[ExtractedTable]: """Parse a sequence of table lines into an ExtractedTable. Args: lines: List of markdown table lines start_line: Starting line number in source end_line: Ending line number in source Returns: ExtractedTable object or None if parsing fails """ if len(lines) < 2: # Need at least header and separator return None # Extract header (first line) header_line = lines[0] headers = self._parse_row(header_line) if not headers: return None # Find separator line (should be second line) separator_found = False data_start_idx = 1 for i in range(1, min(3, len(lines))): # Check first 2-3 lines for separator if self.SEPARATOR_PATTERN.match(lines[i]): separator_found = True data_start_idx = i + 1 break if not separator_found: logger.debug(f"No separator found in table at line {start_line}") return None # Parse data rows rows = [] for row_idx, line in enumerate(lines[data_start_idx:]): if self.TABLE_ROW_PATTERN.match(line): cells = self._parse_row(line) if cells: # Ensure row has same number of columns as header # Pad with empty strings if needed while len(cells) < len(headers): cells.append("") # Truncate if too many columns cells = cells[:len(headers)] rows.append(TableRow(cells=cells, row_index=row_idx)) # Create table object raw_markdown = '\n'.join(lines) table = ExtractedTable( headers=headers, rows=rows, source_start_line=start_line, source_end_line=end_line, raw_markdown=raw_markdown ) logger.debug( f"Parsed table at line {start_line}: " f"{len(headers)} columns, {len(rows)} rows" ) return table def _parse_row(self, line: str) -> list[str]: """Parse a single table row line into cells. Args: line: Markdown table row line Returns: List of cell contents (stripped of whitespace) """ # Remove leading/trailing pipe and whitespace line = line.strip() if line.startswith('|'): line = line[1:] if line.endswith('|'): line = line[:-1] # Split by pipe and strip whitespace from each cell cells = [cell.strip() for cell in line.split('|')] return cells def get_table_extractor() -> MarkdownTableExtractor: """Factory function to get an instance of MarkdownTableExtractor.""" return MarkdownTableExtractor()