From fedb1c88a889561e8a86ebe69baa942ae9f54dd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cau=C3=AA=20Faleiros?= Date: Tue, 7 Jul 2026 14:41:35 -0300 Subject: [PATCH] Add cutting workbook normalizer --- README.md | 15 + backend/scripts/normalize_cut_workbook.py | 519 ++++++++++++++++++++++ 2 files changed, 534 insertions(+) create mode 100644 backend/scripts/normalize_cut_workbook.py diff --git a/README.md b/README.md index b7b533d..f0d3e05 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,21 @@ The scheduled endpoint is API-key protected and returns a summary: } ``` +### Cutting Workbook Import + +Manual cutting/replenishment workbooks can be normalized into JSON before they +are persisted or shown in the app: + +```bash +python3 backend/scripts/normalize_cut_workbook.py "/path/to/NECESSIDADE DE CORTE.xlsx" --pretty -o /tmp/necessidade_corte_normalized.json +``` + +The output includes purchase need rows, production orders, finished stock, +stock plus OP availability, real purchase need, outside items, and cut-plan +sections by family (`BLCS`, `BLMC`, `BLOS`, `BLPM`). The importer reads cached +workbook values and reports broken formula references instead of making the app +depend on Excel formulas at runtime. + ## Local Development Start PostgreSQL: diff --git a/backend/scripts/normalize_cut_workbook.py b/backend/scripts/normalize_cut_workbook.py new file mode 100644 index 0000000..7b37139 --- /dev/null +++ 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())