feat: extract pure fiscal core from sowai-auto (parity-preserving copy)
Copies the 6 pure fiscal modules (domains, chave_acesso, tpag, presets, resolver, xml_builder) from sowai-auto's backend/app/modules/fiscal/ into this standalone lib, with only the two mechanical changes required to drop the app.* dependency: import prefix rename, and TaxRule (ORM) -> TaxRuleLike (structural Protocol) in the resolver. No other line changes -- byte-level parity with the auto is the point. Ports the pure test suites (domains, chave_acesso, resolver with a local FakeTaxRule satisfying TaxRuleLike, xml_builder) plus a new test_presets_data covering the PRESETS dict directly (the auto's equivalent test exercises the apply-preset HTTP/DB flow, which isn't part of the extracted pure core). 55 tests pass locally via `uv run pytest`, no Postgres/k8s required.
This commit is contained in:
jonatanritter
@@ -0,0 +1,2 @@
|
||||
"""sowai_fiscal — núcleo fiscal puro da SowAI (motor de regras, chave de
|
||||
acesso, builder de XML NF-e). Ver README.md para o mapa dos módulos."""
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Chave de acesso da NF-e (funções PURAS -- zero I/O, zero sessão).
|
||||
|
||||
Estrutura (44 dígitos): cUF(2) + AAMM(4) + CNPJ(14) + mod(2) + série(3) +
|
||||
nNF(9) + tpEmis(1) + cNF(8) + DV(1).
|
||||
|
||||
cNF é sorteado UMA vez por documento e PERSISTE (`FiscalDocument.
|
||||
codigo_numerico`, Task 4): retry de transmissão usa a MESMA chave --
|
||||
re-sortear cNF é o que fabrica a Rejeição 539 (duplicidade com chave
|
||||
diferente). Regra NT 2019.001: cNF != nNF.
|
||||
"""
|
||||
import secrets
|
||||
|
||||
_PESOS = (2, 3, 4, 5, 6, 7, 8, 9)
|
||||
|
||||
|
||||
def dv_modulo11(chave43: str) -> str:
|
||||
"""Dígito verificador módulo 11 da chave de acesso.
|
||||
|
||||
Pesos 2..9 aplicados da direita para a esquerda (repetindo em ciclos de
|
||||
8); soma dos produtos; resto = soma % 11; DV = 0 se resto in (0, 1),
|
||||
senão DV = 11 - resto. Validado contra chaves reais -- ver
|
||||
`tests/modules/fiscal/test_chave_acesso.py`.
|
||||
"""
|
||||
if len(chave43) != 43 or not chave43.isdigit():
|
||||
raise ValueError("chave sem DV deve ter 43 dígitos numéricos")
|
||||
soma = 0
|
||||
for i, digito in enumerate(reversed(chave43)):
|
||||
soma += int(digito) * _PESOS[i % 8]
|
||||
resto = soma % 11
|
||||
return "0" if resto in (0, 1) else str(11 - resto)
|
||||
|
||||
|
||||
def gerar_cnf(numero_nnf: int) -> str:
|
||||
"""Sorteia o código numérico (cNF) de 8 dígitos, criptograficamente
|
||||
aleatório, garantindo cNF != nNF (regra NT 2019.001 -- SEFAZ rejeita
|
||||
quando os dois coincidem).
|
||||
|
||||
F3a (review 2026-07-16): também rejeita `cNF == "00000000"`. All-zeros
|
||||
é um valor técnicamente válido pelo formato (8 dígitos), mas é o chute
|
||||
óbvio de qualquer gerador ingênuo/mockado -- aceitá-lo aqui abriria
|
||||
espaço para uma chave previsível se algum outro ponto do código algum
|
||||
dia trocasse `secrets.randbelow` por algo mais fraco sem essa rede de
|
||||
segurança. Nunca colide com `numero_nnf` de propósito (nNF sempre >= 1,
|
||||
nunca 0 -- `montar_chave_acesso` já exige `numero >= 1`), então esta
|
||||
checagem nunca é a mesma da de `cNF != nNF` acima."""
|
||||
while True:
|
||||
cnf = f"{secrets.randbelow(100_000_000):08d}"
|
||||
if int(cnf) != numero_nnf and int(cnf) != 0:
|
||||
return cnf
|
||||
|
||||
|
||||
def montar_chave_acesso(
|
||||
*,
|
||||
uf_ibge: str,
|
||||
aamm: str,
|
||||
cnpj: str,
|
||||
modelo: str,
|
||||
serie: int,
|
||||
numero: int,
|
||||
tp_emis: str,
|
||||
cnf: str,
|
||||
) -> str:
|
||||
"""Monta a chave de acesso de 44 dígitos a partir dos componentes já
|
||||
formatados/validados pelo chamador (o orquestrador de emissão, Task 6).
|
||||
Pura: não consulta banco, não decide nenhum dos valores -- só concatena
|
||||
e calcula o DV.
|
||||
|
||||
F1 (review 2026-07-16): valida CADA componente individualmente (tamanho
|
||||
exato + só dígitos), não apenas o tamanho AGREGADO de 43 dígitos do
|
||||
`chave43` final. Só checar o total é insuficiente -- dois componentes
|
||||
errados que se COMPENSAM em dígitos (ex.: um `cnpj` de 13 dígitos + um
|
||||
`cnf` de 9, em vez dos 14+8 corretos) somam exatamente 43 e passavam
|
||||
batendo pela checagem antiga, produzindo uma chave de 44 dígitos
|
||||
SINTATICAMENTE válida (DV correto) mas que aponta para o CNPJ/emitente
|
||||
ERRADO -- um bug silencioso que só se manifestaria como uma rejeição
|
||||
(ou pior, uma autorização indevida) na SEFAZ, não em teste algum. Cada
|
||||
checagem abaixo isola o componente específico que a violaria."""
|
||||
if len(uf_ibge) != 2 or not uf_ibge.isdigit():
|
||||
raise ValueError(f"uf_ibge deve ter 2 dígitos numéricos: {uf_ibge!r}")
|
||||
if len(aamm) != 4 or not aamm.isdigit():
|
||||
raise ValueError(f"aamm deve ter 4 dígitos numéricos: {aamm!r}")
|
||||
if len(cnpj) != 14 or not cnpj.isdigit():
|
||||
raise ValueError(f"cnpj deve ter 14 dígitos numéricos: {cnpj!r}")
|
||||
if len(modelo) != 2 or not modelo.isdigit():
|
||||
raise ValueError(f"modelo deve ter 2 dígitos numéricos: {modelo!r}")
|
||||
if len(tp_emis) != 1 or not tp_emis.isdigit():
|
||||
raise ValueError(f"tp_emis deve ter 1 dígito numérico: {tp_emis!r}")
|
||||
if len(cnf) != 8 or not cnf.isdigit():
|
||||
raise ValueError(f"cnf deve ter 8 dígitos numéricos: {cnf!r}")
|
||||
if not (0 <= serie <= 999):
|
||||
raise ValueError(f"serie fora do intervalo 0-999: {serie}")
|
||||
if not (1 <= numero <= 999_999_999):
|
||||
raise ValueError(f"numero fora do intervalo 1-999999999: {numero}")
|
||||
|
||||
chave43 = f"{uf_ibge}{aamm}{cnpj}{modelo}{serie:03d}{numero:09d}{tp_emis}{cnf}"
|
||||
# Invariante garantido pelas checagens de componente acima -- mantido
|
||||
# como cinto-e-suspensório (nunca deve disparar; se disparar, é sinal
|
||||
# de que uma checagem de componente ficou desalinhada com o format
|
||||
# string abaixo dela).
|
||||
if len(chave43) != 43:
|
||||
raise ValueError(f"componentes somam {len(chave43)} dígitos, esperado 43")
|
||||
return chave43 + dv_modulo11(chave43)
|
||||
@@ -0,0 +1,140 @@
|
||||
"""Catálogo de domínios tributários do motor fiscal (Bloco B).
|
||||
|
||||
`tax_domain` é `String(20)` no banco DE PROPÓSITO — validado por ESTE enum
|
||||
Python na borda Pydantic. Adicionar um tributo (IBS/CBS/qualquer futuro) é
|
||||
um membro novo aqui + uma entrada em DOMAIN_FIELDS: sem ALTER TYPE, sem
|
||||
migration, sem o gotcha enum-NAME-vs-value que já quebrou produção neste
|
||||
projeto (ver project_enum_migration_gotcha). Este é o "imposto-como-dado"
|
||||
literal da Opção 1 do spec.
|
||||
"""
|
||||
import enum
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
class TaxDomain(str, enum.Enum):
|
||||
ICMS = "icms"
|
||||
ICMSST = "icmsst"
|
||||
IPI = "ipi"
|
||||
PIS = "pis"
|
||||
COFINS = "cofins"
|
||||
DIFAL = "difal"
|
||||
FCP = "fcp"
|
||||
IBS = "ibs"
|
||||
CBS = "cbs"
|
||||
ISS = "iss"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DomainFieldSpec:
|
||||
"""O que o form da UI mostra/exige para uma regra deste domínio.
|
||||
|
||||
Nomes de campo = colunas de resultado da TaxRule. O frontend NÃO
|
||||
hard-coda isto — consome via GET /fiscal/tax-domains.
|
||||
|
||||
`campos_um_de`: grupos "exatamente um de" (review do frontend,
|
||||
2026-07-15) — para ICMS/ICMS-ST a situação tributária é exatamente um
|
||||
de {cst, csosn} (qual, depende do CRT da filial: Simples → CSOSN,
|
||||
Normal → CST). Sem isto no catálogo, o form dinâmico não tem como
|
||||
exigir situação tributária e o usuário come um 422 que o form não
|
||||
previu. O validator das regras consome ESTES grupos — uma fonte só.
|
||||
"""
|
||||
|
||||
label: str
|
||||
campos_aplicaveis: tuple[str, ...]
|
||||
campos_obrigatorios: tuple[str, ...]
|
||||
campos_um_de: tuple[tuple[str, ...], ...] = ()
|
||||
|
||||
|
||||
_SITUACAO = ("cst", "csosn") # exatamente um dos dois, conforme o CRT
|
||||
_VALORES = ("base_calc_percent", "aliquota")
|
||||
|
||||
DOMAIN_FIELDS: dict[str, DomainFieldSpec] = {
|
||||
TaxDomain.ICMS.value: DomainFieldSpec(
|
||||
label="ICMS",
|
||||
campos_aplicaveis=(*_SITUACAO, "cfop", *_VALORES, "codigo_beneficio"),
|
||||
campos_obrigatorios=("cfop",),
|
||||
campos_um_de=(_SITUACAO,),
|
||||
),
|
||||
TaxDomain.ICMSST.value: DomainFieldSpec(
|
||||
label="ICMS-ST",
|
||||
campos_aplicaveis=(*_SITUACAO, "cfop", *_VALORES, "mva", "aliquota_st", "codigo_beneficio"),
|
||||
campos_obrigatorios=("cfop",),
|
||||
campos_um_de=(_SITUACAO,),
|
||||
),
|
||||
TaxDomain.IPI.value: DomainFieldSpec(
|
||||
label="IPI",
|
||||
campos_aplicaveis=("cst", *_VALORES, "codigo_beneficio"),
|
||||
campos_obrigatorios=("cst",),
|
||||
),
|
||||
TaxDomain.PIS.value: DomainFieldSpec(
|
||||
label="PIS", campos_aplicaveis=("cst", *_VALORES), campos_obrigatorios=("cst",),
|
||||
),
|
||||
TaxDomain.COFINS.value: DomainFieldSpec(
|
||||
label="COFINS", campos_aplicaveis=("cst", *_VALORES), campos_obrigatorios=("cst",),
|
||||
),
|
||||
TaxDomain.DIFAL.value: DomainFieldSpec(
|
||||
label="DIFAL", campos_aplicaveis=_VALORES, campos_obrigatorios=("aliquota",),
|
||||
),
|
||||
TaxDomain.FCP.value: DomainFieldSpec(
|
||||
label="FCP", campos_aplicaveis=("fcp_percent",), campos_obrigatorios=("fcp_percent",),
|
||||
),
|
||||
# Reforma tributária: domínios ACEITOS hoje (regra pode ser cadastrada),
|
||||
# cálculo entra no roadmap R1 do spec quando SEFAZ publicar o leiaute.
|
||||
TaxDomain.IBS.value: DomainFieldSpec(
|
||||
label="IBS", campos_aplicaveis=_VALORES, campos_obrigatorios=("aliquota",),
|
||||
),
|
||||
TaxDomain.CBS.value: DomainFieldSpec(
|
||||
label="CBS", campos_aplicaveis=_VALORES, campos_obrigatorios=("aliquota",),
|
||||
),
|
||||
TaxDomain.ISS.value: DomainFieldSpec(
|
||||
label="ISS", campos_aplicaveis=_VALORES, campos_obrigatorios=("aliquota",),
|
||||
),
|
||||
}
|
||||
|
||||
# Especificidade ponderada (spec, seção resolvedor): potências de 2 em ordem
|
||||
# ESTRITA — cada matcher supera a soma de todos os mais fracos (64 > 63),
|
||||
# então nenhuma combinação de matchers genéricos vence um mais seletivo.
|
||||
# Caveat registrado no spec: ncm > cest; se um preset futuro criar regra-NCM
|
||||
# que sombreie regra-CEST-de-ST, revisitar estes dois.
|
||||
#
|
||||
# M5 (auditoria Fable Bloco B) — pegadinha dos prefixos NCM ANINHADOS: o
|
||||
# peso de `ncm_prefix` é fixo (64) independente do COMPRIMENTO do prefixo —
|
||||
# uma regra com `ncm_prefix="87"` e outra com `ncm_prefix="8708"` casando a
|
||||
# MESMA peça (NCM 87089990) EMPATAM em peso (ambas ganham 64), mesmo a
|
||||
# segunda sendo estritamente mais específica. Isto é AMBIGUIDADE DELIBERADA
|
||||
# (vira `AmbiguousRuleError`, fail-closed, nunca escolha silenciosa do
|
||||
# prefixo mais longo) — o resolvedor NÃO faz "longest-prefix-wins" como um
|
||||
# roteador de IP faria. A UI de cadastro de regras deve orientar o tenant a
|
||||
# usar prefixos NCM DISJUNTOS (não aninhados) dentro do mesmo domínio/
|
||||
# situação, ou aceitar o 409 de ambiguidade como sinal de conflito a
|
||||
# resolver manualmente.
|
||||
MATCHER_WEIGHTS: dict[str, int] = {
|
||||
"ncm_prefix": 64,
|
||||
"cest": 32,
|
||||
"consumidor_final": 16,
|
||||
"indicador_ie": 8,
|
||||
"uf_destino_tipo": 4,
|
||||
"tipo_operacao": 2,
|
||||
"crt": 1,
|
||||
}
|
||||
|
||||
# Transposição CFOP intra→interestadual. A tabela explícita existe porque a
|
||||
# regra geral (trocar 5→6 / 1→2) está ERRADA para ST: 5405 → 6403, não 6405.
|
||||
# CFOPs de ST/exceção DEVEM estar aqui; o fallback genérico cobre o resto.
|
||||
TRANSPOSICAO_CFOP: dict[str, str] = {
|
||||
"5102": "6102",
|
||||
"5405": "6403",
|
||||
"1202": "2202",
|
||||
}
|
||||
|
||||
_GENERIC_FIRST_DIGIT = {"5": "6", "1": "2"}
|
||||
|
||||
|
||||
def transpose_cfop(cfop: str) -> str:
|
||||
"""CFOP-base intra-estadual → interestadual (requisito herdado #3 da 1b)."""
|
||||
if cfop in TRANSPOSICAO_CFOP:
|
||||
return TRANSPOSICAO_CFOP[cfop]
|
||||
first = cfop[0]
|
||||
if first in _GENERIC_FIRST_DIGIT:
|
||||
return _GENERIC_FIRST_DIGIT[first] + cfop[1:]
|
||||
return cfop
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Presets fiscais seed-editáveis (decisão batida #3/#4 do spec): as regras
|
||||
são COPIADAS como TaxRule reais do tenant, que edita/apaga à vontade.
|
||||
|
||||
Valores REAIS extraídos do iCode da Thiago
|
||||
(thoughts/2026-07-13-icode-config-fiscal-extraida.md) — perfis "Padrão"
|
||||
(cód 110) e "Óleos/ST" (cód 108). NÃO inventar valores aqui.
|
||||
|
||||
PIS/COFINS do Óleos: CST 04 (monofásico — revenda a alíquota zero). O
|
||||
`~49/52` da extração tinha um til de incerteza; óleo lubrificante revendido
|
||||
é monofásico POR LEGISLAÇÃO (não é regra do cliente), logo CST 04.
|
||||
⚠️ Confirmar contra a aba PIS/COFINS do perfil 108 no iCode antes do
|
||||
primeiro tenant novo usar este preset em emissão real (tarefa registrada).
|
||||
|
||||
NENHUM preset carrega `origem`: origem é sempre do Part (review do
|
||||
frontend, 2026-07-14).
|
||||
"""
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import Decimal
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PresetRule:
|
||||
tax_domain: str
|
||||
crt: str | None = None
|
||||
tipo_operacao: str | None = None
|
||||
cst: str | None = None
|
||||
csosn: str | None = None
|
||||
cfop: str | None = None
|
||||
aliquota: Decimal | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FiscalPreset:
|
||||
key: str
|
||||
name: str
|
||||
description: str
|
||||
# M2 (auditoria Fable Bloco B): CRT-alvo do preset ("1" = Simples
|
||||
# Nacional para os dois presets extraídos do iCode) -- exposto no
|
||||
# catálogo (`GET /fiscal/presets`) pra UI avisar quando o CRT da filial
|
||||
# não bate com o preset que o tenant está aplicando.
|
||||
regime_alvo: str = "1"
|
||||
rules: tuple[PresetRule, ...] = field(default_factory=tuple)
|
||||
|
||||
|
||||
PRESETS: dict[str, FiscalPreset] = {
|
||||
"autopecas_simples_padrao": FiscalPreset(
|
||||
key="autopecas_simples_padrao",
|
||||
name="Autopeças — Simples Nacional (Padrão)",
|
||||
description=(
|
||||
"Venda de peças no Simples: CSOSN 102, CFOP 5102 (interno; o motor "
|
||||
"transpõe para 6102 fora do estado), PIS/COFINS CST 08 alíquota 0, "
|
||||
"IPI não destacado. Devolução (entrada): CSOSN 102, CFOP 1202/2202."
|
||||
),
|
||||
rules=(
|
||||
# ICMS escopado a `tipo_operacao="venda"` (review 2026-07-15,
|
||||
# IMPORTANTE 1): sem escopo, esta regra também casava devolução
|
||||
# (tipo_operacao=NULL casa qualquer) e devolvia CFOP de SAÍDA
|
||||
# (5102/6102) para uma operação de ENTRADA — fail-open num motor
|
||||
# fail-closed. A regra de devolução abaixo é a única que cobre
|
||||
# tipo_operacao="devolucao" para este preset.
|
||||
PresetRule(
|
||||
tax_domain="icms", crt="1", tipo_operacao="venda",
|
||||
csosn="102", cfop="5102", aliquota=Decimal("0"),
|
||||
),
|
||||
PresetRule(
|
||||
tax_domain="icms", crt="1", tipo_operacao="devolucao",
|
||||
csosn="102", cfop="1202", aliquota=Decimal("0"),
|
||||
),
|
||||
# PIS/COFINS ficam sem escopo de tipo_operacao: os mesmos CSTs
|
||||
# valem para venda e devolução.
|
||||
PresetRule(tax_domain="pis", cst="08", aliquota=Decimal("0")),
|
||||
PresetRule(tax_domain="cofins", cst="08", aliquota=Decimal("0")),
|
||||
),
|
||||
),
|
||||
"autopecas_simples_st": FiscalPreset(
|
||||
key="autopecas_simples_st",
|
||||
name="Autopeças — Simples Nacional (Óleos/ST)",
|
||||
description=(
|
||||
"Peça/óleo com ICMS-ST retido pelo fornecedor: CSOSN 500, CFOP 5405 "
|
||||
"(interno; 6403 fora do estado), PIS/COFINS CST 04 (monofásico). "
|
||||
"Devolução (entrada): CSOSN 500, CFOP 1202/2202."
|
||||
),
|
||||
rules=(
|
||||
# Mesmo escopo de tipo_operacao que o preset Padrão — ver
|
||||
# comentário acima.
|
||||
PresetRule(
|
||||
tax_domain="icmsst", crt="1", tipo_operacao="venda",
|
||||
csosn="500", cfop="5405",
|
||||
),
|
||||
PresetRule(
|
||||
tax_domain="icmsst", crt="1", tipo_operacao="devolucao",
|
||||
csosn="500", cfop="1202",
|
||||
),
|
||||
PresetRule(tax_domain="pis", cst="04", aliquota=Decimal("0")),
|
||||
PresetRule(tax_domain="cofins", cst="04", aliquota=Decimal("0")),
|
||||
),
|
||||
),
|
||||
}
|
||||
@@ -0,0 +1,227 @@
|
||||
"""Resolvedor fiscal puro do Bloco B: FiscalOperation × TaxRule → FiscalResult.
|
||||
|
||||
Função pura, determinística, ZERO I/O — recebe as regras já carregadas.
|
||||
Quem toca o banco é fiscal/service.py. Fail-closed: sem regra para um
|
||||
domínio OBRIGATÓRIO (icms/icmsst com CFOP + situação tributária), levanta
|
||||
FiscalConfigError — nunca emite imposto adivinhado (decisão batida #2 do
|
||||
spec). `origem` é ecoada do ITEM (Part.icms_origem) e jamais de regra.
|
||||
"""
|
||||
import uuid
|
||||
from decimal import ROUND_HALF_UP, Decimal
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from sowai_fiscal.domains import MATCHER_WEIGHTS, TaxDomain, transpose_cfop
|
||||
from sowai_fiscal.types import TaxRuleLike
|
||||
|
||||
_CENT = Decimal("0.01")
|
||||
|
||||
|
||||
class FiscalItem(BaseModel):
|
||||
tax_profile_id: uuid.UUID
|
||||
ncm: str
|
||||
cest: str | None
|
||||
origem: str | None
|
||||
# M1 (auditoria Fable Bloco B): input completo que o spec lista (costura
|
||||
# do R4/ISS) -- nenhuma regra o consome hoje, ecoado do
|
||||
# `Part.tipo_fiscal` pelo simulate.
|
||||
tipo_fiscal: str | None = None
|
||||
quantity: Decimal
|
||||
unit_price: Decimal
|
||||
|
||||
|
||||
class FiscalOperation(BaseModel):
|
||||
crt: str
|
||||
uf_origem: str
|
||||
uf_destino: str
|
||||
indicador_ie: str
|
||||
consumidor_final: bool
|
||||
tipo_operacao: str
|
||||
item: FiscalItem
|
||||
|
||||
@property
|
||||
def uf_destino_tipo(self) -> str:
|
||||
return "interna" if self.uf_origem == self.uf_destino else "interestadual"
|
||||
|
||||
|
||||
class TributoLinha(BaseModel):
|
||||
tax_domain: str
|
||||
cst: str | None
|
||||
csosn: str | None
|
||||
base_calc: Decimal
|
||||
base_calc_percent: Decimal
|
||||
aliquota: Decimal | None
|
||||
valor: Decimal
|
||||
mva: Decimal | None
|
||||
aliquota_st: Decimal | None
|
||||
fcp_percent: Decimal | None
|
||||
codigo_beneficio: str | None
|
||||
rule_id: uuid.UUID
|
||||
|
||||
|
||||
class FiscalResult(BaseModel):
|
||||
cfop: str
|
||||
cst: str | None
|
||||
csosn: str | None
|
||||
origem: str | None
|
||||
consumidor_final: bool
|
||||
indicador_ie: str
|
||||
tributos: list[TributoLinha]
|
||||
|
||||
|
||||
class FiscalConfigError(Exception):
|
||||
"""Fail-closed: configuração fiscal faltante — a emissão DEVE bloquear."""
|
||||
|
||||
def __init__(self, missing: list[str]):
|
||||
self.missing = missing
|
||||
super().__init__(
|
||||
"Configuração fiscal ausente para: " + ", ".join(missing)
|
||||
)
|
||||
|
||||
|
||||
class AmbiguousRuleError(Exception):
|
||||
"""Duas regras casam com o MESMO peso — erro de configuração, nunca
|
||||
escolha silenciosa (decisão do spec, review do frontend)."""
|
||||
|
||||
def __init__(self, tax_domain: str, rule_ids: list[uuid.UUID]):
|
||||
self.tax_domain = tax_domain
|
||||
self.rule_ids = rule_ids
|
||||
super().__init__(
|
||||
f"Regras ambíguas para o domínio {tax_domain!r}: "
|
||||
+ ", ".join(str(r) for r in rule_ids)
|
||||
)
|
||||
|
||||
|
||||
def _matches(rule: TaxRuleLike, op: FiscalOperation) -> int | None:
|
||||
"""Peso da regra para a operação, ou None se algum matcher não casa."""
|
||||
weight = 0
|
||||
checks: list[tuple[str, bool]] = [
|
||||
("crt", rule.crt is None or rule.crt == op.crt),
|
||||
("uf_destino_tipo", rule.uf_destino_tipo is None or rule.uf_destino_tipo == op.uf_destino_tipo),
|
||||
("consumidor_final", rule.consumidor_final is None or rule.consumidor_final == op.consumidor_final),
|
||||
("indicador_ie", rule.indicador_ie is None or rule.indicador_ie == op.indicador_ie),
|
||||
("tipo_operacao", rule.tipo_operacao is None or rule.tipo_operacao == op.tipo_operacao),
|
||||
("ncm_prefix", rule.ncm_prefix is None or op.item.ncm.startswith(rule.ncm_prefix)),
|
||||
("cest", rule.cest is None or rule.cest == op.item.cest),
|
||||
]
|
||||
for campo, ok in checks:
|
||||
if not ok:
|
||||
return None
|
||||
if getattr(rule, campo) is not None:
|
||||
weight += MATCHER_WEIGHTS[campo]
|
||||
return weight
|
||||
|
||||
|
||||
def _pick(rules: list[TaxRuleLike], domain: str, op: FiscalOperation) -> TaxRuleLike | None:
|
||||
candidates: list[tuple[int, TaxRuleLike]] = []
|
||||
for rule in rules:
|
||||
if rule.tax_domain != domain or rule.deleted_at is not None:
|
||||
continue
|
||||
weight = _matches(rule, op)
|
||||
if weight is not None:
|
||||
candidates.append((weight, rule))
|
||||
if not candidates:
|
||||
return None
|
||||
top = max(w for w, _ in candidates)
|
||||
winners = [r for w, r in candidates if w == top]
|
||||
if len(winners) > 1:
|
||||
raise AmbiguousRuleError(domain, [r.id for r in winners])
|
||||
return winners[0]
|
||||
|
||||
|
||||
def _linha(rule: TaxRuleLike, base: Decimal) -> TributoLinha:
|
||||
pct = rule.base_calc_percent if rule.base_calc_percent is not None else Decimal("100")
|
||||
base_efetiva = (base * pct / Decimal("100")).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
aliquota = rule.aliquota
|
||||
if rule.tax_domain == TaxDomain.FCP.value:
|
||||
# I1 (auditoria Fable Bloco B): o domínio `fcp` NÃO tem `aliquota`
|
||||
# no catálogo (DOMAIN_FIELDS['fcp'] só declara `fcp_percent`) — o
|
||||
# ramo genérico abaixo (aliquota is None → 0.00) fazia toda linha
|
||||
# FCP sair com valor zerado, um preview que mente. Domínio-explícito
|
||||
# de propósito: NÃO um fallback genérico "aliquota None usa
|
||||
# fcp_percent", que surpreenderia outros domínios (ex.: PIS/COFINS
|
||||
# CST 08/04, onde aliquota None É zero de verdade).
|
||||
taxa = rule.fcp_percent if rule.fcp_percent is not None else Decimal("0")
|
||||
valor = (base_efetiva * taxa / Decimal("100")).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
else:
|
||||
valor = (
|
||||
(base_efetiva * aliquota / Decimal("100")).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
if aliquota is not None
|
||||
else Decimal("0.00")
|
||||
)
|
||||
return TributoLinha(
|
||||
tax_domain=rule.tax_domain,
|
||||
cst=rule.cst, csosn=rule.csosn,
|
||||
base_calc=base_efetiva, base_calc_percent=pct,
|
||||
aliquota=aliquota, valor=valor,
|
||||
mva=rule.mva, aliquota_st=rule.aliquota_st,
|
||||
fcp_percent=rule.fcp_percent, codigo_beneficio=rule.codigo_beneficio,
|
||||
rule_id=rule.id,
|
||||
)
|
||||
|
||||
|
||||
def _linha_st(rule: TaxRuleLike, base: Decimal, valor_icms_proprio: Decimal) -> TributoLinha:
|
||||
"""ICMS-ST: base_st = base × (1 + MVA%); valor = base_st × aliq_st − ICMS próprio.
|
||||
Para CSOSN 500 (ST retida) mva/aliquota_st são nulos → valor 0, correto."""
|
||||
linha = _linha(rule, base)
|
||||
if rule.mva is not None and rule.aliquota_st is not None:
|
||||
base_st = (base * (Decimal("100") + rule.mva) / Decimal("100")).quantize(
|
||||
_CENT, rounding=ROUND_HALF_UP
|
||||
)
|
||||
bruto = (base_st * rule.aliquota_st / Decimal("100")).quantize(
|
||||
_CENT, rounding=ROUND_HALF_UP
|
||||
)
|
||||
linha = linha.model_copy(
|
||||
update={"base_calc": base_st, "valor": max(Decimal("0.00"), bruto - valor_icms_proprio)}
|
||||
)
|
||||
return linha
|
||||
|
||||
|
||||
def resolve_fiscal(rules: list[TaxRuleLike], op: FiscalOperation) -> FiscalResult:
|
||||
base = (op.item.quantity * op.item.unit_price).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
# 1. Âncora: ICMS ou ICMS-ST — deve existir, com CFOP + situação.
|
||||
icms_rule = _pick(rules, TaxDomain.ICMS.value, op)
|
||||
icmsst_rule = _pick(rules, TaxDomain.ICMSST.value, op)
|
||||
anchor = icmsst_rule or icms_rule # ST é a situação mais específica quando ambos casam
|
||||
missing: list[str] = []
|
||||
if anchor is None:
|
||||
missing.append("icms (ou icmsst): nenhuma regra casa com a operação")
|
||||
else:
|
||||
if anchor.cfop is None:
|
||||
missing.append(f"cfop na regra {anchor.id} ({anchor.tax_domain})")
|
||||
if anchor.cst is None and anchor.csosn is None:
|
||||
missing.append(f"cst/csosn na regra {anchor.id} ({anchor.tax_domain})")
|
||||
if missing:
|
||||
raise FiscalConfigError(missing)
|
||||
|
||||
# 2. CFOP com transposição (requisito herdado #3).
|
||||
cfop = anchor.cfop
|
||||
if op.uf_destino_tipo == "interestadual":
|
||||
cfop = transpose_cfop(cfop)
|
||||
|
||||
# 3. Tributos: uma linha por domínio que tiver regra casando.
|
||||
tributos: list[TributoLinha] = []
|
||||
icms_linha = _linha(icms_rule, base) if icms_rule is not None else None
|
||||
if icms_linha is not None:
|
||||
tributos.append(icms_linha)
|
||||
if icmsst_rule is not None:
|
||||
valor_proprio = icms_linha.valor if icms_linha is not None else Decimal("0.00")
|
||||
tributos.append(_linha_st(icmsst_rule, base, valor_proprio))
|
||||
for domain in (
|
||||
TaxDomain.IPI, TaxDomain.PIS, TaxDomain.COFINS,
|
||||
TaxDomain.DIFAL, TaxDomain.FCP, TaxDomain.IBS, TaxDomain.CBS, TaxDomain.ISS,
|
||||
):
|
||||
rule = _pick(rules, domain.value, op)
|
||||
if rule is not None:
|
||||
tributos.append(_linha(rule, base))
|
||||
|
||||
return FiscalResult(
|
||||
cfop=cfop,
|
||||
cst=anchor.cst,
|
||||
csosn=anchor.csosn,
|
||||
origem=op.item.origem, # SEMPRE do item (Part) — nunca de regra
|
||||
consumidor_final=op.consumidor_final,
|
||||
indicador_ie=op.indicador_ie,
|
||||
tributos=tributos,
|
||||
)
|
||||
@@ -0,0 +1,23 @@
|
||||
"""Códigos tPag da NF-e (tabela do MOC, grupo YA02). String no banco,
|
||||
validação aqui -- um código novo (ex.: futuro meio de pagamento) é uma linha
|
||||
neste dict, sem migration."""
|
||||
|
||||
TPAG_CODES: dict[str, str] = {
|
||||
"01": "Dinheiro",
|
||||
"02": "Cheque",
|
||||
"03": "Cartão de Crédito",
|
||||
"04": "Cartão de Débito",
|
||||
"05": "Crédito Loja",
|
||||
"10": "Vale Alimentação",
|
||||
"11": "Vale Refeição",
|
||||
"12": "Vale Presente",
|
||||
"13": "Vale Combustível",
|
||||
"14": "Duplicata Mercantil",
|
||||
"15": "Boleto Bancário",
|
||||
"16": "Depósito Bancário",
|
||||
"17": "Pagamento Instantâneo (PIX)",
|
||||
"18": "Transferência bancária, Carteira Digital",
|
||||
"19": "Programa de fidelidade, Cashback, Crédito Virtual",
|
||||
"90": "Sem pagamento",
|
||||
"99": "Outros",
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
"""Interface estrutural da regra fiscal — o desacoplamento que permite a lib
|
||||
existir. O produto passa seus objetos (no auto, o ORM TaxRule) e eles
|
||||
satisfazem o Protocol por estrutura, sem herdar nada daqui."""
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Protocol
|
||||
|
||||
|
||||
class TaxRuleLike(Protocol):
|
||||
id: uuid.UUID
|
||||
tax_domain: str
|
||||
crt: str | None
|
||||
uf_destino_tipo: str | None
|
||||
consumidor_final: bool | None
|
||||
indicador_ie: str | None
|
||||
tipo_operacao: str | None
|
||||
ncm_prefix: str | None
|
||||
cest: str | None
|
||||
cst: str | None
|
||||
csosn: str | None
|
||||
cfop: str | None
|
||||
base_calc_percent: Decimal | None
|
||||
aliquota: Decimal | None
|
||||
mva: Decimal | None
|
||||
aliquota_st: Decimal | None
|
||||
fcp_percent: Decimal | None
|
||||
codigo_beneficio: str | None
|
||||
deleted_at: datetime | None
|
||||
@@ -0,0 +1,563 @@
|
||||
"""Montagem PURA do XML da NF-e 55 (leiaute NT 2025.002 v1.40) -- 1b.1
|
||||
Task 5. Recebe um `DadosEmissao` já COMPLETO (o orquestrador de emissão,
|
||||
`emissao.py`/Task 6, é quem coleta/valida os dados de Branch/Person/Part/
|
||||
FiscalResult e monta este dataclass) e devolve um objeto `Nfe` (bindings
|
||||
xsdata do pacote `nfelib`, os MESMOS usados por `estoque/nfe_import.py`
|
||||
para PARSE -- aqui construímos os mesmos dataclasses em vez de lê-los).
|
||||
|
||||
ZERO I/O, ZERO sessão de banco, ZERO decisão de negócio: este módulo não
|
||||
sabe o que é uma Branch ou uma Sale, só sabe montar o XML a partir dos
|
||||
campos já resolvidos. Fail-closed (409 `fiscal_config_missing` nomeando
|
||||
o campo) é responsabilidade do ORQUESTRADOR -- este builder assume que
|
||||
`DadosEmissao` está completo e, se não estiver (ex.: `None` onde um campo
|
||||
obrigatório do leiaute era esperado), deixa o próprio xsdata/lxml estourar
|
||||
uma exceção genérica (AttributeError/TypeError), não um 409 estruturado.
|
||||
|
||||
Nomes dos campos vêm do binding REAL instalado (nfelib 2.5.2, inspecionado
|
||||
diretamente no venv -- `nfelib.nfe.bindings.v4_0.leiaute_nfe_v4_00.Tnfe.
|
||||
InfNfe` e seus subgrupos -- não chutados): `nfelib.nfe.bindings.v4_0.
|
||||
nfe_v4_00.Nfe` é a raiz `<NFe>`; `.infNFe` é o `Tnfe.InfNfe`.
|
||||
|
||||
Mapeamento de situação tributária ICMS (CST/CSOSN -> grupo do binding):
|
||||
DELIBERADAMENTE parcial -- cobre exatamente as situações alcançáveis pelos
|
||||
presets hoje cadastrados (`fiscal.presets`, CSOSN 102/103/300/400 e 500) e
|
||||
mais alguns CSTs de regime normal (00/40/41/50/60/90) de cobertura óbvia.
|
||||
Uma situação fora desta tabela levanta `UnsupportedIcmsSituationError` --
|
||||
erro claro, não um XML silenciosamente errado -- e a tabela deve crescer
|
||||
quando um preset/regra novo precisar de outra situação.
|
||||
"""
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import ROUND_HALF_UP, Decimal
|
||||
|
||||
from nfelib.nfe.bindings.v4_0.leiaute_nfe_v4_00 import Tendereco, TenderEmi, Tnfe
|
||||
from nfelib.nfe.bindings.v4_0.nfe_v4_00 import Nfe
|
||||
|
||||
from sowai_fiscal.resolver import FiscalResult, TributoLinha
|
||||
|
||||
_InfNfe = Tnfe.InfNfe
|
||||
_Det = _InfNfe.Det
|
||||
_Icms = _Det.Imposto.Icms
|
||||
_Pis = _Det.Imposto.Pis
|
||||
_Cofins = _Det.Imposto.Cofins
|
||||
|
||||
_CENT = Decimal("0.01")
|
||||
|
||||
# Texto oficial (NT NF-e -- ambiente de homologação): substitui o xNome do
|
||||
# destinatário em toda nota emitida em homologação (tpAmb=2), independente
|
||||
# do nome real informado.
|
||||
TEXTO_HOMOLOGACAO = "NF-E EMITIDA EM AMBIENTE DE HOMOLOGACAO - SEM VALOR FISCAL"
|
||||
|
||||
_SEM_GTIN = "SEM GTIN"
|
||||
|
||||
|
||||
class UnsupportedIcmsSituationError(ValueError):
|
||||
"""CST/CSOSN fora da tabela de mapeamento deste builder (ver docstring
|
||||
do módulo) -- NÃO é um 409 de dado faltante (o motor resolveu uma
|
||||
situação válida), é uma lacuna de COBERTURA deste builder. Deve ser
|
||||
tratado como bug/gap a fechar quando aparecer, não silenciado."""
|
||||
|
||||
def __init__(self, cst: str | None, csosn: str | None):
|
||||
self.cst = cst
|
||||
self.csosn = csosn
|
||||
super().__init__(
|
||||
f"Situação tributária ICMS não suportada pelo builder: cst={cst!r} csosn={csosn!r}"
|
||||
)
|
||||
|
||||
|
||||
# --- dataclasses de input (montados pelo orquestrador, Task 6) -------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class EmitenteData:
|
||||
cnpj: str
|
||||
razao_social: str
|
||||
nome_fantasia: str | None
|
||||
ie: str
|
||||
crt: str # "1"|"2"|"3" (CRT.value)
|
||||
address_street: str
|
||||
address_number: str
|
||||
address_complement: str | None
|
||||
address_district: str
|
||||
address_city: str
|
||||
address_state: str
|
||||
address_zip: str
|
||||
address_city_ibge_code: str
|
||||
fone: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DestinatarioData:
|
||||
"""`None` no orquestrador == consumidor final não identificado (o
|
||||
grupo `<dest>` inteiro fica ausente -- opcional no schema)."""
|
||||
|
||||
nome: str
|
||||
cnpj: str | None
|
||||
cpf: str | None
|
||||
indicador_ie: str # "1"|"2"|"9" (DestIndIedest)
|
||||
ie: str | None = None
|
||||
address_street: str | None = None
|
||||
address_number: str | None = None
|
||||
address_complement: str | None = None
|
||||
address_district: str | None = None
|
||||
address_city: str | None = None
|
||||
address_state: str | None = None
|
||||
address_zip: str | None = None
|
||||
address_city_ibge_code: str | None = None
|
||||
email: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ItemData:
|
||||
codigo: str # cProd
|
||||
descricao: str # xProd
|
||||
ncm: str
|
||||
cfop: str # já transposto pelo resolvedor (resolve_fiscal)
|
||||
unidade_comercial: str # uCom
|
||||
unidade_tributavel: str # uTrib
|
||||
quantidade: Decimal
|
||||
valor_unitario: Decimal
|
||||
fiscal_result: FiscalResult
|
||||
gtin: str | None = None # None -> "SEM GTIN"
|
||||
cest: str | None = None
|
||||
peso_liquido_kg: Decimal | None = None
|
||||
peso_bruto_kg: Decimal | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class PagamentoData:
|
||||
tpag: str
|
||||
valor: Decimal
|
||||
indpag: str = "0" # 0=à vista, 1=a prazo
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DadosEmissao:
|
||||
emitente: EmitenteData
|
||||
itens: list[ItemData]
|
||||
pagamento: PagamentoData
|
||||
ambiente: str # "homologacao"|"producao"
|
||||
chave_acesso: str # 44 dígitos, já montada (chave_acesso.montar_chave_acesso)
|
||||
numero: int
|
||||
serie: int
|
||||
cnf: str
|
||||
dh_emi: str # ISO 8601 com timezone, ex. "2026-07-16T10:00:00-03:00"
|
||||
uf_destino_tipo: str # "interna"|"interestadual" (mesmo campo do resolvedor)
|
||||
destinatario: DestinatarioData | None = None
|
||||
nat_op: str = "Venda"
|
||||
tp_emis: str = "1"
|
||||
ind_final: str = "1" # 1 = consumidor final (venda de balcão, o caso comum da 1b.1)
|
||||
ind_pres: str = "1" # 1 = operação presencial
|
||||
fin_nfe: str = "1" # 1 = NF-e normal
|
||||
ver_proc: str = "sowai-auto/1b.1"
|
||||
|
||||
|
||||
# --- mapeamento CST/CSOSN -> grupo ICMS do binding --------------------------
|
||||
|
||||
|
||||
def _pct(linha: TributoLinha | None) -> str:
|
||||
if linha is None or linha.aliquota is None:
|
||||
return "0.0000"
|
||||
return str(linha.aliquota.quantize(Decimal("0.0001")))
|
||||
|
||||
|
||||
def _val(linha: TributoLinha | None) -> str:
|
||||
if linha is None:
|
||||
return "0.00"
|
||||
return str(linha.valor.quantize(_CENT, rounding=ROUND_HALF_UP))
|
||||
|
||||
|
||||
def _base(linha: TributoLinha | None) -> str:
|
||||
if linha is None:
|
||||
return "0.00"
|
||||
return str(linha.base_calc.quantize(_CENT, rounding=ROUND_HALF_UP))
|
||||
|
||||
|
||||
_CSOSN_SEM_PERMISSAO_CREDITO = {"102", "103", "300", "400"}
|
||||
|
||||
|
||||
def _build_icms(fiscal_result: FiscalResult, icms_linha, icmsst_linha) -> _Icms:
|
||||
csosn = fiscal_result.csosn
|
||||
cst = fiscal_result.cst
|
||||
# I3 (review Opus, 2026-07-16): SEM fallback `or "0"` -- origem ausente
|
||||
# aqui é erro de PROGRAMAÇÃO (a validação de completude do orquestrador,
|
||||
# `emissao.emitir_documento`, barra peça sem `icms_origem` com 409 antes
|
||||
# de chegar neste builder), nunca um default silencioso: emitir "0"
|
||||
# (nacional) para uma peça importada é imposto errado na SEFAZ.
|
||||
if fiscal_result.origem is None:
|
||||
raise ValueError(
|
||||
"fiscal_result.origem é None -- input incompleto para o builder; "
|
||||
"peça sem icms_origem deveria ter sido barrada na validação de "
|
||||
"completude da emissão (409 fiscal_config_missing)"
|
||||
)
|
||||
orig = fiscal_result.origem
|
||||
|
||||
if csosn is not None:
|
||||
if csosn in _CSOSN_SEM_PERMISSAO_CREDITO:
|
||||
return _Icms(ICMSSN102=_Icms.Icmssn102(orig=orig, CSOSN=csosn))
|
||||
if csosn == "500":
|
||||
return _Icms(
|
||||
ICMSSN500=_Icms.Icmssn500(
|
||||
orig=orig,
|
||||
CSOSN=csosn,
|
||||
vBCSTRet=_base(icmsst_linha),
|
||||
pST=_pct(icmsst_linha),
|
||||
vICMSSTRet=_val(icmsst_linha),
|
||||
)
|
||||
)
|
||||
if csosn == "900":
|
||||
return _Icms(
|
||||
ICMSSN900=_Icms.Icmssn900(
|
||||
orig=orig,
|
||||
CSOSN=csosn,
|
||||
modBC="0",
|
||||
vBC=_base(icms_linha),
|
||||
pICMS=_pct(icms_linha),
|
||||
vICMS=_val(icms_linha),
|
||||
)
|
||||
)
|
||||
raise UnsupportedIcmsSituationError(cst, csosn)
|
||||
|
||||
if cst is not None:
|
||||
if cst == "00":
|
||||
return _Icms(
|
||||
ICMS00=_Icms.Icms00(
|
||||
orig=orig, CST=cst, modBC="0",
|
||||
vBC=_base(icms_linha), pICMS=_pct(icms_linha), vICMS=_val(icms_linha),
|
||||
)
|
||||
)
|
||||
if cst in ("40", "41", "50"):
|
||||
return _Icms(ICMS40=_Icms.Icms40(orig=orig, CST=cst))
|
||||
if cst == "60":
|
||||
return _Icms(
|
||||
ICMS60=_Icms.Icms60(
|
||||
orig=orig, CST=cst,
|
||||
vBCSTRet=_base(icmsst_linha), pST=_pct(icmsst_linha),
|
||||
vICMSSTRet=_val(icmsst_linha),
|
||||
)
|
||||
)
|
||||
if cst == "90":
|
||||
return _Icms(
|
||||
ICMS90=_Icms.Icms90(
|
||||
orig=orig, CST=cst, modBC="0",
|
||||
vBC=_base(icms_linha), pICMS=_pct(icms_linha), vICMS=_val(icms_linha),
|
||||
vBCST=_base(icmsst_linha), pICMSST=_pct(icmsst_linha), vICMSST=_val(icmsst_linha),
|
||||
)
|
||||
)
|
||||
raise UnsupportedIcmsSituationError(cst, csosn)
|
||||
|
||||
raise UnsupportedIcmsSituationError(cst, csosn)
|
||||
|
||||
|
||||
# CST 04-09 -> "NT" (não tributado/isento/suspenso -- só CST, sem base/aliq);
|
||||
# 01/02 -> Aliq (percentual); 03 -> Qtde; qualquer outro -> Outr ("99"),
|
||||
# mesma tabela para PIS e COFINS (os dois têm as mesmas 4 famílias de grupo).
|
||||
_PIS_COFINS_NT_CSTS = {"04", "05", "06", "07", "08", "09"}
|
||||
|
||||
|
||||
def _build_pis(linha: TributoLinha | None) -> _Pis:
|
||||
if linha is None or linha.cst is None:
|
||||
return _Pis(PISNT=_Pis.Pisnt(CST="08"))
|
||||
cst = linha.cst
|
||||
if cst in _PIS_COFINS_NT_CSTS:
|
||||
return _Pis(PISNT=_Pis.Pisnt(CST=cst))
|
||||
if cst in ("01", "02"):
|
||||
return _Pis(
|
||||
PISAliq=_Pis.Pisaliq(CST=cst, vBC=_base(linha), pPIS=_pct(linha), vPIS=_val(linha))
|
||||
)
|
||||
return _Pis(PISOutr=_Pis.Pisoutr(CST="99", vBC=_base(linha), pPIS=_pct(linha), vPIS=_val(linha)))
|
||||
|
||||
|
||||
def _build_cofins(linha: TributoLinha | None) -> _Cofins:
|
||||
if linha is None or linha.cst is None:
|
||||
return _Cofins(COFINSNT=_Cofins.Cofinsnt(CST="08"))
|
||||
cst = linha.cst
|
||||
if cst in _PIS_COFINS_NT_CSTS:
|
||||
return _Cofins(COFINSNT=_Cofins.Cofinsnt(CST=cst))
|
||||
if cst in ("01", "02"):
|
||||
return _Cofins(
|
||||
COFINSAliq=_Cofins.Cofinsaliq(
|
||||
CST=cst, vBC=_base(linha), pCOFINS=_pct(linha), vCOFINS=_val(linha)
|
||||
)
|
||||
)
|
||||
return _Cofins(
|
||||
COFINSOutr=_Cofins.Cofinsoutr(
|
||||
CST="99", vBC=_base(linha), pCOFINS=_pct(linha), vCOFINS=_val(linha)
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _linha_por_dominio(fiscal_result: FiscalResult, dominio: str) -> TributoLinha | None:
|
||||
for linha in fiscal_result.tributos:
|
||||
if linha.tax_domain == dominio:
|
||||
return linha
|
||||
return None
|
||||
|
||||
|
||||
# M4 (auditoria Fable, 2026-07-16): SEM fonte real -- ver o docstring de
|
||||
# `_build_ibscbs`. Módulo-level (não literais inline) de propósito, para o
|
||||
# nome do valor já denunciar no call-site que é fabricado.
|
||||
_PLACEHOLDER_CST_SEM_FONTE_REAL = "000"
|
||||
_PLACEHOLDER_CCLASSTRIB_SEM_FONTE_REAL = "000001"
|
||||
|
||||
|
||||
def _build_ibscbs(fiscal_result: FiscalResult):
|
||||
"""Grupo UB (IBS/CBS, RTC v1.40) -- só presente quando o `FiscalResult`
|
||||
tiver linhas `ibs`/`cbs` (CRT 1 em 2026 nunca tem -- R1 do motor exige
|
||||
isso a partir de 04/01/2027, ver spec). Best-effort ESTRUTURAL: os
|
||||
valores de `vBC`/`pIBSUF`/`vIBSUF`/`pCBS`/`vCBS` vêm das linhas
|
||||
resolvidas (REAIS -- `aliquota`/`valor`/`base_calc` do `TributoLinha`,
|
||||
configurados pelo tenant na regra), mas `CST`/`cClassTrib` abaixo são
|
||||
FABRICADOS ("000"/"000001") -- não existe fonte real para nenhum dos
|
||||
dois hoje: `DOMAIN_FIELDS['ibs'|'cbs']` nem aceita `cst`/
|
||||
`codigo_beneficio` no cadastro da regra, só `aliquota`/
|
||||
`base_calc_percent` (ver `fiscal.domains`). O grupo RTC completo
|
||||
(gIBSUF/gIBSMun/gCBS com suas dezenas de sub-campos) também ainda não
|
||||
tem um mapeamento tão maduro quanto o ICMS legado.
|
||||
|
||||
M4 (auditoria Fable, 2026-07-16): esta função é PROVADAMENTE
|
||||
inalcançável a partir da emissão real -- `fiscal.emissao.
|
||||
emitir_documento` barra com 409 `fiscal_config_missing` ANTES de
|
||||
montar o XML sempre que algum item resolve uma linha ibs/cbs
|
||||
(exatamente o caso em que este `if` abaixo deixaria de devolver
|
||||
`None`). Só existe aqui para o builder PURO continuar com o suporte
|
||||
ESTRUTURAL testável isoladamente (`test_grupo_ibscbs_presente_com_
|
||||
linhas_ibs_cbs`) -- endurecer (CST/cClassTrib reais) junto com o R1 do
|
||||
motor, quando o catálogo de domínios ganhar esses campos."""
|
||||
ibs_linha = _linha_por_dominio(fiscal_result, "ibs")
|
||||
cbs_linha = _linha_por_dominio(fiscal_result, "cbs")
|
||||
if ibs_linha is None and cbs_linha is None:
|
||||
return None
|
||||
|
||||
from nfelib.nfe.bindings.v4_0.dfe_tipos_basicos_v1_00 import Tcibs, TtribNfe
|
||||
|
||||
base = ibs_linha or cbs_linha
|
||||
g_ibs_uf = (
|
||||
Tcibs.GIbsuf(pIBSUF=_pct(ibs_linha), vIBSUF=_val(ibs_linha)) if ibs_linha else None
|
||||
)
|
||||
g_cbs = Tcibs.GCbs(pCBS=_pct(cbs_linha), vCBS=_val(cbs_linha)) if cbs_linha else None
|
||||
|
||||
gibscbs = Tcibs(vBC=_base(base), gIBSUF=g_ibs_uf, gCBS=g_cbs)
|
||||
# M4: nomeados _PLACEHOLDER_* (não `CST`/`cClassTrib` soltos) de
|
||||
# propósito -- deixa explícito, no próprio call-site, que estes DOIS
|
||||
# valores não vêm de lugar nenhum real (ver o docstring acima).
|
||||
return TtribNfe(
|
||||
CST=_PLACEHOLDER_CST_SEM_FONTE_REAL,
|
||||
cClassTrib=_PLACEHOLDER_CCLASSTRIB_SEM_FONTE_REAL,
|
||||
gIBSCBS=gibscbs,
|
||||
)
|
||||
|
||||
|
||||
def _build_det(item: ItemData, n_item: int) -> _Det:
|
||||
fr = item.fiscal_result
|
||||
icms_linha = _linha_por_dominio(fr, "icms")
|
||||
icmsst_linha = _linha_por_dominio(fr, "icmsst")
|
||||
pis_linha = _linha_por_dominio(fr, "pis")
|
||||
cofins_linha = _linha_por_dominio(fr, "cofins")
|
||||
|
||||
v_prod = (item.quantidade * item.valor_unitario).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
prod = _Det.Prod(
|
||||
cProd=item.codigo,
|
||||
cEAN=item.gtin or _SEM_GTIN,
|
||||
xProd=item.descricao,
|
||||
NCM=item.ncm,
|
||||
CEST=item.cest,
|
||||
CFOP=item.cfop,
|
||||
uCom=item.unidade_comercial,
|
||||
qCom=str(item.quantidade),
|
||||
vUnCom=str(item.valor_unitario),
|
||||
vProd=str(v_prod),
|
||||
cEANTrib=item.gtin or _SEM_GTIN,
|
||||
uTrib=item.unidade_tributavel,
|
||||
qTrib=str(item.quantidade),
|
||||
vUnTrib=str(item.valor_unitario),
|
||||
indTot="1",
|
||||
)
|
||||
imposto = _Det.Imposto(
|
||||
ICMS=_build_icms(fr, icms_linha, icmsst_linha),
|
||||
PIS=_build_pis(pis_linha),
|
||||
COFINS=_build_cofins(cofins_linha),
|
||||
IBSCBS=_build_ibscbs(fr),
|
||||
)
|
||||
return _Det(prod=prod, imposto=imposto, nItem=str(n_item))
|
||||
|
||||
|
||||
def soma_itens_quantizados(itens: list[ItemData]) -> Decimal:
|
||||
"""Fonte ÚNICA do somatório de produtos (I1, review Opus 2026-07-16):
|
||||
soma dos `vProd` POR ITEM já quantizados (2 casas, ROUND_HALF_UP) --
|
||||
exatamente o valor que aparece em cada `<det><prod><vProd>` e, portanto,
|
||||
o único somatório que fecha com `vProd`/`vNF` do `<ICMSTot>`. O `vPag`
|
||||
do orquestrador (`emissao.py`) TEM que derivar daqui também: qualquer
|
||||
outro caminho (ex.: somar `qtd*preço` cru e quantizar só o agregado)
|
||||
diverge em 1 centavo quando um item cai em fração de centavo, e a SEFAZ
|
||||
rejeita com "Valor do Pagamento difere do total"."""
|
||||
total = sum(
|
||||
(item.quantidade * item.valor_unitario).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
for item in itens
|
||||
)
|
||||
return Decimal(total).quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
|
||||
def _sum_valor(itens: list[ItemData], dominio: str) -> Decimal:
|
||||
total = Decimal("0.00")
|
||||
for item in itens:
|
||||
linha = _linha_por_dominio(item.fiscal_result, dominio)
|
||||
if linha is not None:
|
||||
total += linha.valor
|
||||
return total.quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
|
||||
def _sum_base(itens: list[ItemData], dominio: str) -> Decimal:
|
||||
total = Decimal("0.00")
|
||||
for item in itens:
|
||||
linha = _linha_por_dominio(item.fiscal_result, dominio)
|
||||
if linha is not None:
|
||||
total += linha.base_calc
|
||||
return total.quantize(_CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
|
||||
def _build_total(itens: list[ItemData]) -> _InfNfe.Total:
|
||||
v_prod = soma_itens_quantizados(itens)
|
||||
v_icms = _sum_valor(itens, "icms")
|
||||
v_bc = _sum_base(itens, "icms")
|
||||
v_st = _sum_valor(itens, "icmsst")
|
||||
v_bcst = _sum_base(itens, "icmsst")
|
||||
v_pis = _sum_valor(itens, "pis")
|
||||
v_cofins = _sum_valor(itens, "cofins")
|
||||
zero = "0.00"
|
||||
|
||||
icms_tot = _InfNfe.Total.Icmstot(
|
||||
vBC=str(v_bc), vICMS=str(v_icms), vICMSDeson=zero, vFCP=zero,
|
||||
vBCST=str(v_bcst), vST=str(v_st), vFCPST=zero, vFCPSTRet=zero,
|
||||
vProd=str(v_prod), vFrete=zero, vSeg=zero, vDesc=zero,
|
||||
vII=zero, vIPI=zero, vIPIDevol=zero, vPIS=str(v_pis), vCOFINS=str(v_cofins),
|
||||
vOutro=zero, vNF=str(v_prod),
|
||||
)
|
||||
return _InfNfe.Total(ICMSTot=icms_tot)
|
||||
|
||||
|
||||
def _build_ide(dados: DadosEmissao) -> _InfNfe.Ide:
|
||||
id_dest = "1" if dados.uf_destino_tipo == "interna" else "2"
|
||||
return _InfNfe.Ide(
|
||||
cUF=dados.chave_acesso[0:2],
|
||||
cNF=dados.cnf,
|
||||
natOp=dados.nat_op,
|
||||
mod="55",
|
||||
serie=str(dados.serie),
|
||||
nNF=str(dados.numero),
|
||||
dhEmi=dados.dh_emi,
|
||||
tpNF="1",
|
||||
idDest=id_dest,
|
||||
cMunFG=dados.emitente.address_city_ibge_code,
|
||||
tpImp="1",
|
||||
tpEmis=dados.tp_emis,
|
||||
cDV=dados.chave_acesso[-1],
|
||||
tpAmb="2" if dados.ambiente == "homologacao" else "1",
|
||||
finNFe=dados.fin_nfe,
|
||||
indFinal=dados.ind_final,
|
||||
indPres=dados.ind_pres,
|
||||
procEmi="0",
|
||||
verProc=dados.ver_proc,
|
||||
)
|
||||
|
||||
|
||||
def _build_emit(emitente: EmitenteData) -> _InfNfe.Emit:
|
||||
ender = TenderEmi(
|
||||
xLgr=emitente.address_street,
|
||||
nro=emitente.address_number,
|
||||
xCpl=emitente.address_complement,
|
||||
xBairro=emitente.address_district,
|
||||
cMun=emitente.address_city_ibge_code,
|
||||
xMun=emitente.address_city,
|
||||
UF=emitente.address_state,
|
||||
CEP=emitente.address_zip,
|
||||
cPais="1058",
|
||||
xPais="Brasil",
|
||||
fone=emitente.fone,
|
||||
)
|
||||
return _InfNfe.Emit(
|
||||
CNPJ=emitente.cnpj,
|
||||
xNome=emitente.razao_social,
|
||||
xFant=emitente.nome_fantasia,
|
||||
enderEmit=ender,
|
||||
IE=emitente.ie,
|
||||
CRT=emitente.crt,
|
||||
)
|
||||
|
||||
|
||||
def _build_dest(dados: DadosEmissao) -> _InfNfe.Dest | None:
|
||||
dest = dados.destinatario
|
||||
if dest is None:
|
||||
return None
|
||||
|
||||
x_nome = TEXTO_HOMOLOGACAO if dados.ambiente == "homologacao" else dest.nome
|
||||
|
||||
ender = None
|
||||
if dest.address_street is not None:
|
||||
ender = Tendereco(
|
||||
xLgr=dest.address_street,
|
||||
nro=dest.address_number,
|
||||
xCpl=dest.address_complement,
|
||||
xBairro=dest.address_district,
|
||||
cMun=dest.address_city_ibge_code,
|
||||
xMun=dest.address_city,
|
||||
UF=dest.address_state,
|
||||
CEP=dest.address_zip,
|
||||
cPais="1058",
|
||||
xPais="Brasil",
|
||||
)
|
||||
|
||||
return _InfNfe.Dest(
|
||||
CNPJ=dest.cnpj,
|
||||
CPF=dest.cpf,
|
||||
xNome=x_nome,
|
||||
enderDest=ender,
|
||||
indIEDest=dest.indicador_ie,
|
||||
IE=dest.ie,
|
||||
email=dest.email,
|
||||
)
|
||||
|
||||
|
||||
def _build_transp() -> _InfNfe.Transp:
|
||||
# 1b.1: venda de balcão/PDV -- sem transportador/volume rastreado ainda
|
||||
# (Sale/SaleItem não carregam esses dados). modFrete=9 "Sem Ocorrência
|
||||
# de Transporte" é o valor correto para essa realidade, não um chute.
|
||||
return _InfNfe.Transp(modFrete="9")
|
||||
|
||||
|
||||
def _build_pag(pagamento: PagamentoData) -> _InfNfe.Pag:
|
||||
det_pag = _InfNfe.Pag.DetPag(
|
||||
indPag=pagamento.indpag,
|
||||
tPag=pagamento.tpag,
|
||||
vPag=str(pagamento.valor.quantize(_CENT, rounding=ROUND_HALF_UP)),
|
||||
)
|
||||
return _InfNfe.Pag(detPag=[det_pag])
|
||||
|
||||
|
||||
def build_nfe(dados: DadosEmissao) -> Nfe:
|
||||
"""Monta o objeto `Nfe` (nfelib) completo a partir de `DadosEmissao` --
|
||||
PURA: nenhuma chamada de rede/banco, nenhuma decisão de negócio (tudo
|
||||
que precisava de contexto -- CFOP, CST/CSOSN, ambiente, chave -- já
|
||||
veio resolvido no input pelo orquestrador). O `Id` do `infNFe` é
|
||||
`"NFe" + chave_acesso` (44 dígitos), exatamente o formato que
|
||||
`erpbrasil.assinatura` referencia na assinatura (Task 6)."""
|
||||
ide = _build_ide(dados)
|
||||
emit = _build_emit(dados.emitente)
|
||||
dest = _build_dest(dados)
|
||||
dets = [_build_det(item, n) for n, item in enumerate(dados.itens, start=1)]
|
||||
total = _build_total(dados.itens)
|
||||
transp = _build_transp()
|
||||
pag = _build_pag(dados.pagamento)
|
||||
|
||||
inf_nfe = _InfNfe(
|
||||
ide=ide,
|
||||
emit=emit,
|
||||
dest=dest,
|
||||
det=dets,
|
||||
total=total,
|
||||
transp=transp,
|
||||
pag=pag,
|
||||
versao="4.00",
|
||||
Id="NFe" + dados.chave_acesso,
|
||||
)
|
||||
return Nfe(infNFe=inf_nfe)
|
||||
Reference in New Issue
Block a user