-
Notifications
You must be signed in to change notification settings - Fork 1
050 ‐ A Complexidade do XML
No ecossistema fiscal brasileiro, a Nota Fiscal Eletrônica (NFe) representa muito mais do que um simples documento fiscal — é a espinha dorsal digital que conecta operações comerciais, obrigações tributárias e conformidade regulatória. Este artigo técnico explora a arquitetura inovadora do SPEDIR para transformar dados operacionais complexos em XMLs da NFe perfeitamente válidos, através de um sistema de mapeamento hierárquico que combina precisão matemática com flexibilidade operacional.
O Brasil possui um dos sistemas fiscais mais complexos do mundo, com mais de 90 códigos de situação tributária (CSTs e CSOSNs), múltiplos regimes tributários e regras específicas por unidade federativa. A NFe, com seu esquema XML de mais de 600 elementos possíveis, encapsula essa complexidade em uma estrutura rigidamente definida.
O desafio técnico é duplo:
- Completude: Suportar todos os cenários fiscais possíveis
- Simplicidade: Oferecer uma experiência intuitiva para usuários finais
- Precisão: Garantir conformidade absoluta com as especificações da SEFAZ
O SPEDIR adota uma abordagem baseada em mapeamento declarativo, onde a relação entre os dados do sistema e a estrutura XML é definida através de configuração, não de código hardcoded. Esta arquitetura oferece três vantagens fundamentais:
# Exemplo conceitual do mapeamento
MAPEAMENTO = {
"Emitente": {
"tag": "emit",
"mapeamentos": {
"numero_documento": "CNPJ",
"razao_social": "xNome",
# ... outros campos
}
}
}O sistema organiza-se em uma hierarquia lógica que espelha a estrutura do XML da NFe:
SPEDIR → XML NFe
├── Documento Principal (NFe)
│ ├── Identificação (ide)
│ ├── Emitente (emit)
│ ├── Destinatário (dest)
│ ├── Itens (det)
│ │ ├── Produto (prod)
│ │ └── Impostos (imposto)
│ ├── Totais (total)
│ └── Módulos Condicionais
flowchart TD
A[SPEDIR] --> B[XML NFe]
B --> C{Raiz Principal}
C --> D[NFe<br/>Documento Fiscal]
D --> E[ide<br/>Identificação]
D --> F[emit<br/>Emitente]
D --> G[dest<br/>Destinatário]
D --> H[det<br/>Itens]
D --> I[total<br/>Totais]
D --> J[transp<br/>Transporte]
D --> K[cobr<br/>Cobrança]
D --> L[infAdic<br/>Informações Adicionais]
%% Estrutura principal
F --> F1[Emitente Cadastrado]
F1 --> F2[Mapeamento Campos]
F2 --> F3["CNPJ → emit/CNPJ<br/>IE → emit/IE"]
G --> G1[Cliente/Fornecedor]
G1 --> G2[Mapeamento Campos]
G2 --> G3["CNPJ → dest/CNPJ<br/>email → dest/email"]
%% Itens - Estrutura Complexa
H --> H1[Produto/Serviço]
H1 --> H2[det/prod<br/>Dados do Produto]
H1 --> H3[det/imposto<br/>Tributação]
H2 --> H2A["cProd → código<br/>NCM → NCM"]
H3 --> H3A[ICMS]
H3 --> H3B[IPI]
H3 --> H3C[PIS]
H3 --> H3D[COFINS]
H3 --> H3E[ISSQN]
%% ICMS - CSTs
H3A --> H3A1[ICMS00<br/>Tributada Integralmente]
H3A --> H3A2[ICMS10<br/>Tributada com ST]
H3A --> H3A3[ICMS20<br/>Com Redução BC]
H3A --> H3A4[ICMS30<br/>Isenta/Não Tributada]
H3A --> H3A5[ICMS40<br/>Isenção]
H3A --> H3A6[ICMS51<br/>Diferimento]
H3A --> H3A7[ICMS60<br/>ST Retida]
H3A --> H3A8[ICMS70<br/>Com Redução + ST]
H3A --> H3A9[ICMS90<br/>Outros]
%% CSOSN - Simples Nacional
H3A --> H3A10[ICMSSN101<br/>Crédito SN]
H3A --> H3A11[ICMSSN201<br/>ST com Crédito]
H3A --> H3A12[ICMSSN500<br/>ICMS Efetivo]
H3A --> H3A13[ICMSSN900<br/>Outros SN]
%% Totais
I --> I1[ICMSTot<br/>Totais ICMS]
I --> I2[ISSQNTot<br/>Totais ISSQN]
I --> I3[retTrib<br/>Retenções]
I1 --> I1A["vBC → Base Cálculo<br/>vICMS → Valor ICMS"]
I2 --> I2A["vServ → Valor Serviços<br/>vISS → Valor ISS"]
%% Transporte
J --> J1[transporta<br/>Transportadora]
J --> J2[vol<br/>Volumes]
J --> J3[veicTransp<br/>Veículo]
J --> J4[reboque<br/>Reboques]
J2 --> J2A[lacres<br/>Lacres]
%% Cobrança
K --> K1[fat<br/>Fatura]
K --> K2[dup<br/>Duplicatas]
K --> K3[detPag<br/>Pagamentos]
%% Informações Adicionais
L --> L1[obsCont<br/>Observações]
L --> L2[procRef<br/>Processos Ref.]
%% Módulos Especiais
D --> M[exporta<br/>Exportação]
D --> N[compra<br/>Compras]
D --> O[cana<br/>Cana-de-Açúcar]
D --> P[infRespTec<br/>Resp. Técnico]
%% Estruturas Complexas nos Itens
H1 --> Q[DI<br/>Decl. Importação]
Q --> Q1[adi<br/>Adições DI]
H1 --> R[detExport<br/>Exportação]
R --> R1[exportInd<br/>Exp. Indireta]
H1 --> S[rastro<br/>Rastreamento]
%% Links do JSON
style C fill:#1e3a8a,color:white
style F1 fill:#3b82f6
style G1 fill:#3b82f6
style H1 fill:#10b981
style H3A fill:#f59e0b
style I1 fill:#ef4444
style J1 fill:#8b5cf6
style K1 fill:#ec4899
%% Legenda
subgraph Legend
L1[Modelo SPEDIR] --> L2[Tag XML]
L2 --> L3[Mapeamento de Campos]
end
2.1.1. Estrutura Principal (NFe)
SPEDIR.NFe.versao → infNFe/versao
SPEDIR.NFe.id → infNFe/id
SPEDIR.NFe.uf → infNFe/ide/cUF
SPEDIR.NFe.ide_numero → infNFe/ide/nNF
2.2.2. Emitente
SPEDIR.Emitente.numero_documento → emit/CNPJ
SPEDIR.Emitente.razao_social → emit/xNome
SPEDIR.Emitente.inscricao_estadual → emit/IE
SPEDIR.Emitente.endereco_cep → enderEmit/CEP
2.2.3. Destinatário
SPEDIR.Destinatario.numero_documento → dest/CNPJ|CPF
SPEDIR.Destinatario.email → dest/email
SPEDIR.Destinatario.indicador_ie → dest/indIEDest
2.2.4. Itens (det) - Hierarquia Complexa
<det nItem="1">
<prod>
<cProd>COD001</cProd> ← SPEDIR.Produto.codigo
<xProd>Produto Teste</xProd> ← SPEDIR.Produto.descricao
<NCM>12345678</NCM> ← SPEDIR.Produto.ncm
</prod>
<imposto>
<ICMS>
<ICMS00> ← Depende do CST
<CST>00</CST> ← SPEDIR.Produto.icms_cst
<vBC>100.00</vBC> ← SPEDIR.ICMS00.base_calculo
<pICMS>18.00</pICMS> ← SPEDIR.ICMS00.aliquota
</ICMS00>
</ICMS>
</imposto>
</det>2.2.5. ICMS por CST - Múltiplos Cenários
graph LR
A[CST/CSOSN] --> B{Mapeamento}
B --> C["CST '00' → ICMS00"]
B --> D["CST '10' → ICMS10<br/>ICMS com ST"]
B --> E["CST '20' → ICMS20<br/>Com Redução BC"]
B --> F["CST '30' → ICMS30<br/>Isenta"]
B --> G["CST '40' → ICMS40<br/>Isenção"]
B --> H["CST '51' → ICMS51<br/>Diferimento"]
B --> I["CST '60' → ICMS60<br/>ST Retida"]
B --> J["CST '70' → ICMS70<br/>Redução + ST"]
B --> K["CST '90' → ICMS90<br/>Outros"]
B --> L["CSOSN '101' → ICMSSN101"]
B --> M["CSOSN '201' → ICMSSN201"]
B --> N["CSOSN '500' → ICMSSN500"]
B --> O["CSOSN '900' → ICMSSN900"]
style C fill:#dbeafe
style D fill:#dbeafe
style L fill:#dcfce7
style M fill:#dcfce7
2.2.6. Totais - Agregação Automática
SPEDIR.NFe.total_icms_base_calculo → total/ICMSTot/vBC
SPEDIR.NFe.total_icms_icms_valor → total/ICMSTot/vICMS
SPEDIR.NFe.total_icms_st_valor → total/ICMSTot/vST
2.2.7. Transporte - Estrutura Opcional
SPEDIR.NFe.frete_modalidade → transp/modFrete
SPEDIR.Transportadora.numero_documento → transp/transporta/CNPJ
SPEDIR.NFe.veiculo_placa → transp/veicTransp/placa
2.2.8. Cobrança - Flexível
SPEDIR.NFe.cobr_fatura_nro → cob/fat/nFat
SPEDIR.NFe.cobr_fatura_valor → cob/fat/vLiq
SPEDIR.Duplicatas.nro → cob/dup/nDup
flowchart TD
XML[NFe.xml] --> InfNFe[infNFe]
InfNFe --> Ide[ide]
InfNFe --> Emit[emit]
InfNFe --> Dest[dest]
InfNFe --> Det[det]
InfNFe --> Total[total]
InfNFe --> Transp[transp]
InfNFe --> Cobr[cobr]
InfNFe --> InfAdic[infAdic]
%% Ide
Ide --> CUF[cUF]
Ide --> CNF[cNF]
Ide --> NatOp[natOp]
Ide --> Mod[mod]
Ide --> Serie[serie]
Ide --> NNF[nNF]
%% Det - Estrutura Aninhada
Det --> Prod[prod]
Det --> Imposto[imposto]
Prod --> CProd[cProd]
Prod --> XProd[xProd]
Prod --> NCM[NCM]
Prod --> CFOP[CFOP]
Imposto --> ICMS[ICMS]
Imposto --> IPI[IPI]
Imposto --> PIS[PIS]
Imposto --> COFINS[COFINS]
Imposto --> ISSQN[ISSQN]
ICMS --> ICMSXX[ICMS*<br/>Depende do CST]
%% Total
Total --> ICMSTot[ICMSTot]
Total --> ISSQNTot[ISSQNTot]
Total --> RetTrib[retTrib]
%% Módulos Condicionais
InfNFe --> Avulsa[avulsa]
InfNFe --> Exporta[exporta]
InfNFe --> Compra[compra]
InfNFe --> Cana[cana]
InfNFe --> InfRespTec[infRespTec]
sequenceDiagram
participant S as SPEDIR
participant M as Mapper
participant X as XML Generator
participant V as Validator
S->>M: Dados SPEDIR Completos
Note over M: Aplica mapeamento JSON
M->>X: Estrutura XML Mapeada
X->>X: Monta hierarquia XML
X->>X: Aplica formatação campos
X->>V: XML Gerado
V->>V: Valida contra XSD SEFAZ
V->>V: Valida regras de negócio
alt Validação OK
V->>S: ✅ XML Válido
S->>S: Assina digitalmente
else Erros
V->>S: ❌ Erros de validação
S->>S: Corrige automaticamente
end
####### 2.5.1. Mapeamento Condicional por CST
def map_icms_by_cst(cst: str, product_data: dict) -> str:
mapping = {
"00": "ICMS00",
"10": "ICMS10",
"20": "ICMS20",
"30": "ICMS30",
"40": "ICMS40",
"51": "ICMS51",
"60": "ICMS60",
"70": "ICMS70",
"90": "ICMS90",
"101": "ICMSSN101",
"201": "ICMSSN201",
"500": "ICMSSN500",
"900": "ICMSSN900"
}
return mapping.get(cst, "ICMS00") # Default####### 2.5.2. Campos Opcionais e Condicionais
-
transporta/CNPJsó existe setransp/modFrete != 9 -
dest/CPFsó existe sedest/indIEDest = 9(Consumidor Final) -
ICMSUFDestsó para operações interestaduais
####### 2.5.3. Valores Calculados vs. Informados
# Calculados automaticamente:
total/ICMSTot/vProd = Σ(det/prod/vProd)
total/ICMSTot/vICMS = Σ(det/imposto/ICMS/ICMS*/vICMS)
# Informados manualmente:
infNFe/ide/cNF = Código numérico aleatório
O coração da complexidade da NFe reside nos múltiplos cenários tributários. O SPEDIR implementa um sistema de mapeamento dinâmico que seleciona automaticamente a estrutura XML correta baseada no CST ou CSOSN:
class MapeadorICMS:
def obter_estrutura_xml(self, cst: str, regime: str) -> dict:
"""Retorna a estrutura XML específica para o CST/CSOSN"""
# Mapeamento CST → Tag XML
mapeamento_cst = {
"00": "ICMS00", # Tributada integralmente
"10": "ICMS10", # Tributada com ST
"20": "ICMS20", # Com redução de base
"30": "ICMS30", # Isenta/não tributada
"40": "ICMS40", # Isenção
"51": "ICMS51", # Diferimento
"60": "ICMS60", # ST retida anteriormente
"70": "ICMS70", # Com redução + ST
"90": "ICMS90", # Outros
}
# Mapeamento CSOSN → Tag XML (Simples Nacional)
mapeamento_csosn = {
"101": "ICMSSN101", # Tributada com crédito
"201": "ICMSSN201", # ST com crédito
"500": "ICMSSN500", # ICMS efetivo
"900": "ICMSSN900", # Outros
}
# Lógica de seleção baseada no regime
if regime == "SimplesNacional":
return mapeamento_csosn.get(cst, "ICMSSN900")
else:
return mapeamento_cst.get(cst, "ICMS00")O mapeamento não é uma simples tradução 1:1. Campos podem ser:
- Obrigatórios: Sempre presentes (ex: CNPJ do emitente)
- Condicionais: Dependem de outros campos (ex: CPF do destinatário só para consumidor final)
- Calculados: Derivados de outros valores (ex: totais da nota)
- Opcionais: Podem ser omitidos (ex: dados de reboque)
O fluxo de transformação segue um pipeline bem definido:
Entrada → Validação → Mapeamento → Montagem → Validação → Saída
↓ ↓ ↓ ↓ ↓ ↓
Dados Regras de Aplicação Construção XSD XML
SPEDIR Negócio Mapeamento Hierarquia SEFAZ Assinado
class PipelineNFe:
def processar(self, dados_spedir: dict) -> XMLValido:
# 1. Validação de regras de negócio
erros = self.validar_regras_negocio(dados_spedir)
if erros:
raise ValidacaoException(erros)
# 2. Aplicação do mapeamento hierárquico
estrutura_xml = self.aplicar_mapeamento(dados_spedir)
# 3. Montagem da hierarquia XML
xml_bruto = self.montar_hierarquia(estrutura_xml)
# 4. Validação contra XSD da SEFAZ
if not self.validar_xsd(xml_bruto):
raise XSDValidationException()
# 5. Cálculo e inclusão de valores derivados
xml_com_totais = self.calcular_totais(xml_bruto)
# 6. Geração de elementos automáticos
xml_final = self.gerar_elementos_automaticos(xml_com_totais)
return xml_finalA substituição tributária é um dos cenários mais complexos, envolvendo múltiplos cálculos e estruturas XML específicas:
class ProcessadorICMSST:
def processar(self, produto: dict) -> dict:
# Diferentes estruturas para diferentes CSTs com ST
estruturas = {
"10": self._estrutura_icms10, # Tributada com ST
"30": self._estrutura_icms30, # Isenta com ST
"70": self._estrutura_icms70, # Redução + ST
"90": self._estrutura_icms90, # Outros com ST
}
cst = produto.get("cst")
return estruturas.get(cst, self._estrutura_default)(produto)
def _estrutura_icms10(self, produto: dict) -> dict:
return {
"tag": "ICMS10",
"mapeamentos": {
"modBC": "icms_modalidade",
"vBC": "icms_base_calculo",
"pICMS": "icms_aliquota",
"vICMS": "icms_valor",
"modBCST": "icms_st_modalidade",
"pMVAST": "icms_st_mva",
"pRedBCST": "icms_st_reducao",
"vBCST": "icms_st_base_calculo",
"pICMSST": "icms_st_aliquota",
"vICMSST": "icms_st_valor",
}
}Para operações entre estados, o SPEDIR gerencia automaticamente o DIFAL e o FCP:
<!-- Exemplo de estrutura ICMSUFDest no XML -->
<ICMSUFDest>
<vBCUFDest>1000.00</vBCUFDest>
<vBCFCPUFDest>1000.00</vBCFCPUFDest>
<pICMSUFDest>18.00</pICMSUFDest>
<pICMSInter>12.00</pICMSInter>
<pICMSInterPart>40.00</pICMSInterPart>
<vFCPUFDest>0.00</vFCPUFDest>
<vICMSUFDest>60.00</vICMSUFDest>
<vICMSUFRemet>40.00</vICMSUFRemet>
</ICMSUFDest>O regime do Simples Nacional possui suas próprias estruturas XML (ICMSSN*), que o SPEDIR gerencia através de mapeamentos específicos:
MAAPEAMENTO_SIMPLES_NACIONAL = {
"101": {
"tag": "ICMSSN101",
"campos": ["orig", "CSOSN", "pCredSN", "vCredICMSSN"]
},
"201": {
"tag": "ICMSSN201",
"campos": ["modBCST", "pMVAST", "pRedBCST", "vBCST", "pICMSST", "vICMSST", "pCredSN", "vCredICMSSN"]
}
}O SPEDIR mantém compatibilidade com diferentes versões do layout da NFe através de perfis de mapeamento:
class GerenciadorVersoes:
def __init__(self):
self.versoes = {
"4.00": MapeamentoV4_00(),
"3.10": MapeamentoV3_10(),
"2.00": MapeamentoV2_00(),
}
def obter_mapeamento(self, versao: str) -> MapeamentoBase:
return self.versoes.get(versao, self.versoes["4.00"])Quando uma empresa precisa atualizar a versão do layout, o SPEDIR oferece migração automática dos mapeamentos:
class MigradorVersoes:
def migrar(self, dados_v_antiga: dict, versao_origem: str, versao_destino: str) -> dict:
# Converte campos depreciados para novos
# Mantém compatibilidade com alterações de estrutura
# Preserva dados importantes durante a transição
passPara operações em lote com milhares de notas, o SPEDIR implementa cache inteligente:
class CacheMapeamentos:
def __init__(self):
self.cache_estruturas = LRUCache(maxsize=1000)
self.cache_xml = LRUCache(maxsize=500)
def obter_estrutura(self, cst: str, regime: str) -> dict:
chave = f"{cst}_{regime}"
if chave not in self.cache_estruturas:
estrutura = self.calcular_estrutura(cst, regime)
self.cache_estruturas[chave] = estrutura
return self.cache_estruturas[chave]Para grandes volumes, o sistema utiliza processamento paralelo por item:
from concurrent.futures import ThreadPoolExecutor
class ProcessadorLote:
def processar_lote(self, itens: list) -> list:
with ThreadPoolExecutor(max_workers=8) as executor:
resultados = list(executor.map(self.processar_item, itens))
return resultadosCada transformação é registrada para auditoria e troubleshooting:
class LoggerTransformacao:
def registrar_transformacao(self, origem: dict, destino: dict, mapeamento: dict):
logger.info(f"Transformação realizada:", extra={
"origem": origem,
"destino": destino.keys(),
"mapeamento_usado": mapeamento["tag"],
"timestamp": datetime.now().isoformat()
})O sistema coleta métricas para otimização contínua:
metricas = {
"tempo_medio_mapeamento": 0.15, # segundos
"cache_hit_rate": 0.92, # 92% de acertos no cache
"erros_validacao_xsd": 0.001, # 0.1% de erros
}O SPEDIR implementa estratégias robustas para lidar com dados inesperados:
class TratadorCamposDesconhecidos:
def tratar(self, campo: str, valor: any, contexto: dict) -> Optional[dict]:
# 1. Tentar inferir mapeamento pelo nome do campo
# 2. Se for campo obrigatório desconhecido, erro
# 3. Se for campo opcional desconhecido, log warning
# 4. Se for possível calcular, calcular
passEm caso de falhas em mapeamentos específicos, o sistema oferece fallbacks:
class FallbackMapeamento:
def mapear_com_fallback(self, dados: dict) -> dict:
try:
return self.mapeamento_primario(dados)
except MapeamentoException:
logger.warning("Falha no mapeamento primário, usando fallback")
return self.mapeamento_fallback(dados)