instant-bom/bom_assistant/ingestion/parser.py

from __future__ import annotations

import csv
import io
import re
from datetime import datetime
from pathlib import Path


_FOOTER_RE = re.compile(
    r"^(Component\s+Groups?|Component\s+Count|Fitted|Total\s+comp|"
    r"Schematic|PCB\s+Variant|BoM\s+Date)\s*[:\-]",
    re.IGNORECASE,
)

_TOTALS_RE = re.compile(
    r"^(Estimated\s+Total|Total\s+Cost|Grand\s+Total)",
    re.IGNORECASE,
)


def parse_bom(content: bytes, filename: str) -> list[dict[str, str]]:
    suffix = Path(filename).suffix.lower()
    if suffix == ".csv":
        return _parse_csv(content)
    elif suffix == ".xlsx":
        return _parse_xlsx(content)
    elif suffix == ".xls":
        return _parse_xls(content)
    else:
        raise ValueError(f"Unsupported format: {suffix!r}")


# ---------------------------------------------------------------------------
# CSV
# ---------------------------------------------------------------------------

def _parse_csv(content: bytes) -> list[dict[str, str]]:
    text = content.decode("utf-8-sig")
    delimiter = _detect_delimiter(text)
    reader = csv.reader(io.StringIO(text), delimiter=delimiter)
    all_rows = list(reader)
    if not all_rows:
        return []

    headers = [h.strip() for h in all_rows[0]]

    result: list[dict[str, str]] = []
    for row in all_rows[1:]:
        # pad short rows
        while len(row) < len(headers):
            row.append("")
        row_dict = {headers[i]: row[i].strip() for i in range(len(headers)) if headers[i]}

        # stop at footer
        first_val = next((v for v in row_dict.values() if v), "")
        if _FOOTER_RE.match(first_val) or _TOTALS_RE.match(first_val):
            break

        # skip all-empty or all-non-alpha rows (e.g. blank semicolon rows)
        vals = [v for v in row_dict.values() if v]
        if not vals or all(not re.search(r"[A-Za-z0-9]", v) for v in vals):
            continue

        result.append(row_dict)
    return result


def _detect_delimiter(text: str) -> str:
    first_line = text.split("\n", 1)[0]
    comma_count = first_line.count(",")
    semi_count = first_line.count(";")
    return ";" if semi_count > comma_count else ","


# ---------------------------------------------------------------------------
# XLSX
# ---------------------------------------------------------------------------

def _parse_xlsx(content: bytes) -> list[dict[str, str]]:
    import openpyxl

    wb = openpyxl.load_workbook(io.BytesIO(content), data_only=True)
    ws = wb.active
    all_rows = list(ws.iter_rows(values_only=True))
    return _extract_table(all_rows)


# ---------------------------------------------------------------------------
# XLS
# ---------------------------------------------------------------------------

def _parse_xls(content: bytes) -> list[dict[str, str]]:
    import xlrd

    wb = xlrd.open_workbook(file_contents=content)
    ws = wb.sheet_by_index(0)
    all_rows = []
    for r in range(ws.nrows):
        row = []
        for c in range(ws.ncols):
            cell = ws.cell(r, c)
            if cell.ctype == xlrd.XL_CELL_DATE:
                try:
                    dt = xlrd.xldate_as_datetime(cell.value, wb.datemode)
                    row.append(dt.isoformat())
                except Exception:
                    row.append(str(cell.value))
            elif cell.ctype == xlrd.XL_CELL_EMPTY:
                row.append(None)
            else:
                row.append(cell.value)
        all_rows.append(tuple(row))
    return _extract_table(all_rows)


# ---------------------------------------------------------------------------
# Shared table extraction (XLSX + XLS)
# ---------------------------------------------------------------------------

def _extract_table(all_rows: list[tuple]) -> list[dict[str, str]]:
    if not all_rows:
        return []

    header_idx, col_offset = _find_header(all_rows)
    if header_idx is None:
        return []

    raw_headers = all_rows[header_idx]
    headers: list[str] = []
    for cell in raw_headers[col_offset:]:
        if cell is None:
            headers.append("")
        else:
            h = str(cell).replace("\n", " ").strip()
            headers.append(h)

    # drop trailing None/empty headers
    while headers and not headers[-1]:
        headers.pop()

    result: list[dict[str, str]] = []
    for row in all_rows[header_idx + 1:]:
        row_slice = row[col_offset: col_offset + len(headers)]
        # pad
        row_slice = list(row_slice) + [""] * (len(headers) - len(row_slice))

        row_dict: dict[str, str] = {}
        for i, h in enumerate(headers):
            if not h:
                continue
            cell = row_slice[i]
            row_dict[h] = _cell_to_str(cell)

        # stop at footer / totals
        first_val = next((v for v in row_dict.values() if v), "")
        if _FOOTER_RE.match(first_val) or _TOTALS_RE.match(first_val):
            break

        # skip rows where every value is empty (blank separator rows)
        if not any(row_dict.values()):
            continue

        result.append(row_dict)
    return result


def _find_header(all_rows: list[tuple]) -> tuple[int | None, int]:
    from bom_assistant.ingestion.column_mapper import _load_synonyms

    all_synonyms: set[str] = set()
    for syns in _load_synonyms().values():
        all_synonyms.update(s.lower() for s in syns)

    for row_idx, row in enumerate(all_rows[:15]):
        string_cells = [c for c in row if c is not None and isinstance(c, str) and c.strip()]
        if len(string_cells) < 2:
            continue
        matches = sum(1 for c in string_cells if c.strip().lower() in all_synonyms)
        if matches >= 1:
            # find first non-None column offset
            col_offset = next((i for i, c in enumerate(row) if c is not None), 0)
            return row_idx, col_offset
    return None, 0


def _cell_to_str(cell) -> str:
    if cell is None:
        return ""
    if isinstance(cell, datetime):
        return cell.isoformat()
    if isinstance(cell, float) and cell == int(cell):
        return str(int(cell))
    return str(cell).strip()