55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""Fail-closed parser for the official Receita Federal TIPI workbook."""
|
|
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
import hashlib
|
|
import unicodedata
|
|
|
|
import openpyxl
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TipiEntry:
|
|
ncm: str
|
|
ex: str | None
|
|
description: str
|
|
rate: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TipiPayload:
|
|
sha256: str
|
|
entries: tuple[TipiEntry, ...]
|
|
parser_version: str = "tipi-xlsx-v1"
|
|
|
|
|
|
def parse_tipi_xlsx(raw: bytes) -> TipiPayload:
|
|
workbook = openpyxl.load_workbook(BytesIO(raw), read_only=True, data_only=True)
|
|
sheet = workbook.active
|
|
normalize = lambda value: "".join(char for char in unicodedata.normalize("NFKD", str(value or "")).encode("ascii", "ignore").decode().upper() if char.isalnum())
|
|
header_row = next((row for row in sheet.iter_rows(values_only=True) if {"NCM", "DESCRICAO", "ALIQUOTA"} <= {normalize(cell) for cell in row}), None)
|
|
if header_row is None:
|
|
raise ValueError("TIPI workbook has no recognized official header")
|
|
headers = [normalize(cell) for cell in header_row]
|
|
ncm_index = headers.index("NCM")
|
|
ex_index = headers.index("EX")
|
|
description_index = next(index for index, value in enumerate(headers) if value == "DESCRICAO")
|
|
rate_index = next(index for index, value in enumerate(headers) if value.startswith("ALIQUOTA"))
|
|
entries = []
|
|
started = False
|
|
for row in sheet.iter_rows(values_only=True):
|
|
if tuple(row) == tuple(header_row):
|
|
started = True; continue
|
|
if not started or ncm_index >= len(row):
|
|
continue
|
|
ncm = "".join(char for char in str(row[ncm_index] or "") if char.isdigit())
|
|
if len(ncm) != 8:
|
|
continue
|
|
description = str(row[description_index] or "").strip()
|
|
if not description:
|
|
raise ValueError("TIPI NCM row has no description")
|
|
ex = str(row[ex_index] or "").strip() or None
|
|
entries.append(TipiEntry(ncm, ex, description, str(row[rate_index] or "").strip()))
|
|
if not entries:
|
|
raise ValueError("TIPI workbook has no 8-digit NCM rows")
|
|
return TipiPayload(hashlib.sha256(raw).hexdigest(), tuple(entries))
|