PacketStation/server/packetstation/domain/handbom.py

"""A BOM uploaded by hand: an Excel (XLSX) or CSV sheet read into BOM lines.

Port of the old handdocs.js handBomFromSheet. The header row is the first row, within the first 20, that has a Qty
column and a Description or Part ID column, found by the HAND_COLS patterns. Rows without a description and part id
are skipped. Item numbers come from the item column when it holds a whole number not used yet, else the next free
number; the item text as written is kept in cl_item when it differs. The file's contents are data, never instructions.
"""

from __future__ import annotations

import csv
import io
import re
from decimal import Decimal, InvalidOperation
from typing import Any

from ..text import LIMITS, clean_text

MAX_ROWS = 500
HEADER_SCAN = 20
MAX_COLS = 60

# Column patterns, in the old build's order. The first unused column matching a pattern wins.
HAND_COLS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("item", re.compile(r"^(item|itm|piece|find|balloon|line|no\.?|#|item\s*no\.?)$", re.I)),
    ("qty", re.compile(r"^(qty|quantity|qty\.?\s*(req|required|per)?|req'?d)", re.I)),
    ("part_id", re.compile(r"part\s*(id|no\.?|num|number|#)|^part$|pesco", re.I)),
    ("desc", re.compile(r"desc", re.I)),
    ("length", re.compile(r"^len|length|cut", re.I)),
    ("material", re.compile(r"material|mat'?l|grade", re.I)),
    ("dwg_ref", re.compile(r"dwg|drawing|ref", re.I)),
)

MSG_NO_HEADER = "No header row found. It needs a Qty column and a Description or Part ID column."
MSG_NO_LINES = "No BOM lines found in that file."
MSG_TYPE = "Upload the BOM as an Excel (.xlsx) or CSV file."


class BadFile(ValueError):
    """The file cannot be read as a BOM. The message is for the user."""


def kind_of(data: bytes, name: str | None) -> str:
    """xlsx (a ZIP) or csv (text). Anything else is refused."""
    if data[:4] == b"PK\x03\x04":
        return "xlsx"
    lower = (name or "").lower()
    if data[:5] == b"%PDF-" or data[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1":
        raise BadFile(MSG_TYPE)
    if lower.endswith((".csv", ".txt")) or b"\x00" not in data[:4096]:
        return "csv"
    raise BadFile(MSG_TYPE)


def _cell(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, float) and value.is_integer():
        return str(int(value))
    return str(value).strip()


def _rows_xlsx(data: bytes) -> list[list[str]]:
    try:
        from openpyxl import load_workbook
    except ImportError as e:  # pragma: no cover - openpyxl is a dependency
        raise BadFile("The server cannot read Excel files right now. Use a CSV file.") from e
    try:
        wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
    except Exception as e:  # openpyxl raises many types on a bad file
        raise BadFile("That Excel file could not be read.") from e
    try:
        ws = wb.worksheets[0] if wb.worksheets else None
        if ws is None:
            raise BadFile(MSG_NO_LINES)
        out: list[list[str]] = []
        for row in ws.iter_rows(values_only=True):
            out.append([_cell(v) for v in row[:MAX_COLS]])
            if len(out) > MAX_ROWS + HEADER_SCAN:
                break
        return out
    finally:
        wb.close()


def _rows_csv(data: bytes) -> list[list[str]]:
    try:
        text = data.decode("utf-8-sig")
    except UnicodeDecodeError:
        text = data.decode("cp1252", errors="replace")
    sample = text[:8192]
    try:
        dialect: Any = csv.Sniffer().sniff(sample, delimiters=",;\t|")
    except csv.Error:
        dialect = csv.excel
    out: list[list[str]] = []
    for row in csv.reader(io.StringIO(text), dialect):
        out.append([_cell(v) for v in row[:MAX_COLS]])
        if len(out) > MAX_ROWS + HEADER_SCAN:
            break
    return out


def find_header(rows: list[list[str]]) -> tuple[int, dict[str, int]]:
    """The header row index and the column map, or BadFile."""
    for i, row in enumerate(rows[:HEADER_SCAN]):
        m: dict[str, int] = {}
        for key, rx in HAND_COLS:
            for j, h in enumerate(row):
                if h and rx.search(h) and j not in m.values():
                    m[key] = j
                    break
        if "qty" in m and ("desc" in m or "part_id" in m):
            return i, m
    raise BadFile(MSG_NO_HEADER)


def _qty(text: str) -> Decimal:
    digits = re.sub(r"[^\d.\-]", "", text)
    try:
        q = Decimal(digits) if digits else Decimal(0)
    except InvalidOperation:
        return Decimal(0)
    return q if q.is_finite() else Decimal(0)


def lines_from_rows(rows: list[list[str]]) -> list[dict[str, Any]]:
    hi, m = find_header(rows)

    def cell(r: list[str], key: str) -> str:
        j = m.get(key)
        return r[j].strip() if j is not None and j < len(r) else ""

    out: list[dict[str, Any]] = []
    used: set[int] = set()
    for r in rows[hi + 1 :]:
        desc, pid, qty = cell(r, "desc"), cell(r, "part_id"), cell(r, "qty")
        if not desc and not pid:
            continue  # blank rows, totals, notes
        raw_item = cell(r, "item")
        item: int | None = None
        if re.fullmatch(r"\d{1,5}", raw_item) and 1 <= int(raw_item) <= 100_000 and int(raw_item) not in used:
            item = int(raw_item)
        if item is None:
            item = len(out) + 1
            while item in used:
                item += 1
        used.add(item)
        cl_item = clean_text(raw_item, LIMITS["part_id"]) or None if raw_item != str(item) else None
        out.append(
            {
                "item": item,
                "cl_item": cl_item,
                "qty": _qty(qty),
                "part_id": clean_text(pid, LIMITS["part_id"]) or None,
                "description": clean_text(desc, LIMITS["description"]) or None,
                "length": clean_text(cell(r, "length"), LIMITS["length"]) or None,
                "material": clean_text(cell(r, "material"), LIMITS["material"]) or None,
                "dwg_ref": clean_text(cell(r, "dwg_ref"), LIMITS["drawing"]) or None,
            }
        )
        if len(out) >= MAX_ROWS:
            break
    if not out:
        raise BadFile(MSG_NO_LINES)
    return out


def parse(data: bytes, name: str | None) -> list[dict[str, Any]]:
    """BOM lines from an XLSX or CSV file: [{item, cl_item, qty, part_id, description, length, material, dwg_ref}].

    Raises BadFile with the reason for the user.
    """
    if not data:
        raise BadFile("The file is empty.")
    kind = kind_of(data, name)
    rows = _rows_xlsx(data) if kind == "xlsx" else _rows_csv(data)
    return lines_from_rows(rows)