gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""Validated parser for the official Siscomex NCM JSON download."""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NcmEntry:
|
|
code: str
|
|
description: str
|
|
valid_from: str
|
|
valid_to: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NcmPayload:
|
|
version: str
|
|
act: str
|
|
sha256: str
|
|
entries: tuple[NcmEntry, ...]
|
|
|
|
|
|
def parse_ncm_payload(payload: bytes) -> NcmPayload:
|
|
try:
|
|
document = json.loads(payload)
|
|
except (TypeError, json.JSONDecodeError) as error:
|
|
raise ValueError("payload is not valid JSON") from error
|
|
if not isinstance(document, dict):
|
|
raise ValueError("payload root must be an object")
|
|
nomenclaturas = document.get("Nomenclaturas")
|
|
version = document.get("Data_Ultima_Atualizacao_NCM")
|
|
act = document.get("Ato")
|
|
if not isinstance(nomenclaturas, list) or not nomenclaturas:
|
|
raise ValueError("Nomenclaturas must be a nonempty list")
|
|
if not isinstance(version, str) or not isinstance(act, str):
|
|
raise ValueError("dataset metadata is invalid")
|
|
entries = []
|
|
for item in nomenclaturas:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("Nomenclaturas item must be an object")
|
|
values = tuple(item.get(name) for name in ("Codigo", "Descricao", "Data_Inicio", "Data_Fim"))
|
|
if not all(isinstance(value, str) for value in values):
|
|
raise ValueError("Nomenclaturas item schema is invalid")
|
|
code, description, valid_from, valid_to = values
|
|
entries.append(NcmEntry("".join(char for char in code if char.isdigit()), description, valid_from, valid_to))
|
|
return NcmPayload(version, act, hashlib.sha256(payload).hexdigest(), tuple(entries))
|