Skip to content

Commit

Permalink
[ADD] MDFe with xsdata bindings
Browse files Browse the repository at this point in the history
  • Loading branch information
@felipezago
felipezago committed Oct 2, 2023
1 parent e2c72d2 commit f8e1670
Showing 1 changed file with 75 additions and 97 deletions.
172 changes: 75 additions & 97 deletions src/erpbrasil/edoc/mdfe.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,34 @@
from erpbrasil.edoc.edoc import DocumentoEletronico

try:
from mdfelib.v3_00 import retConsMDFeNaoEnc
from mdfelib.v3_00 import retConsReciMDFe
from mdfelib.v3_00 import retConsSitMDFe
from mdfelib.v3_00 import retConsStatServMDFe
from mdfelib.v3_00 import retEnviMDFe
from mdfelib.v3_00 import procMDFe
from mdfelib.v3_00 import retEventoMDFe
from mdfelib.v3_00 import evCancMDFe
from mdfelib.v3_00 import evEncMDFe
# Consulta Status
from nfelib.mdfe.bindings.v3_0.cons_stat_serv_mdfe_v3_00 import ConsStatServMdfe
from nfelib.mdfe.bindings.v3_0.ret_cons_stat_serv_mdfe_v3_00 import RetConsStatServMdfe

# Consulta Documento
from nfelib.mdfe.bindings.v3_0.cons_sit_mdfe_v3_00 import ConsSitMdfe
from nfelib.mdfe.bindings.v3_0.ret_cons_sit_mdfe_v3_00 import RetConsSitMdfe

# Consulta Não Encerrados
from nfelib.mdfe.bindings.v3_0.cons_mdfe_nao_enc_v3_00 import ConsMdfeNaoEnc
from nfelib.mdfe.bindings.v3_0.ret_cons_mdfe_nao_enc_v3_00 import RetConsMdfeNaoEnc

# Envio
from nfelib.mdfe.bindings.v3_0.envi_mdfe_v3_00 import EnviMdfe
from nfelib.mdfe.bindings.v3_0.ret_envi_mdfe_v3_00 import RetEnviMdfe

# Consulta Recibo
from nfelib.mdfe.bindings.v3_0.cons_reci_mdfe_v3_00 import ConsReciMdfe
from nfelib.mdfe.bindings.v3_0.ret_cons_reci_mdfe_v3_00 import RetConsReciMdfe

# Processamento
from nfelib.mdfe.bindings.v3_0.proc_mdfe_v3_00 import MdfeProc

# Eventos
from nfelib.mdfe.bindings.v3_0.evento_mdfe_v3_00 import EventoMdfe
from nfelib.mdfe.bindings.v3_0.ret_evento_mdfe_v3_00 import RetEventoMdfe
from nfelib.mdfe.bindings.v3_0.ev_canc_mdfe_v3_00 import EvCancMdfe
from nfelib.mdfe.bindings.v3_0.ev_enc_mdfe_v3_00 import EvEncMdfe
except ImportError:
pass

Expand Down Expand Up @@ -104,6 +123,9 @@ def _verifica_resposta_envio_sucesso(self, proc_envio):
return proc_envio.resposta.cStat == \
self._edoc_situacao_arquivo_recebido_com_sucesso

def _verifica_servico_em_operacao(self, proc_servico):
return proc_servico.resposta.cStat == self._edoc_situacao_servico_em_operacao

def _aguarda_tempo_medio(self, proc_envio):
time.sleep(float(proc_envio.resposta.infRec.tMed) * 1.3)

Expand All @@ -126,49 +148,37 @@ def monta_qrcode_contingencia(self, edoc, xml_assinado):
return f"{self.monta_qrcode(chave)}&sign={digest_value_hex}"

def status_servico(self):
raiz = retConsStatServMDFe.TConsStatServ(
versao=self.versao,
tpAmb=self.ambiente,
cUF=self.uf,
xServ="STATUS",
)
raiz.original_tagname_ = "consStatServMDFe"
return self._post(
raiz,
ConsStatServMdfe(tpAmb=self.ambiente, versao=self.versao),
localizar_url(WS_MDFE_SITUACAO, int(self.ambiente)),
"mdfeStatusServicoMDF" ,
retConsStatServMDFe
RetConsStatServMdfe
)

def consulta_documento(self, chave):
raiz = retConsSitMDFe.TConsSitMDFe(
raiz = ConsSitMdfe(
versao=self.versao,
tpAmb=self.ambiente,
xServ="CONSULTAR",
chMDFe=chave,
)
raiz.original_tagname_ = "consSitMDFe"
return self._post(
raiz,
localizar_url(WS_MDFE_CONSULTA, int(self.ambiente)),
"mdfeConsultaMDF",
retConsSitMDFe
RetConsSitMdfe
)

def consulta_nao_encerrados(self, cnpj):
raiz = retConsMDFeNaoEnc.TConsMDFeNaoEnc(
raiz = ConsMdfeNaoEnc(
versao=self.versao,
tpAmb=self.ambiente,
xServ="CONSULTAR NÃO ENCERRADOS",
CNPJ=cnpj,
)
raiz.original_tagname_ = "consMDFeNaoEnc"

return self._post(
raiz,
localizar_url(WS_MDFE_CONSULTA_NAO_ENCERRADOS, int(self.ambiente)),
"mdfeConsNaoEnc",
retConsMDFeNaoEnc,
RetConsMdfeNaoEnc,
)

def envia_documento(self, edoc):
Expand All @@ -181,23 +191,17 @@ def envia_documento(self, edoc):
:param edoc:
:return:
"""
xml_assinado = self.assina_raiz(edoc, edoc.infMDFe.Id)

raiz = retEnviMDFe.TEnviMDFe(
raiz = EnviMdfe(
versao=self.versao,
idLote=datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
MDFe=edoc
)
raiz.original_tagname_ = "enviMDFe"
xml_envio_string, xml_envio_etree = self._generateds_to_string_etree(
raiz
)
xml_envio_etree.append(etree.fromstring(xml_assinado))

xml_assinado = self.assina_raiz(raiz, edoc.infMDFe.Id)
return self._post(
xml_envio_etree,
xml_assinado,
localizar_url(WS_MDFE_RECEPCAO, int(self.ambiente)),
"mdfeRecepcaoLote",
retEnviMDFe
RetEnviMdfe
)

def consulta_recibo(self, numero=False, proc_envio=False):
Expand All @@ -207,17 +211,16 @@ def consulta_recibo(self, numero=False, proc_envio=False):
if not numero:
return

raiz = retConsReciMDFe.TConsReciMDFe(
raiz = ConsReciMdfe(
versao=self.versao,
tpAmb=self.ambiente,
nRec=numero,
)
raiz.original_tagname_ = "consReciMDFe"
return self._post(
raiz,
localizar_url(WS_MDFE_RET_RECEPCAO, int(self.ambiente)),
"mdfeRetRecepcao",
retConsReciMDFe,
RetConsReciMdfe,
)

def monta_processo(self, edoc, proc_envio, proc_recibo):
Expand All @@ -227,87 +230,62 @@ def monta_processo(self, edoc, proc_envio, proc_recibo):
if type(protocolos) != list:
protocolos = [protocolos]
for protocolo in protocolos:
mdfe_proc = procMDFe.mdfeProc(
versao=self.versao,
protMDFe=protocolo,
)
mdfe_proc.original_tagname_ = 'mdfeProc'
xml_file, mdfe_proc = self._generateds_to_string_etree(mdfe_proc)
mdfe_proc = MdfeProc(versao=self.versao, protMDFe=protocolo)
proc_recibo.processo = mdfe_proc
proc_recibo.processo_xml = \
self._generateds_to_string_etree(mdfe_proc)[0]
proc_recibo.processo_xml = mdfe_proc.to_xml()
proc_recibo.protocolo = protocolo

def send_event(self, evento):
raiz = retEventoMDFe.TEvento(versao="3.00", infEvento=evento)
raiz.original_tagname_ = "eventoMDFe"
def envia_evento(self, evento, tipo, chave, sequencia="001", data_hora=False):
inf_evento = EventoMdfe.InfEvento(
Id="ID" + tipo + chave + sequencia.zfill(2),
cOrgao=self.uf,
tpAmb=self.ambiente,
CNPJ=chave[6:20],
chMDFe=chave,
dhEvento=data_hora or self._hora_agora(),
tpEvento=tipo,
nSeqEvento=sequencia,
detEvento=EventoMdfe.InfEvento.DetEvento(
versaoEvento="3.00",
any_element=evento
),
)
raiz = EventoMdfe(versao="3.00", infEvento=inf_evento)
xml_assinado = etree.fromstring(self.assina_raiz(raiz, raiz.infEvento.Id))

return self._post(
xml_assinado,
localizar_url(WS_MDFE_RECEPCAO_EVENTO, int(self.ambiente)),
"mdfeRecepcaoEvento",
retEventoMDFe
RetEventoMdfe
)

def cancela_documento(self, chave, protocolo_autorizacao, justificativa,
data_hora_evento=False):
tipo = "110111"
sequencia = "1"
data_hora_evento = data_hora_evento or self._hora_agora()

cancelamento = evCancMDFe.evCancMDFe(
evento_canc = EvCancMdfe(
descEvento="Cancelamento",
nProt=protocolo_autorizacao,
xJust=justificativa
)
cancelamento_string, _ = self._generateds_to_string_etree(cancelamento)

raiz = evCancMDFe.infEventoType(
Id="ID" + tipo + chave + sequencia.zfill(2),
cOrgao=self.uf,
tpAmb=self.ambiente,
CNPJ=chave[6:20],
chMDFe=chave,
dhEvento=data_hora_evento,
tpEvento=tipo,
nSeqEvento=sequencia,
detEvento=evCancMDFe.detEventoType(
versaoEvento="3.00",
anytypeobjs_=cancelamento_string
),
return self.envia_evento(
evento=evento_canc,
tipo="110111",
chave=chave,
data_hora=data_hora_evento
)
raiz.original_tagname_ = "infEvento"
return self.send_event(raiz)

def encerra_documento(self, chave, protocolo_autorizacao, estado, municipio,
data_hora_evento=False):
tipo = "110112"
sequencia = "1"
data_hora_evento = data_hora_evento or self._hora_agora()

encerramento = evEncMDFe.evEncMDFe(
encerramento = EvEncMdfe(
descEvento="Encerramento",
dtEnc=self._data_hoje(),
nProt=protocolo_autorizacao,
cUF=estado,
cMun=municipio
)
encerramento_string, _ = self._generateds_to_string_etree(encerramento)

raiz = evEncMDFe.infEventoType(
Id="ID" + tipo + chave + sequencia.zfill(2),
cOrgao=self.uf,
tpAmb=self.ambiente,
CNPJ=chave[6:20],
chMDFe=chave,
dhEvento=data_hora_evento,
tpEvento=tipo,
nSeqEvento=sequencia,
detEvento=evEncMDFe.detEventoType(
versaoEvento="3.00",
anytypeobjs_=encerramento_string
),
return self.envia_evento(
evento=encerramento,
tipo="110112",
chave=chave,
data_hora=data_hora_evento
)
raiz.original_tagname_ = "infEvento"
return self.send_event(raiz)

0 comments on commit f8e1670

Please sign in to comment.