feat: suporte a gIBSCBSMono para CST 620 (Tributacao Monofasica) — NT 2025.002-RTC

docs/reforma_tributaria.md

@@ -37,8 +37,43 @@ A implementacao cobre:
 - `vNF` **NAO inclui** IBS/CBS (proibido em 2025-2026)
 - `finNFe=5` (Nota de Debito) e `finNFe=6` (Nota de Credito)
 - Campos de entidade para IS (Imposto Seletivo) — **armazenados mas nao serializados** ate o schema suportar (2027)
+- Tributacao monofasica (`gIBSCBSMono`) para CST 620 — combustiveis e demais produtos sujeitos ao regime monofasico de IBS/CBS
 
-**Nao inclui** (ainda): Split Payment, cashback, eventos de apuracao assistida, Grupo VB (total do item), Grupo VC (referenciamento de DF-e), Grupo BB (antecipacao de pagamento), tributacao monofasica (`gIBSCBSMono`), diferimento per-item (`gDif`), devolucao de tributos per-item (`gDevTrib`), reducao de aliquota per-item (`gRed`), estorno de credito (`gEstornoCred`), credito presumido per-item (`gCredPresOper`, `gCredPresIBSZFM`).
+**Nao inclui** (ainda): Split Payment, cashback, eventos de apuracao assistida, Grupo VB (total do item), Grupo VC (referenciamento de DF-e), Grupo BB (antecipacao de pagamento), diferimento per-item (`gDif`), devolucao de tributos per-item (`gDevTrib`), reducao de aliquota per-item (`gRed`), estorno de credito (`gEstornoCred`), credito presumido per-item (`gCredPresOper`, `gCredPresIBSZFM`).
+
+### Tributacao monofasica — `gIBSCBSMono`
+
+Para produtos com CST 620 (combustiveis, etc.) o grupo emitido dentro de `<IBSCBS>` e `<gIBSCBSMono>` ao inves de `<gIBSCBS>`. Conforme o schema oficial (`DFeTiposBasicos_v1.00.xsd`, type `TMonofasia`), os cinco campos monofasicos vivem sob o wrapper obrigatorio `<gMonoPadrao>` e na ordem definida pelo schema. Alem disso, `<vTotIBSMonoItem>` e `<vTotCBSMonoItem>` sao SIBLINGS de `<gMonoPadrao>` (NAO filhos) e ambos sao OBRIGATORIOS por schema (sem `minOccurs=0`):
+
+```xml
+<gIBSCBSMono>
+  <gMonoPadrao>
+    <qBCMono>18.0000</qBCMono>
+    <adRemIBS>0.1000</adRemIBS>
+    <adRemCBS>0.0000</adRemCBS>
+    <vIBSMono>1.80</vIBSMono>
+    <vCBSMono>0.00</vCBSMono>
+  </gMonoPadrao>
+  <vTotIBSMonoItem>1.80</vTotIBSMonoItem>
+  <vTotCBSMonoItem>0.00</vTotCBSMonoItem>
+</gIBSCBSMono>
+```
+
+| Campo | Tipo | Descricao |
+|-------|------|-----------|
+| `qBCMono` | TDec1104RTC | Quantidade tributada na base monofasica |
+| `adRemIBS` | TDec_0302_04RTC | Aliquota ad rem IBS (valor em BRL por unidade) |
+| `adRemCBS` | TDec_0302_04RTC | Aliquota ad rem CBS (valor em BRL por unidade) |
+| `vIBSMono` | TDec1302RTC | Valor IBS monofasico |
+| `vCBSMono` | TDec1302RTC | Valor CBS monofasico |
+| `vTotIBSMonoItem` | TDec1302RTC | Total IBS monofasico do item (sibling de `gMonoPadrao`) |
+| `vTotCBSMonoItem` | TDec1302RTC | Total CBS monofasico do item (sibling de `gMonoPadrao`) |
+
+Atributos na entidade `NotaFiscalProduto`: `ibscbs_q_bc_mono`, `ibscbs_ad_rem_ibs`, `ibscbs_v_ibs_mono`, `ibscbs_ad_rem_cbs`, `ibscbs_v_cbs_mono`, `ibscbs_v_tot_ibs_mono_item`, `ibscbs_v_tot_cbs_mono_item`.
+
+Para itens single-line (sem retencao / retencao anterior / diferimento), `vTotIBSMonoItem == vIBSMono` e `vTotCBSMonoItem == vCBSMono`. Quando os atributos nao sao informados pelo caller, o serializador emite `0.00` (default seguro durante o Teste de Carga 2026 com ad rem zerados).
+
+Durante o Teste de Carga 2026 os ad rem ainda nao foram publicados pela SEFAZ, entao os valores podem ser zerados — o grupo `gIBSCBSMono` ainda sera emitido corretamente.
 
 ## CSTs disponiveis
 
@@ -300,13 +335,22 @@ Os totais ficam em um grupo **separado** de `<ICMSTot>`, como irmao dentro de `<
       <vCredPres>0.00</vCredPres>
       <vCredPresCondSus>0.00</vCredPresCondSus>
     </gCBS>
-    <!-- gMono: totais monofasia (nao implementado) -->
+    <gMono>                                <!-- Totais monofasia (DEV-1955) -->
+      <vIBSMono>0.00</vIBSMono>            <!-- Total IBS monofasico padrao -->
+      <vCBSMono>0.00</vCBSMono>            <!-- Total CBS monofasica padrao -->
+      <vIBSMonoReten>0.00</vIBSMonoReten>  <!-- Total IBS monofasico sujeito a retencao -->
+      <vCBSMonoReten>0.00</vCBSMonoReten>  <!-- Total CBS monofasica sujeita a retencao -->
+      <vIBSMonoRet>0.00</vIBSMonoRet>      <!-- Total IBS monofasico retido anteriormente -->
+      <vCBSMonoRet>0.00</vCBSMonoRet>      <!-- Total CBS monofasica retida anteriormente -->
+    </gMono>
     <!-- gEstornoCred: totais estorno de credito (nao implementado) -->
   </IBSCBSTot>
 </total>
 ```
 
 > Os subgrupos `gIBS` e `gCBS` sao opcionais (`minOccurs="0"`) — emitidos apenas quando ha valores. Os campos `vDif`, `vDevTrib`, `vCredPres` e `vCredPresCondSus` sao obrigatorios dentro de cada subgrupo (emitidos como "0.00" quando nao utilizados).
+>
+> O subgrupo `gMono` e opcional, mas obrigatorio sempre que algum item da NF-e carregar `<gIBSCBSMono>` (CST 620, etc.). Quando emitido, **todos os seis filhos** sao obrigatorios (`vIBSMono`, `vCBSMono`, `vIBSMonoReten`, `vCBSMonoReten`, `vIBSMonoRet`, `vCBSMonoRet`). Omitir `<gMono>` em uma NF-e com items monofasicos faz a SEFAZ rejeitar com `cStat 1119 - "Total de IBS e CBS nao informado"`. Os totais `Reten`/`Ret` ainda nao sao acumulados a nivel de item (PyNFe ainda so emite `<gMonoPadrao>`), entao serao "0.00" ate que `<gMonoReten>` / `<gMonoRet>` / `<gMonoDif>` sejam suportados a nivel de item.
 
 ### Cabecalho — `cMunFGIBS` no `<ide>`
 
@@ -338,7 +382,7 @@ Esses CSTs geram apenas `<CST>` e `<cClassTrib>`, sem `<gIBSCBS>`.
 
 - **`cClassTrib`**: Emitido quando informado (campo obrigatorio na pratica)
 - **`cMunFGIBS`**: Emitido no `<ide>` apenas quando informado
-- **`<IBSCBSTot>`**: Tipo `TIBSCBSMonoTot`. Omitido se todos os totais forem zero. Quando emitido, `vBCIBSCBS` e obrigatorio como primeiro filho; `gIBS` e `gCBS` sao opcionais
+- **`<IBSCBSTot>`**: Tipo `TIBSCBSMonoTot`. Omitido se todos os totais forem zero E nenhum item carregar `<gIBSCBSMono>`. Quando emitido, `vBCIBSCBS` e obrigatorio como primeiro filho; `gIBS`, `gCBS` e `gMono` sao opcionais (mas `gMono` e obrigatorio sempre que houver items monofasicos)
 - **`<IBSCBS>`**: Tipo `TTribNFe`. Omitido completamente se `ibscbs_cst` nao for informado
 - **IS (`<IS>`)**: Tipo `TIS`. **Nao emitido no XML** — serializacao desabilitada ate 2027
 - **`<ISTot>`**: Tipo `TISTot`. **Nao emitido** — sera irmao de `<IBSCBSTot>` (antes dele no schema)

pynfe/entidades/notafiscal.py

@@ -10,6 +10,12 @@
 
 from .base import CampoDeprecated, Entidade
 
+# CSTs IBS/CBS that route to <gIBSCBSMono> instead of <gIBSCBS>. Mirrors
+# the constant in pynfe.processamento.serializacao but defined here so the
+# NotaFiscal accumulator can detect monofasic items without importing the
+# serializer. Keep in sync with SerializacaoXML._IBSCBS_CST_MONOFASICO.
+_IBSCBS_CST_MONOFASICO = ("620",)
+
 
 class NotaFiscal(Entidade):
     # campos deprecados
@@ -310,6 +316,24 @@ class NotaFiscal(Entidade):
     totais_cbs = Decimal()
     totais_is = Decimal()
 
+    # Reforma Tributaria - Totais Monofasia (Group W03 - IBSCBSTot/gMono)
+    # Per NT 2025.002-RTC schema TIBSCBSMonoTot, when ANY item carries
+    # <gIBSCBSMono>, the IBSCBSTot wrapper MUST emit a <gMono> sibling with
+    # six TDec1302RTC values (all REQUIRED inside <gMono>): vIBSMono,
+    # vCBSMono, vIBSMonoReten, vCBSMonoReten, vIBSMonoRet, vCBSMonoRet.
+    # SEFAZ rejects with cStat 1119 ("Total de IBS e CBS nao informado")
+    # when items emit <gIBSCBSMono> but the totalizer omits <gMono>.
+    # ``totais_mono_item_count`` tracks the number of monofasic items so
+    # the serializer can decide to emit <gMono> even when all six values
+    # are zero (Teste de Carga 2026 scenario, where every ad rem is 0).
+    totais_v_ibs_mono = Decimal()  # sum of vTotIBSMonoItem (gMonoPadrao)
+    totais_v_cbs_mono = Decimal()  # sum of vTotCBSMonoItem (gMonoPadrao)
+    totais_v_ibs_mono_reten = Decimal()  # sum of vTotIBSMonoItem (gMonoReten)
+    totais_v_cbs_mono_reten = Decimal()  # sum of vTotCBSMonoItem (gMonoReten)
+    totais_v_ibs_mono_ret = Decimal()  # sum of vTotIBSMonoItem (gMonoRet)
+    totais_v_cbs_mono_ret = Decimal()  # sum of vTotCBSMonoItem (gMonoRet)
+    totais_mono_item_count = int()
+
     # Reforma Tributaria - cMunFGIBS (Group B)
     municipio_fato_gerador_ibs = str()
 
@@ -483,6 +507,21 @@ def adicionar_produto_servico(self, **kwargs):
         self.totais_cbs += obj.ibscbs_v_cbs
         self.totais_is += obj.is_valor
 
+        # Reforma Tributaria - Totais Monofasia (IBSCBSTot/gMono per NT 2025.002-RTC)
+        # Aggregate item-level vTotIBSMonoItem / vTotCBSMonoItem so the
+        # serializer can emit <gMono> with the required totals. PyNFe today
+        # only supports <gMonoPadrao> at item level (see _serializar_gibscbs_mono),
+        # so the Reten/Ret accumulators stay at zero, but the schema requires
+        # all six fields to be emitted whenever <gMono> is present, so they
+        # are kept here to align with TIBSCBSMonoTot/gMono.
+        # ``totais_mono_item_count`` is incremented when CST routes to mono,
+        # giving the serializer a stable signal to emit <gMono> even if every
+        # value is zero (Teste de Carga 2026 scenario).
+        self.totais_v_ibs_mono += obj.ibscbs_v_tot_ibs_mono_item
+        self.totais_v_cbs_mono += obj.ibscbs_v_tot_cbs_mono_item
+        if obj.ibscbs_cst in _IBSCBS_CST_MONOFASICO:
+            self.totais_mono_item_count += 1
+
         # TODO calcular impostos aproximados
         # self.totais_tributos_aproximado += obj.tributos
 
@@ -1046,6 +1085,23 @@ class NotaFiscalProduto(Entidade):
     ibscbs_p_cbs = Decimal()  # pCBS
     ibscbs_v_cbs = Decimal()  # vCBS
 
+    # gIBSCBSMono - Tributacao monofasica (CST 620)
+    # Emitted as <gIBSCBSMono> instead of <gIBSCBS> for CST 620 items.
+    # qBCMono = quantity in monophasic base unit (TDec_1104v)
+    # adRemIBS / adRemCBS = ad rem rate in BRL per unit (TDec_0302a10)
+    # vIBSMono / vCBSMono = final value in BRL
+    # vTotIBSMonoItem / vTotCBSMonoItem = item-level mono totals (TDec1302RTC). Per
+    # NT 2025.002-RTC schema TMonofasia, these are REQUIRED siblings of
+    # <gMonoPadrao> inside <gIBSCBSMono> (NOT children of <gMonoPadrao>). For a
+    # single-line item without retencao/diferimento, they equal vIBSMono/vCBSMono.
+    ibscbs_q_bc_mono = Decimal()  # qBCMono
+    ibscbs_ad_rem_ibs = Decimal()  # adRemIBS
+    ibscbs_v_ibs_mono = Decimal()  # vIBSMono
+    ibscbs_ad_rem_cbs = Decimal()  # adRemCBS
+    ibscbs_v_cbs_mono = Decimal()  # vCBSMono
+    ibscbs_v_tot_ibs_mono_item = Decimal()  # vTotIBSMonoItem
+    ibscbs_v_tot_cbs_mono_item = Decimal()  # vTotCBSMonoItem
+
     # IS (Imposto Seletivo) - Group UB-IS
     is_cst_selec = str()  # CSTSelec (2-digit)
     is_c_class_trib = str()  # cClassTribIS 6-digit

pynfe/processamento/comunicacao.py

@@ -1,5 +1,7 @@
 # -*- coding: utf-8 -*-
+import base64
 import datetime
+import gzip
 import re
 
 import requests
@@ -876,39 +878,34 @@ def autorizacao(self, manifesto, id_lote=1, ind_sinc=1):
         Método para realizar autorização do manifesto
         :param manifesto: XML assinado
         :param id_lote: Id do lote - numero autoincremental gerado pelo sistema
+            (usado apenas no envio assincrono)
         :param ind_sinc: Indicador de sincrono e assincrono, 0 para assincrono, 1 para sincrono
         :return:  Uma tupla que em caso de sucesso,
             retorna xml com manifesto e protocolo de autorização.
-        Caso contrário, envia todo o soap de resposta da Sefaz para decisão do usuário.
+        Caso contrário, retorna o retMDFe parseado (quando a SEFAZ respondeu) ou a
+        resposta HTTP crua para decisão do usuário.
         """
-        # url do serviço
-        if ind_sinc == 0:
-            url = self._get_url(consulta="RECEPCAO")
-        elif ind_sinc == 1:
-            url = self._get_url(consulta="RECEPCAO_SINC")
-        else:
-            raise "ind_sinc deve ser 0=assincrono ou 1=sincrono"
+        if ind_sinc == 1:
+            return self._autorizacao_sincrona(manifesto)
+        if ind_sinc != 0:
+            raise ValueError("ind_sinc deve ser 0=assincrono ou 1=sincrono")
+
+        # Envio assincrono: lote enviMDFe para o servico MDFeRecepcao
+        url = self._get_url(consulta="RECEPCAO")
 
         # Monta XML do corpo da requisição
         raiz = etree.Element("enviMDFe", xmlns=NAMESPACE_MDFE, versao=VERSAO_MDFE)
         etree.SubElement(raiz, "idLote").text = str(
             id_lote
         )  # numero autoincremental gerado pelo sistema
-        # etree.SubElement(raiz, 'indSinc').text = str(ind_sinc)
-        # # 0 para assincrono, 1 para sincrono
         raiz.append(manifesto)
 
         # Monta XML para envio da requisição
-        if ind_sinc == 0:
-            xml = self._construir_xml_soap("MDFeRecepcao", raiz)
-        elif ind_sinc == 1:
-            xml = self._construir_xml_soap("MDFeRecepcaoSinc", raiz)
+        xml = self._construir_xml_soap("MDFeRecepcao", raiz)
 
         # Faz request no Servidor da Sefaz
         retorno = self._post(url, xml)
 
-        # Em caso de sucesso, retorna xml com o mdfe e protocolo de autorização.
-        # Caso contrário, envia todo o soap de resposta da Sefaz para decisão do usuário.
         if retorno.status_code == 200:
             # namespace
             ns = {"ns": NAMESPACE_MDFE}
@@ -919,41 +916,59 @@ def autorizacao(self, manifesto, id_lote=1, ind_sinc=1):
                 # em SP retorno.text apresenta erro
                 prot = etree.fromstring(retorno.content)
 
-            if ind_sinc == 1:
-                try:
-                    # Protocolo com envio OK
-                    inf_prot = prot[1][0]
-                    lote_status = inf_prot.xpath("ns:retEnviMDFe/ns:cStat", namespaces=ns)[0].text
+            # Retorna id do protocolo para posterior consulta em caso de sucesso.
+            rec = prot[1][0]
+            status = rec.xpath("ns:retEnviMDFe/ns:cStat", namespaces=ns)[0].text
+            # Lote Recebido com Sucesso!
+            if status in (
+                self._edoc_situacao_arquivo_recebido_com_sucesso,
+                self._edoc_situacao_em_processamento,
+            ):
+                nrec = rec.xpath("ns:retEnviMDFe/ns:infRec/ns:nRec", namespaces=ns)[0].text
+                return 0, nrec, manifesto
+        return 1, retorno, manifesto
 
-                    # Lote processado
-                    if lote_status == self._edoc_situacao_lote_processado:
-                        prot_mdfe = inf_prot.xpath("ns:retEnviMDFe/ns:protMDFe", namespaces=ns)[0]
-                        status = prot_mdfe.xpath("ns:infProt/ns:cStat", namespaces=ns)[0].text
+    def _autorizacao_sincrona(self, manifesto):
+        """Envia o MDF-e para o servico sincrono MDFeRecepcaoSinc.
 
-                        # autorizado uso do MDF-e
-                        # retorna xml final (protMDFe + MDFe)
-                        if status in self._edoc_situacao_ja_enviado:  # if status == '100':
-                            raiz = etree.Element(
-                                "mdfeProc", xmlns=NAMESPACE_MDFE, versao=VERSAO_MDFE
-                            )
-                            raiz.append(manifesto)
-                            raiz.append(prot_mdfe)
-                            return 0, raiz
-                except IndexError:
-                    # Protocolo com algum erro no Envio
-                    return 1, retorno, manifesto
-            else:
-                # Retorna id do protocolo para posterior consulta em caso de sucesso.
-                rec = prot[1][0]
-                status = rec.xpath("ns:retEnviMDFe/ns:cStat", namespaces=ns)[0].text
-                # Lote Recebido com Sucesso!
-                if status in (
-                    self._edoc_situacao_arquivo_recebido_com_sucesso,
-                    self._edoc_situacao_em_processamento,
-                ):
-                    nrec = rec.xpath("ns:retEnviMDFe/ns:infRec/ns:nRec", namespaces=ns)[0].text
-                    return 0, nrec, manifesto
-        return 1, retorno, manifesto
+        Conforme o MOC do MDF-e, o servico de recepcao sincrona recebe o XML
+        assinado do <MDFe> compactado com gzip e codificado em base64 como
+        conteudo texto de mdfeDadosMsg — sem o envelope <enviMDFe>/<idLote>
+        usado no envio assincrono.
+        """
+        url = self._get_url(consulta="RECEPCAO_SINC")
+
+        mdfe_compactado = base64.b64encode(gzip.compress(etree.tostring(manifesto))).decode("ascii")
+        xml = self._construir_xml_soap("MDFeRecepcaoSinc", mdfe_compactado)
+
+        retorno = self._post(url, xml, content_type=b"application/soap+xml; charset=utf-8;")
+
+        if retorno.status_code != 200:
+            return 1, retorno, manifesto
+
+        ns = {"ns": NAMESPACE_MDFE}
+        try:
+            envelope = etree.fromstring(retorno.content)
+            ret_mdfe = envelope.xpath("//ns:retMDFe", namespaces=ns)[0]
+        except (ValueError, IndexError, etree.XMLSyntaxError):
+            # Resposta nao parseavel: devolve a resposta HTTP crua
+            return 1, retorno, manifesto
+
+        # Rejeicao pode vir no nivel do retMDFe (ex.: 580, sem protMDFe)
+        # ou no nivel do protMDFe/infProt.
+        prot_mdfe_list = ret_mdfe.xpath("ns:protMDFe", namespaces=ns)
+        if prot_mdfe_list:
+            prot_mdfe = prot_mdfe_list[0]
+            status = prot_mdfe.xpath("ns:infProt/ns:cStat", namespaces=ns)
+            # autorizado uso do MDF-e: retorna xml final (MDFe + protMDFe)
+            if status and status[0].text in self._edoc_situacao_ja_enviado:
+                raiz = etree.Element("mdfeProc", xmlns=NAMESPACE_MDFE, versao=VERSAO_MDFE)
+                raiz.append(manifesto)
+                raiz.append(prot_mdfe)
+                return 0, raiz
+
+        # Devolve o retMDFe parseado para o chamador consultar cStat/xMotivo
+        return 1, ret_mdfe, manifesto
 
     def status_servico(self):
         url = self._get_url("STATUS")
@@ -1037,16 +1052,17 @@ def _construir_xml_soap(self, metodo, dados):
 
         a = etree.SubElement(body, self._envio_mensagem, xmlns=self._namespace_metodo + metodo)
 
-        # if metodo == 'MDFeRecepcaoSinc':
-        #     body_base64 = base64.b16encode(a).decode()
-
-        a.append(dados)
+        if isinstance(dados, str):
+            # payload compactado (gzip+base64) vai como texto de mdfeDadosMsg
+            a.text = dados
+        else:
+            a.append(dados)
         return raiz
 
-    def _post_header(self, soap_webservice_method=False):
+    def _post_header(self, soap_webservice_method=False, content_type=None):
         """Retorna um dicionário com os atributos para o cabeçalho da requisição HTTP"""
         header = {
-            b"content-type": b"text/xml; charset=utf-8;",
+            b"content-type": content_type or b"text/xml; charset=utf-8;",
         }
 
         # PE é a únca UF que exige SOAPAction no header
@@ -1060,7 +1076,7 @@ def _post_header(self, soap_webservice_method=False):
 
         return header
 
-    def _post(self, url, xml):
+    def _post(self, url, xml, content_type=None):
         certificado_a1 = CertificadoA1(self.certificado)
         chave, cert = certificado_a1.separar_arquivo(self.certificado_senha, caminho=True)
         chave_cert = (cert, chave)
@@ -1084,7 +1100,7 @@ def _post(self, url, xml):
             result = requests.post(
                 url,
                 xml,
-                headers=self._post_header(),
+                headers=self._post_header(content_type=content_type),
                 cert=chave_cert,
                 verify=False,
                 timeout=50,