Add cutting workbook normalizer

2026-07-07 14:41:35 -03:00
parent c881dd7dde
commit fedb1c88a8
2 changed files with 534 additions and 0 deletions
--- a/backend/scripts/normalize_cut_workbook.py
+++ b/backend/scripts/normalize_cut_workbook.py
@@ -0,0 +1,519 @@
+#!/usr/bin/env python3
+"""Normalize Nexstar cutting/replenishment workbooks into JSON.
+
+The workbook is used as an operational planning model. This importer treats it
+as an input source and emits auditable tables that the app can later persist or
+render, without depending on the workbook formulas at runtime.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import posixpath
+import re
+import sys
+import zipfile
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any, Dict, Iterable, List, Optional, Tuple
+from xml.etree import ElementTree as ET
+
+
+NS = {
+    "main": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
+    "rel": "http://schemas.openxmlformats.org/officeDocument/2006/relationships",
+    "pkgrel": "http://schemas.openxmlformats.org/package/2006/relationships",
+}
+
+DATE_NUM_FMT_IDS = {
+    14,
+    15,
+    16,
+    17,
+    22,
+    27,
+    30,
+    36,
+    45,
+    46,
+    47,
+    50,
+    57,
+}
+
+CUT_SHEETS = {"BLCS", "BLMC", "BLOS", "BLPM"}
+COLOR_WORDS = {
+    "amarelo",
+    "azul",
+    "azul pr.",
+    "azul prussia",
+    "bege",
+    "bordo",
+    "bordô",
+    "branco",
+    "cafe",
+    "café",
+    "cinza",
+    "cor??",
+    "grafite",
+    "marinho",
+    "marrom",
+    "perola",
+    "pérola",
+    "preto",
+    "rosa",
+    "rox0",
+    "roxo",
+    "royal",
+    "verde b.",
+    "verde bandeira",
+    "verde militar",
+    "vermelho",
+}
+
+
+@dataclass(frozen=True)
+class Cell:
+    value: Any
+    formula: Optional[str] = None
+    has_broken_reference: bool = False
+
+
+def local_path(base: str, target: str) -> str:
+    if target.startswith("/"):
+        return target.lstrip("/")
+    return posixpath.normpath(posixpath.join(posixpath.dirname(base), target))
+
+
+def xml_root(archive: zipfile.ZipFile, path: str) -> ET.Element:
+    with archive.open(path) as handle:
+        return ET.parse(handle).getroot()
+
+
+def parse_relationships(archive: zipfile.ZipFile, path: str) -> Dict[str, str]:
+    if path not in archive.namelist():
+        return {}
+
+    root = xml_root(archive, path)
+    relationships: Dict[str, str] = {}
+    for rel in root:
+        rel_id = rel.attrib.get("Id")
+        target = rel.attrib.get("Target")
+        if rel_id and target:
+            relationships[rel_id] = target
+    return relationships
+
+
+def load_shared_strings(archive: zipfile.ZipFile) -> List[str]:
+    if "xl/sharedStrings.xml" not in archive.namelist():
+        return []
+
+    root = xml_root(archive, "xl/sharedStrings.xml")
+    strings: List[str] = []
+    for item in root.findall("main:si", NS):
+        parts = [node.text or "" for node in item.findall(".//main:t", NS)]
+        strings.append("".join(parts))
+    return strings
+
+
+def load_date_style_ids(archive: zipfile.ZipFile) -> set[int]:
+    if "xl/styles.xml" not in archive.namelist():
+        return set()
+
+    root = xml_root(archive, "xl/styles.xml")
+    date_num_fmt_ids = set(DATE_NUM_FMT_IDS)
+    for fmt in root.findall("main:numFmts/main:numFmt", NS):
+        num_fmt_id = fmt.attrib.get("numFmtId")
+        format_code = fmt.attrib.get("formatCode", "").lower()
+        if not num_fmt_id:
+            continue
+        if re.search(r"(^|[^\\[])[dmy]([^\\]]|$)", format_code):
+            date_num_fmt_ids.add(int(num_fmt_id))
+
+    style_ids: set[int] = set()
+    for index, cell_format in enumerate(root.findall("main:cellXfs/main:xf", NS)):
+        num_fmt_id = int(cell_format.attrib.get("numFmtId", "0"))
+        if num_fmt_id in date_num_fmt_ids:
+            style_ids.add(index)
+    return style_ids
+
+
+def excel_date(value: float) -> str:
+    # Excel's 1900 date system includes the leap-year bug; this origin mirrors it
+    # for normal post-1900 workbook dates.
+    date = datetime(1899, 12, 30) + timedelta(days=value)
+    return date.date().isoformat()
+
+
+def parse_scalar(value: Optional[str], style_id: Optional[int], cell_type: Optional[str], shared_strings: List[str], date_style_ids: set[int]) -> Any:
+    if value is None:
+        return None
+
+    if cell_type == "s":
+        try:
+            return shared_strings[int(value)]
+        except (ValueError, IndexError):
+            return value
+
+    if cell_type == "b":
+        return value == "1"
+
+    if cell_type in {"str", "inlineStr"}:
+        return value
+
+    try:
+        number = float(value)
+    except ValueError:
+        return value
+
+    if style_id in date_style_ids:
+        return excel_date(number)
+
+    if number.is_integer():
+        return int(number)
+    return number
+
+
+def cell_column(cell_ref: str) -> int:
+    letters = re.sub(r"[^A-Z]", "", cell_ref.upper())
+    column = 0
+    for letter in letters:
+        column = column * 26 + (ord(letter) - ord("A") + 1)
+    return max(1, column)
+
+
+def parse_inline_string(cell_node: ET.Element) -> Optional[str]:
+    inline = cell_node.find("main:is", NS)
+    if inline is None:
+        return None
+    return "".join(node.text or "" for node in inline.findall(".//main:t", NS))
+
+
+def parse_sheet_rows(
+    archive: zipfile.ZipFile,
+    sheet_path: str,
+    shared_strings: List[str],
+    date_style_ids: set[int],
+) -> List[List[Cell]]:
+    root = xml_root(archive, sheet_path)
+    rows: List[List[Cell]] = []
+
+    for row in root.findall(".//main:sheetData/main:row", NS):
+        parsed: List[Cell] = []
+        last_column = 0
+        for cell_node in row.findall("main:c", NS):
+            ref = cell_node.attrib.get("r", "")
+            column = cell_column(ref) if ref else last_column + 1
+            while len(parsed) < column - 1:
+                parsed.append(Cell(None))
+
+            cell_type = cell_node.attrib.get("t")
+            style_id = int(cell_node.attrib.get("s", "-1"))
+            formula_node = cell_node.find("main:f", NS)
+            value_node = cell_node.find("main:v", NS)
+            formula = formula_node.text if formula_node is not None else None
+
+            if cell_type == "inlineStr":
+                value = parse_inline_string(cell_node)
+            else:
+                value = parse_scalar(
+                    value_node.text if value_node is not None else None,
+                    style_id,
+                    cell_type,
+                    shared_strings,
+                    date_style_ids,
+                )
+
+            parsed.append(Cell(value=value, formula=formula, has_broken_reference="#REF!" in (formula or "")))
+            last_column = column
+
+        while parsed and parsed[-1].value is None and not parsed[-1].formula:
+            parsed.pop()
+        if parsed:
+            rows.append(parsed)
+
+    return rows
+
+
+def workbook_rows(path: str) -> Dict[str, List[List[Cell]]]:
+    with zipfile.ZipFile(path) as archive:
+        workbook_path = "xl/workbook.xml"
+        workbook = xml_root(archive, workbook_path)
+        rels = parse_relationships(archive, "xl/_rels/workbook.xml.rels")
+        shared_strings = load_shared_strings(archive)
+        date_style_ids = load_date_style_ids(archive)
+
+        rows_by_sheet: Dict[str, List[List[Cell]]] = {}
+        for sheet in workbook.findall("main:sheets/main:sheet", NS):
+            name = sheet.attrib["name"]
+            rel_id = sheet.attrib.get(f"{{{NS['rel']}}}id")
+            target = rels.get(rel_id or "")
+            if not target:
+                continue
+            sheet_path = local_path(workbook_path, target)
+            rows_by_sheet[name] = parse_sheet_rows(archive, sheet_path, shared_strings, date_style_ids)
+
+        return rows_by_sheet
+
+
+def clean_text(value: Any) -> str:
+    return re.sub(r"\s+", " ", str(value or "")).strip()
+
+
+def number_or_zero(value: Any) -> float:
+    if value in (None, ""):
+        return 0
+    if isinstance(value, (int, float)):
+        return float(value)
+    normalized = str(value).strip().replace(".", "").replace(",", ".")
+    try:
+        return float(normalized)
+    except ValueError:
+        return 0
+
+
+def int_if_whole(value: float) -> int | float:
+    return int(value) if float(value).is_integer() else value
+
+
+def row_values(row: List[Cell]) -> List[Any]:
+    return [cell.value for cell in row]
+
+
+def get(row: List[Any], index: int) -> Any:
+    return row[index] if index < len(row) else None
+
+
+def non_empty_rows(rows: Iterable[List[Cell]]) -> Iterable[List[Any]]:
+    for row in rows:
+        values = row_values(row)
+        if any(value not in (None, "") for value in values):
+            yield values
+
+
+def normalize_purchase_need(rows: List[List[Cell]]) -> List[Dict[str, Any]]:
+    output: List[Dict[str, Any]] = []
+    for values in list(non_empty_rows(rows))[1:]:
+        product = clean_text(get(values, 0))
+        sku = clean_text(get(values, 1))
+        if not product:
+            continue
+        output.append(
+            {
+                "product": product,
+                "sku": sku,
+                "supplier": clean_text(get(values, 2)),
+                "unit": clean_text(get(values, 3)),
+                "averageCost": number_or_zero(get(values, 4)),
+                "expectedIn": number_or_zero(get(values, 5)),
+                "expectedOut": number_or_zero(get(values, 6)),
+                "physicalStock": number_or_zero(get(values, 7)),
+                "virtualStock": number_or_zero(get(values, 8)),
+                "periodOut": number_or_zero(get(values, 9)),
+                "monthlyAverage": number_or_zero(get(values, 10)),
+                "coverageMonths": number_or_zero(get(values, 11)),
+                "suggestedPurchase": number_or_zero(get(values, 12)),
+            }
+        )
+    return output
+
+
+def normalize_production_orders(rows: List[List[Cell]]) -> List[Dict[str, Any]]:
+    output: List[Dict[str, Any]] = []
+    for values in list(non_empty_rows(rows))[1:]:
+        product = clean_text(get(values, 4))
+        quantity = number_or_zero(get(values, 6))
+        if not product and not quantity:
+            continue
+        output.append(
+            {
+                "number": clean_text(get(values, 0)),
+                "orderReference": clean_text(get(values, 1)),
+                "startDate": get(values, 2),
+                "expectedDate": get(values, 3),
+                "product": product,
+                "status": clean_text(get(values, 5)),
+                "quantity": int_if_whole(quantity),
+            }
+        )
+    return output
+
+
+def normalize_product_mapping(rows: List[List[Cell]]) -> List[Dict[str, Any]]:
+    output: List[Dict[str, Any]] = []
+    for values in list(non_empty_rows(rows))[1:]:
+        product = clean_text(get(values, 4))
+        if not product:
+            continue
+        output.append(
+            {
+                "product": product,
+                "purchaseNeedProduct": clean_text(get(values, 0)),
+                "physicalStock": number_or_zero(get(values, 1)),
+                "productionOrderProduct": clean_text(get(values, 2)),
+                "productionOrderQuantity": number_or_zero(get(values, 3)),
+                "totalProductionOrder": number_or_zero(get(values, 5)),
+                "stockPlusProductionOrder": number_or_zero(get(values, 6)),
+                "finishedStock": number_or_zero(get(values, 7)),
+            }
+        )
+    return output
+
+
+def normalize_two_column_quantity(rows: List[List[Cell]], quantity_name: str) -> List[Dict[str, Any]]:
+    output: List[Dict[str, Any]] = []
+    for values in non_empty_rows(rows):
+        product = clean_text(get(values, 0))
+        if not product:
+            continue
+        output.append({"product": product, quantity_name: number_or_zero(get(values, 1))})
+    return output
+
+
+def normalize_matrix_sheet(rows: List[List[Cell]]) -> Dict[str, Any]:
+    values = list(non_empty_rows(rows))
+    title = clean_text(get(values[0], 0)) if values else ""
+    total = number_or_zero(get(values[0], 2) if len(values[0]) > 2 else get(values[0], 1)) if values else 0
+    return {
+        "title": title,
+        "total": int_if_whole(total),
+        "rows": [row[:40] for row in values[:120]],
+    }
+
+
+def is_color_label(value: str) -> bool:
+    normalized = value.strip().lower()
+    if not normalized:
+        return False
+    return normalized in COLOR_WORDS
+
+
+def extract_color_labels(rows: List[List[Any]], section_index: int) -> List[str]:
+    candidates: List[Tuple[int, List[str]]] = []
+    for nearby in rows[section_index + 1 : section_index + 8]:
+        labels = [clean_text(value) for value in nearby]
+        colors = [label for label in labels if is_color_label(label)]
+        if len(colors) >= 2:
+            candidates.append((len(colors), colors))
+
+    if not candidates:
+        return []
+
+    return max(candidates, key=lambda item: item[0])[1][:24]
+
+
+def first_number(values: List[Any]) -> float:
+    for value in values:
+        number = number_or_zero(value)
+        if number:
+            return number
+    return 0.0
+
+
+def normalize_cut_sheet(name: str, rows: List[List[Cell]]) -> Dict[str, Any]:
+    values = list(non_empty_rows(rows))
+    sections: List[Dict[str, Any]] = []
+    broken_references = sum(1 for row in rows for cell in row if cell.has_broken_reference)
+
+    for index, row in enumerate(values):
+        labels = [clean_text(value) for value in row]
+        if not any(label.lower() == "total de rolos" for label in labels):
+            continue
+
+        title = next((label for label in labels if "corte" in label.lower()), "")
+        total_rolls = first_number(row)
+        if not total_rolls and index + 1 < len(values):
+            next_row = values[index + 1]
+            next_labels = [clean_text(value) for value in next_row]
+            if any(label.lower() == "total de rolos" for label in next_labels):
+                total_rolls = first_number(next_row)
+
+        total_need = 0.0
+        for position, label in enumerate(labels):
+            if label.lower().startswith("total necc"):
+                total_need = number_or_zero(get(row, position + 1))
+                break
+
+        color_row = extract_color_labels(values, index)
+
+        if title:
+            sections.append(
+                {
+                    "family": name,
+                    "title": title,
+                    "totalRolls": int_if_whole(total_rolls),
+                    "totalNeed": int_if_whole(total_need),
+                    "colors": color_row[:24],
+                }
+            )
+
+    return {
+        "sheet": name,
+        "brokenFormulaReferences": broken_references,
+        "sections": sections,
+    }
+
+
+def normalize_workbook(path: str) -> Dict[str, Any]:
+    sheets = workbook_rows(path)
+    normalized = {
+        "source": path,
+        "generatedAt": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
+        "summary": {
+            "sheetCount": len(sheets),
+            "sheets": list(sheets.keys()),
+        },
+        "purchaseNeed": normalize_purchase_need(sheets.get("NECESSIDADE DE COMPRA (colar)", [])),
+        "productMappings": normalize_product_mapping(sheets.get("INSERIR NOVOS PRODUTOS", [])),
+        "productionOrders": normalize_production_orders(sheets.get("ORDEM DE PRODUÇÃO (colar)", [])),
+        "realPurchaseNeed": normalize_two_column_quantity(sheets.get("NECESSIDADE REAL DE COMPRA", []), "realNeed"),
+        "outsideItems": normalize_two_column_quantity(sheets.get("ITENS POR FORA", []), "quantity"),
+        "finishedStockMatrix": normalize_matrix_sheet(sheets.get("ESTOQUE PA", [])),
+        "stockPlusProductionOrderMatrix": normalize_matrix_sheet(sheets.get("ESTOQUE PA+ ORDEM DE PRODUÇÃO", [])),
+        "cutPlans": [
+            normalize_cut_sheet(sheet_name, rows)
+            for sheet_name, rows in sheets.items()
+            if sheet_name in CUT_SHEETS
+        ],
+    }
+
+    normalized["summary"].update(
+        {
+            "purchaseNeedRows": len(normalized["purchaseNeed"]),
+            "productMappingRows": len(normalized["productMappings"]),
+            "productionOrderRows": len(normalized["productionOrders"]),
+            "realPurchaseNeedRows": len(normalized["realPurchaseNeed"]),
+            "outsideItemRows": len(normalized["outsideItems"]),
+            "cutPlanSections": sum(len(plan["sections"]) for plan in normalized["cutPlans"]),
+            "brokenFormulaReferences": sum(plan["brokenFormulaReferences"] for plan in normalized["cutPlans"]),
+        }
+    )
+    return normalized
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Normalize NECESSIDADE DE CORTE workbooks into JSON.")
+    parser.add_argument("workbook", help="Path to the .xlsx workbook")
+    parser.add_argument("-o", "--output", help="Optional output JSON file. Defaults to stdout.")
+    parser.add_argument("--pretty", action="store_true", help="Pretty-print JSON")
+    args = parser.parse_args()
+
+    try:
+        payload = normalize_workbook(args.workbook)
+    except Exception as error:  # noqa: BLE001 - CLI should report any parse failure cleanly.
+        print(f"Failed to normalize workbook: {error}", file=sys.stderr)
+        return 1
+
+    text = json.dumps(payload, ensure_ascii=False, indent=2 if args.pretty else None)
+    if args.output:
+        with open(args.output, "w", encoding="utf-8") as handle:
+            handle.write(text)
+            handle.write("\n")
+    else:
+        print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())