Melhorias para NF-e

src/erpbrasil/edoc/edoc.py

@@ -34,8 +34,9 @@ class DocumentoEletronico(ABC):
     _consulta_servico_ao_enviar = False
     _consulta_documento_antes_de_enviar = False
 
-    def __init__(self, transmissao):
+    def __init__(self, transmissao, envio_sincrono=False):
         self._transmissao = transmissao
+        self.envio_sincrono = bool(envio_sincrono)
 
     def _generateds_to_string_etree(self, ds, pretty_print=False):
         if type(ds) == _Element:
@@ -68,7 +69,7 @@ def _post(self, raiz, url, operacao, classe):
             retorno = self._transmissao.enviar(operacao, xml_etree)
             return analisar_retorno_raw(operacao, raiz, xml_string, retorno, classe)
 
-    def processar_documento(self, edoc):
+    def processar_documento(self, edoc, envio_sincrono=False):
         """Processar documento executa o envio do documento fiscal de forma
         completa ao serviço relacionado, esta é um método padrão que
         segue o seguinte workflow:
@@ -132,18 +133,19 @@ def processar_documento(self, edoc):
         #
 
         proc_envio = self.envia_documento(edoc)
+        if self.envio_sincrono:
+            self.monta_processo(edoc, proc_envio)
         yield proc_envio
 
-        #
-        # Deu errado?
-        #
-        if not proc_envio.resposta:
-            return
-
-        if not self._verifica_resposta_envio_sucesso(proc_envio):
-            #
-            # Interrompe o processo
-            #
+        # Retorna imediatamente se alguma das condições abaixo for verdadeira:
+        # 1. A resposta do processo de envio é falsa.
+        # 2. A resposta do envio não indica sucesso.
+        # 3. O envio é síncrono (não é necessário consultar o recibo).
+        if (
+            not proc_envio.resposta
+            or not self._verifica_resposta_envio_sucesso(proc_envio)
+            or self.envio_sincrono
+        ):
             return
 
         #

src/erpbrasil/edoc/nfce.py

@@ -258,9 +258,9 @@ def __init__(
         qrcode_versao="2",
         csc_token=None,
         csc_code=None,
+        envio_sincrono=True,
     ):
-        super().__init__(transmissao, uf, versao, ambiente)
-        self.mod = str(mod)
+        super().__init__(transmissao, uf, versao, ambiente, mod, envio_sincrono)
         self.qrcode_versao = str(qrcode_versao)
         self.csc_token = str(csc_token)
         self.csc_code = str(csc_code)
@@ -333,7 +333,7 @@ def envia_documento(self, edoc):
         raiz = retEnviNFe.TEnviNFe(
             versao=self.versao,
             idLote=datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
-            indSinc="1",
+            indSinc="1" if self.envio_sincrono else "0",
         )
         raiz.original_tagname_ = "enviNFe"
         xml_envio_string, xml_envio_etree = self._generateds_to_string_etree(raiz)

src/erpbrasil/edoc/nfe.py

@@ -717,13 +717,25 @@ def localizar_url(servico, estado, mod="55", ambiente=2):
 class NFe(DocumentoEletronico):
     _namespace = "http://www.portalfiscal.inf.br/nfe"
     _edoc_situacao_arquivo_recebido_com_sucesso = "103"
+    _edoc_situacao_arquivo_processado_com_sucesso = "104"
     _edoc_situacao_servico_em_operacao = "107"
-    _consulta_servico_ao_enviar = True
-    _consulta_documento_antes_de_enviar = True
+
+    # Desativado por padrão para evitar 'consumo indevido'
+    _consulta_servico_ao_enviar = False
+    _consulta_documento_antes_de_enviar = False
+
     _maximo_tentativas_consulta_recibo = 5
 
-    def __init__(self, transmissao, uf, versao="4.00", ambiente="2", mod="55"):
-        super().__init__(transmissao)
+    def __init__(
+        self,
+        transmissao,
+        uf,
+        versao="4.00",
+        ambiente="2",
+        mod="55",
+        envio_sincrono=False,
+    ):
+        super().__init__(transmissao, envio_sincrono)
         self.versao = str(versao)
         self.ambiente = str(ambiente)
         self.uf = int(uf)
@@ -754,6 +766,7 @@ def status_servico(self):
         )
 
     def consulta_documento(self, chave):
+        # NfeConsultaProtocolo
         raiz = retConsSitNFe.TConsSitNFe(
             versao=self.versao,
             tpAmb=self.ambiente,
@@ -784,14 +797,11 @@ def envia_documento(self, edoc):
         raiz = retEnviNFe.TEnviNFe(
             versao=self.versao,
             idLote=datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
-            indSinc="0",
+            indSinc="1" if self.envio_sincrono else "0",
         )
         raiz.original_tagname_ = "enviNFe"
         xml_envio_string, xml_envio_etree = self._generateds_to_string_etree(raiz)
         xml_envio_etree.append(etree.fromstring(xml_assinado))
-
-        # teste_string, teste_etree = self._generateds_to_string_etree(xml_envio_etree)
-
         return self._post(
             xml_envio_etree,
             # 'https://hom.sefazvirtual.fazenda.gov.br/NFeAutorizacao4/NFeAutorizacao4.asmx?wsdl',
@@ -953,15 +963,18 @@ def _verifica_documento_ja_enviado(self, proc_consulta):
         return False
 
     def _verifica_resposta_envio_sucesso(self, proc_envio):
-        if (
-            proc_envio.resposta.cStat
-            == self._edoc_situacao_arquivo_recebido_com_sucesso
-        ):
-            return True
-        return False
+        """
+        Verifica se a resposta do envio indica sucesso:
+        - cStat "103" = "Lote recebido com sucesso" (assíncrono)
+        - cStat "104" = "Lote processado com sucesso" (síncrono)
+        """
+        return proc_envio.resposta.cStat in [
+            self._edoc_situacao_arquivo_recebido_com_sucesso,
+            self._edoc_situacao_arquivo_processado_com_sucesso,
+        ]
 
     def _aguarda_tempo_medio(self, proc_envio):
-        time.sleep(float(proc_envio.resposta.infRec.tMed) * 1.3)
+        time.sleep(float(proc_envio.resposta.infRec.tMed))
 
     def _edoc_situacao_em_processamento(self, proc_recibo):
         if proc_recibo.resposta.cStat == "105":
@@ -1017,9 +1030,14 @@ def consultar_distribuicao(
             retDistDFeInt,
         )
 
-    def monta_processo(self, edoc, proc_envio, proc_recibo):
+    def monta_processo(self, edoc, proc_envio, proc_recibo=None):
         nfe = proc_envio.envio_raiz.find("{" + self._namespace + "}NFe")
-        protocolos = proc_recibo.resposta.protNFe
+        if proc_recibo:
+            protocolos = proc_recibo.resposta.protNFe
+        else:
+            # A falta do recibo indica envio no modo síncrono
+            # o protocolo é recuperado diretamente da resposta do envio.
+            protocolos = proc_envio.resposta.protNFe
         if len(nfe) and protocolos:
             if not isinstance(protocolos, list):
                 protocolos = [protocolos]
@@ -1032,11 +1050,27 @@ def monta_processo(self, edoc, proc_envio, proc_recibo):
                 xml_file, nfe_proc = self._generateds_to_string_etree(nfe_proc)
                 prot_nfe = nfe_proc.find("{" + self._namespace + "}protNFe")
                 prot_nfe.addprevious(nfe)
-                proc_recibo.processo = nfe_proc
-                proc_recibo.processo_xml = self._generateds_to_string_etree(nfe_proc)[0]
-                proc_recibo.protocolo = protocolo
+
+                proc = proc_recibo if proc_recibo else proc_envio
+                proc.processo = nfe_proc
+                proc.processo_xml = self._generateds_to_string_etree(nfe_proc)[0]
+                proc.protocolo = protocolo
             return True
 
+    def monta_nfe_proc(self, nfe, prot_nfe):
+        """
+        Constrói e retorna o XML do processo da NF-e,
+        incorporando a NF-e com o seu protocolo de autorização.
+        """
+        nfe_proc = etree.Element(
+            f"{{{self._namespace}}}nfeProc",
+            versao=self.versao,
+            nsmap={None: self._namespace},
+        )
+        nfe_proc.append(nfe)
+        nfe_proc.append(prot_nfe)
+        return etree.tostring(nfe_proc)
+
     def consultar_cadastro(self, uf, cnpj=None, cpf=None, ie=None):
         if not cnpj and not cpf and not ie:
             return

tests/test_erpbrasil_edoc_nfe.py

@@ -0,0 +1,56 @@
+from unittest import TestCase
+
+from erpbrasil.edoc.nfe import NFe
+from lxml import etree
+
+
+class NFeTests(TestCase):
+    def setUp(self):
+        self.nfe = NFe(False, "35", versao="4.00", ambiente="1")
+        nfe_xml_str = """
+        <NFe xmlns="http://www.portalfiscal.inf.br/nfe">
+            <infNFe Id="NFe12345678901234567890123456789012345678901234" versao="4.00">
+                <ide>
+                    <cUF>35</cUF>
+                    <cNF>1234567</cNF>
+                    <natOp>Venda de mercadoria</natOp>
+                    <!-- Outros campos da tag ide -->
+                </ide>
+                <emit>
+                    <CNPJ>12345678000195</CNPJ>
+                    <xNome>Empresa Exemplo</xNome>
+                    <!-- Outros campos da tag emit -->
+                </emit>
+                <!-- Outras tags como det, total, transp, etc. -->
+            </infNFe>
+        </NFe>
+        """
+        prot_nfe_xml_str = """
+        <protNFe xmlns="http://www.portalfiscal.inf.br/nfe" versao="4.00">
+            <infProt>
+                <tpAmb>1</tpAmb>
+                <verAplic>SP_NFE_PL_008i2</verAplic>
+                <chNFe>12345678901234567890123456789012345678901234</chNFe>
+                <dhRecbto>2024-01-16T14:00:00-03:00</dhRecbto>
+                <nProt>13579024681112</nProt>
+                <digVal>abcd1234abcd1234abcd1234abcd1234abcd1234=</digVal>
+                <cStat>100</cStat>
+                <xMotivo>Autorizado o uso da NF-e</xMotivo>
+            </infProt>
+        </protNFe>
+        """
+        self.nfe_element = etree.fromstring(nfe_xml_str)
+        self.prot_nfe_element = etree.fromstring(prot_nfe_xml_str)
+
+    def test_monta_nfe_proc(self):
+        nfe_proc_bytes = self.nfe.monta_nfe_proc(
+            self.nfe_element, self.prot_nfe_element
+        )
+        root = etree.fromstring(nfe_proc_bytes)
+        self.assertIsInstance(nfe_proc_bytes, bytes)
+        self.assertEqual(root.tag, "{http://www.portalfiscal.inf.br/nfe}nfeProc")
+        children = list(root)
+        self.assertEqual(len(children), 2)
+        child_tags = [child.tag for child in children]
+        self.assertIn("{http://www.portalfiscal.inf.br/nfe}NFe", child_tags)
+        self.assertIn("{http://www.portalfiscal.inf.br/nfe}protNFe", child_tags)