parsing.py

from __future__ import annotations

import re
from typing import Iterable

import pandas as pd


HEADING_RE = re.compile(r"^\s{0,3}(#{1,6})\s+(.*?)\s*$")
TABLE_DIVIDER_CELL_RE = re.compile(r"^:?-{3,}:?$")


def extract_markdown_section(markdown: str, heading: str) -> str | None:
    lines = markdown.splitlines()
    heading_normalized = heading.strip().lower()

    start_index: int | None = None
    heading_level: int | None = None

    for index, line in enumerate(lines):
        match = HEADING_RE.match(line)
        if not match:
            continue
        title = match.group(2).strip().lower()
        if title == heading_normalized:
            start_index = index + 1
            heading_level = len(match.group(1))
            break

    if start_index is None or heading_level is None:
        return None

    end_index = len(lines)
    for index in range(start_index, len(lines)):
        match = HEADING_RE.match(lines[index])
        if not match:
            continue
        next_heading_level = len(match.group(1))
        if next_heading_level <= heading_level:
            end_index = index
            break

    section = "\n".join(lines[start_index:end_index]).strip()
    return section or None


def extract_performance_metrics_table(markdown: str, heading: str = "Performance Metrics") -> pd.DataFrame | None:
    section = extract_markdown_section(markdown, heading)
    if not section:
        return None

    blocks = _extract_table_blocks(section)
    for block in blocks:
        table = _parse_markdown_table_block(block)
        if table is None or table.empty:
            continue

        normalized_columns = {str(col).strip().lower() for col in table.columns}
        if "class" in normalized_columns:
            return table

    if blocks:
        fallback = _parse_markdown_table_block(blocks[0])
        if fallback is not None and not fallback.empty:
            return fallback

    return None


def _extract_table_blocks(text: str) -> list[list[str]]:
    lines = text.splitlines()
    blocks: list[list[str]] = []
    current_block: list[str] = []

    for line in lines:
        if "|" in line:
            current_block.append(line)
            continue

        if len(current_block) >= 2:
            blocks.append(current_block)
        current_block = []

    if len(current_block) >= 2:
        blocks.append(current_block)

    return blocks


def _parse_markdown_table_block(lines: Iterable[str]) -> pd.DataFrame | None:
    rows = [_split_markdown_row(line) for line in lines if line.strip()]
    if len(rows) < 2:
        return None

    header_index = None
    for index in range(len(rows) - 1):
        if _is_divider_row(rows[index + 1]):
            header_index = index
            break

    if header_index is None:
        return None

    header = [cell.strip() for cell in rows[header_index]]
    if not header or len(header) < 2:
        return None

    data_rows = rows[header_index + 2 :]
    normalized_rows: list[list[str]] = []

    for row in data_rows:
        if _is_divider_row(row):
            continue

        padded = row[: len(header)] + [""] * max(0, len(header) - len(row))
        if any(cell.strip() for cell in padded):
            normalized_rows.append(padded)

    if not normalized_rows:
        return None

    dataframe = pd.DataFrame(normalized_rows, columns=header)
    dataframe.columns = [str(column).strip() for column in dataframe.columns]
    return dataframe


def _split_markdown_row(line: str) -> list[str]:
    raw = line.strip().strip("|")
    return [cell.strip() for cell in raw.split("|")]


def _is_divider_row(row: Iterable[str]) -> bool:
    cells = [cell.strip() for cell in row]
    if not cells:
        return False
    return all(TABLE_DIVIDER_CELL_RE.match(cell) for cell in cells if cell)