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()