instant-bom/bom_assistant/ingestion/normalizer.py

from __future__ import annotations

import re
import uuid

from bom_assistant.classification.classifier import classify, is_non_electronic
from bom_assistant.classification.param_extractor import extract_params
from bom_assistant.session.models import BomRow, RowState

_DNP_KEYWORDS = re.compile(
    r"\b(dnp|do\s+not\s+place|do\s+not\s+populate|np|not\s+placed)\b",
    re.IGNORECASE,
)
_DNP_TRUTHY = {"1", "true", "yes", "dnp", "x"}

# Range designator: "C1-5", "R56-61"
_RANGE_RE = re.compile(r"^([A-Za-z]+)(\d+)-(\d+)$")


def normalize(mapped_rows: list[dict[str, str]]) -> list[BomRow]:
    rows: list[BomRow] = []
    for raw in mapped_rows:
        row = _normalize_row(raw)
        if row is not None:
            rows.append(row)
    return rows


def _normalize_row(raw: dict[str, str]) -> BomRow | None:
    value = raw.get("value", "").strip()

    # skip empty or KiCad placeholder
    if not value or value == "~":
        return None

    designators = _parse_designators(raw.get("designators", ""))
    quantity = _parse_quantity(raw.get("quantity", ""))
    footprint = raw.get("footprint", "").strip() or None

    # pull part_number / manufacturer from mapped columns if present
    part_number = raw.get("part_number", "").strip() or None
    manufacturer = raw.get("manufacturer", "").strip() or None

    # state determination
    state = _determine_state(value, footprint, quantity, raw)

    category = classify(designators, value)

    params = extract_params(value, footprint, category)

    # inject part_number / manufacturer from mapped columns if extractor didn't find them
    if part_number and not params.part_number:
        params.part_number = part_number
    if manufacturer and not params.manufacturer:
        params.manufacturer = manufacturer

    return BomRow(
        row_id=str(uuid.uuid4()),
        designators=designators,
        quantity=quantity,
        raw_value=value,
        footprint=footprint,
        category=category,
        normalized_params=params,
        state=state,
    )


def _parse_designators(s: str) -> list[str]:
    if not s:
        return []
    # split on comma, semicolon, or whitespace
    parts = re.split(r"[,;\s]+", s.strip())
    expanded: list[str] = []
    for part in parts:
        part = part.strip()
        if not part:
            continue
        expanded.extend(_expand_range(part))
    # deduplicate while preserving order
    seen: set[str] = set()
    result: list[str] = []
    for d in expanded:
        if d not in seen:
            seen.add(d)
            result.append(d)
    return sorted(result, key=lambda x: (re.sub(r"\d", "", x), int(re.sub(r"\D", "", x) or 0)))


def _expand_range(s: str) -> list[str]:
    m = _RANGE_RE.match(s)
    if m:
        prefix = m.group(1)
        start, end = int(m.group(2)), int(m.group(3))
        if start <= end and (end - start) < 200:  # sanity cap
            return [f"{prefix}{i}" for i in range(start, end + 1)]
    return [s]


def _parse_quantity(s: str) -> int:
    s = s.strip()
    if not s:
        return 1
    try:
        return max(0, int(float(s)))
    except (ValueError, TypeError):
        return 1


def _determine_state(
    value: str,
    footprint: str | None,
    quantity: int,
    raw: dict[str, str],
) -> RowState:
    if quantity == 0:
        return RowState.flagged

    combined = f"{value} {footprint or ''}"
    if _DNP_KEYWORDS.search(combined):
        return RowState.flagged

    # check explicit boolean DNP columns
    for col in ("DNP", "Exclude from BOM", "Mounting", "_raw_DNP", "_raw_Exclude from BOM", "_raw_Mounting"):
        col_val = raw.get(col, "").strip().lower()
        if col_val in _DNP_TRUTHY:
            return RowState.flagged

    if is_non_electronic(value):
        return RowState.flagged

    return RowState.pending