diff --git a/src/erpbrasil/edoc/edoc.py b/src/erpbrasil/edoc/edoc.py
index ee839bc..b3080a8 100644
--- a/src/erpbrasil/edoc/edoc.py
+++ b/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
#
diff --git a/src/erpbrasil/edoc/nfce.py b/src/erpbrasil/edoc/nfce.py
index 75a6c60..3407618 100644
--- a/src/erpbrasil/edoc/nfce.py
+++ b/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)
diff --git a/src/erpbrasil/edoc/nfe.py b/src/erpbrasil/edoc/nfe.py
index 67e2731..df034f5 100644
--- a/src/erpbrasil/edoc/nfe.py
+++ b/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
diff --git a/tests/test_erpbrasil_edoc_nfe.py b/tests/test_erpbrasil_edoc_nfe.py
new file mode 100644
index 0000000..87e9260
--- /dev/null
+++ b/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 = """
+
+
+
+ 35
+ 1234567
+ Venda de mercadoria
+
+
+
+ 12345678000195
+ Empresa Exemplo
+
+
+
+
+
+ """
+ prot_nfe_xml_str = """
+
+
+ 1
+ SP_NFE_PL_008i2
+ 12345678901234567890123456789012345678901234
+ 2024-01-16T14:00:00-03:00
+ 13579024681112
+ abcd1234abcd1234abcd1234abcd1234abcd1234=
+ 100
+ Autorizado o uso da NF-e
+
+
+ """
+ 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)