Skip to content

Commit

Permalink
Merge pull request #66 from kmee/fix/mdfe
Browse files Browse the repository at this point in the history 
[NEW] MDFe implementation
  • Loading branch information
@ygcarvalh
ygcarvalh authored Oct 2, 2023
2 parents 914bb8b + f8e1670 commit 2ea4d7d
Show file tree
Hide file tree
Showing 2 changed files with 227 additions and 65 deletions.
3 changes: 3 additions & 0 deletions src/erpbrasil/edoc/edoc.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,9 @@ def _hora_agora(self):
datetime.now(tz=timezone(timedelta(hours=-3))), FORMAT
) + str(timezone(timedelta(hours=-3)))[3:]

def _data_hoje(self):
return datetime.strftime(datetime.now(), "%Y-%m-%d")

def assina_raiz(self, raiz, id, getchildren=False):
xml_string, xml_etree = self._generateds_to_string_etree(raiz)
xml_assinado = Assinatura(self._transmissao.certificado).assina_xml2(
Expand Down
289 changes: 224 additions & 65 deletions src/erpbrasil/edoc/mdfe.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,80 +6,179 @@
from __future__ import unicode_literals

import datetime
import time
import binascii
import base64

from erpbrasil.assinatura.assinatura import Assinatura
from lxml import etree

from erpbrasil.edoc.edoc import DocumentoEletronico

try:
from mdfelib.v3_00 import consMDFeNaoEnc
from mdfelib.v3_00 import consReciMDFe
from mdfelib.v3_00 import consSitMDFe
from mdfelib.v3_00 import consStatServMDFe
from mdfelib.v3_00 import enviMDFe
# 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

WS_MDFE_CONSULTA = "MDFeConsulta"
WS_MDFE_SITUACAO = "MDFeStatusServico"
WS_MDFE_CONSULTA_NAO_ENCERRADOS = "MDFeConsNaoEnc"
WS_MDFE_DISTRIBUICAO = "MDFeDistribuicaoDFe"

WS_MDFE_RECEPCAO = "MDFeRecepcao"
WS_MDFE_RECEPCAO_SINC = "MDFeRecepcaoSinc"
WS_MDFE_RET_RECEPCAO = "MDFeRetRecepcao"
WS_MDFE_RECEPCAO_EVENTO = "MDFeRecepcaoEvento"

AMBIENTE_PRODUCAO = 1
AMBIENTE_HOMOLOGACAO = 2

MDFE_MODELO = "58"

SVC_RS = {
AMBIENTE_PRODUCAO: {
"servidor": "mdfe.svrs.rs.gov.br",
WS_MDFE_RECEPCAO: "ws/MDFeRecepcao/MDFeRecepcao.asmx?wsdl",
WS_MDFE_RET_RECEPCAO: "ws/MDFeRetRecepcao/MDFeRetRecepcao.asmx?wsdl",
WS_MDFE_RECEPCAO_EVENTO: "ws/MDFeRecepcaoEvento/MDFeRecepcaoEvento.asmx?wsdl",
WS_MDFE_CONSULTA: "ws/MDFeConsulta/MDFeConsulta.asmx?wsdl",
WS_MDFE_SITUACAO: "ws/MDFeStatusServico/MDFeStatusServico.asmx?wsdl",
WS_MDFE_CONSULTA_NAO_ENCERRADOS: "ws/MDFeConsNaoEnc/MDFeConsNaoEnc.asmx?wsdl",
WS_MDFE_DISTRIBUICAO: "ws/MDFeDistribuicaoDFe/MDFeDistribuicaoDFe.asmx?wsdl",
WS_MDFE_RECEPCAO_SINC: "ws/MDFeRecepcaoSinc/MDFeRecepcaoSinc.asmx?wsdl",
},
AMBIENTE_HOMOLOGACAO: {
"servidor": "mdfe-homologacao.svrs.rs.gov.br",
WS_MDFE_RECEPCAO: "ws/MDFeRecepcao/MDFeRecepcao.asmx?wsdl",
WS_MDFE_RET_RECEPCAO: "ws/MDFeRetRecepcao/MDFeRetRecepcao.asmx?wsdl",
WS_MDFE_RECEPCAO_EVENTO: "ws/MDFeRecepcaoEvento/MDFeRecepcaoEvento.asmx?wsdl",
WS_MDFE_CONSULTA: "ws/MDFeConsulta/MDFeConsulta.asmx?wsdl",
WS_MDFE_SITUACAO: "ws/MDFeStatusServico/MDFeStatusServico.asmx?wsdl",
WS_MDFE_CONSULTA_NAO_ENCERRADOS: "ws/MDFeConsNaoEnc/MDFeConsNaoEnc.asmx?wsdl",
WS_MDFE_DISTRIBUICAO: "ws/MDFeDistribuicaoDFe/MDFeDistribuicaoDFe.asmx?wsdl",
WS_MDFE_RECEPCAO_SINC: "ws/MDFeRecepcaoSinc/MDFeRecepcaoSinc.asmx?wsdl",
}
}

QR_CODE_URL = "https://dfe-portal.svrs.rs.gov.br/mdfe/qrCode"

NAMESPACES = {
"mdfe": "http://www.portalfiscal.inf.br/mdfe",
"ds": "http://www.w3.org/2000/09/xmldsig#",
}

def localizar_url(servico, ambiente=2):
dominio = SVC_RS[ambiente]["servidor"]
complemento = SVC_RS[ambiente][servico]

return "https://%s/%s" % (dominio, complemento)


class MDFe(DocumentoEletronico):
_namespace = 'http://www.portalfiscal.inf.br/mdfe'
_namespace = "http://www.portalfiscal.inf.br/mdfe"

_edoc_situacao_arquivo_recebido_com_sucesso = '103'
_edoc_situacao_em_processamento = '105'
_edoc_situacao_servico_em_operacao = '107'
_edoc_situacao_ja_enviado = ('100', '101', '132')
_edoc_situacao_arquivo_recebido_com_sucesso = "103"
_edoc_situacao_servico_em_operacao = "107"
_edoc_situacao_ja_enviado = ("100", "101", "132")

_consulta_servico_ao_enviar = True
_maximo_tentativas_consulta_recibo = 5

def __init__(self, transmissao, uf, versao="3.00", ambiente="2",
mod="58"):
super(MDFe, self).__init__(transmissao)
self.versao = str(versao)
self.ambiente = str(ambiente)
self.uf = int(uf)
self.mod = str(mod)

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)

def _edoc_situacao_em_processamento(self, proc_recibo):
return proc_recibo.resposta.cStat == "105"

def get_documento_id(self, edoc):
return edoc.infMDFe.Id[:3], edoc.infMDFe.Id[3:]

def monta_qrcode(self, chave):
return f"{QR_CODE_URL}?chMDFe={chave}&tpAmb={self.ambiente}"

def monta_qrcode_contingencia(self, edoc, xml_assinado):
chave = edoc.infMDFe.Id.replace("MDFe", "")

xml = ET.fromstring(xml_assinado)
digest_value = xml.find('.//ds:DigestValue', namespaces=NAMESPACES).text
digest_value_hex = binascii.hexlify(digest_value.encode()).decode()

return f"{self.monta_qrcode(chave)}&sign={digest_value_hex}"

def status_servico(self):
raiz = consStatServMDFe.TConsStatServ(
versao='4.00',
tpAmb='2',
cUF=35,
xServ='STATUS',
)
raiz.original_tagname_ = 'consStatServMDFe'
return self._post(
raiz,
'https://mdfe-homologacao.svrs.rs.gov.br/wws/MDFeStatusServico/MDFeStatusServico.asmx?wsdl',
'nfeStatusServicoNF',
consStatServMDFe
ConsStatServMdfe(tpAmb=self.ambiente, versao=self.versao),
localizar_url(WS_MDFE_SITUACAO, int(self.ambiente)),
"mdfeStatusServicoMDF" ,
RetConsStatServMdfe
)

def consulta_documento(self, chave):
raiz = consSitMDFe.TConsSitMDFe(
versao='4.00',
tpAmb='2',
xServ='CONSULTAR',
raiz = ConsSitMdfe(
versao=self.versao,
tpAmb=self.ambiente,
chMDFe=chave,
)
raiz.original_tagname_ = 'consSitMDFe'
return self._post(
raiz,
'https://mdfe-homologacao.svrs.rs.gov.br/ws/MDFeConsulta/MDFeConsulta.asmx?wsdl',
'MDFeConsultaNF',
consSitMDFe
localizar_url(WS_MDFE_CONSULTA, int(self.ambiente)),
"mdfeConsultaMDF",
RetConsSitMdfe
)

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

return self._post(
raiz,
'https://mdfe-homologacao.svrs.rs.gov.br/ws/MDFeConsNaoEnc/MDFeConsNaoEnc.asmx?wsdl',
'mdfeConsNaoEnc',
consMDFeNaoEnc,
localizar_url(WS_MDFE_CONSULTA_NAO_ENCERRADOS, int(self.ambiente)),
"mdfeConsNaoEnc",
RetConsMdfeNaoEnc,
)

def envia_documento(self, edoc):
Expand All @@ -92,41 +191,101 @@ def envia_documento(self, edoc):
:param edoc:
:return:
"""
xml_string, xml_etree = self._generateds_to_string_etree(edoc)
xml_assinado = Assinatura(self.certificado).assina_xml2(
xml_etree, edoc.infMDFe.Id
raiz = EnviMdfe(
versao=self.versao,
idLote=datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
MDFe=edoc
)
xml_assinado = self.assina_raiz(raiz, edoc.infMDFe.Id)
return self._post(
xml_assinado,
localizar_url(WS_MDFE_RECEPCAO, int(self.ambiente)),
"mdfeRecepcaoLote",
RetEnviMdfe
)

def consulta_recibo(self, numero=False, proc_envio=False):
if proc_envio:
numero = proc_envio.resposta.infRec.nRec

if not numero:
return

raiz = enviMDFe.TEnviMDFe(
versao='4.00',
idLote=datetime.datetime.now().strftime('%Y%m%d%H%M%S'),
raiz = ConsReciMdfe(
versao=self.versao,
tpAmb=self.ambiente,
nRec=numero,
)
raiz.original_tagname_ = 'enviMDFe'
xml_envio_string, xml_envio_etree = self._generateds_to_string_etree(
raiz
return self._post(
raiz,
localizar_url(WS_MDFE_RET_RECEPCAO, int(self.ambiente)),
"mdfeRetRecepcao",
RetConsReciMdfe,
)

def monta_processo(self, edoc, proc_envio, proc_recibo):
mdfe = proc_envio.envio_raiz.find('{' + self._namespace + '}MDFe')
protocolos = proc_recibo.resposta.protMDFe
if mdfe and protocolos:
if type(protocolos) != list:
protocolos = [protocolos]
for protocolo in protocolos:
mdfe_proc = MdfeProc(versao=self.versao, protMDFe=protocolo)
proc_recibo.processo = mdfe_proc
proc_recibo.processo_xml = mdfe_proc.to_xml()
proc_recibo.protocolo = protocolo

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
),
)
xml_envio_etree.append(etree.fromstring(xml_assinado))
raiz = EventoMdfe(versao="3.00", infEvento=inf_evento)
xml_assinado = etree.fromstring(self.assina_raiz(raiz, raiz.infEvento.Id))

return self._post(
xml_envio_etree,
'https://mdfe-homologacao.svrs.rs.gov.br/ws/MDFerecepcao/MDFeRecepcao.asmx?wsdl',
'mdfeRecepcaoLote',
enviMDFe
xml_assinado,
localizar_url(WS_MDFE_RECEPCAO_EVENTO, int(self.ambiente)),
"mdfeRecepcaoEvento",
RetEventoMdfe
)

def consulta_recibo(self, numero):
raiz = consReciMDFe.TConsReciMDFe(
versao='4.00',
tpAmb='2',
nRec=numero,
def cancela_documento(self, chave, protocolo_autorizacao, justificativa,
data_hora_evento=False):
evento_canc = EvCancMdfe(
descEvento="Cancelamento",
nProt=protocolo_autorizacao,
xJust=justificativa
)
raiz.original_tagname_ = 'consReciMDFe'
return self._post(
raiz,
'https://mdfe-homologacao.svrs.rs.gov.br/ws/MDFeRetRecepcao/MDFeRetRecepcao.asmx?wsdl', # 'ws/MDFeretautorizacao4.asmx'
'mdfeRetRecepcao',
consReciMDFe,
return self.envia_evento(
evento=evento_canc,
tipo="110111",
chave=chave,
data_hora=data_hora_evento
)

def cancela_documento(self):
pass
def encerra_documento(self, chave, protocolo_autorizacao, estado, municipio,
data_hora_evento=False):
encerramento = EvEncMdfe(
descEvento="Encerramento",
dtEnc=self._data_hoje(),
nProt=protocolo_autorizacao,
cUF=estado,
cMun=municipio
)
return self.envia_evento(
evento=encerramento,
tipo="110112",
chave=chave,
data_hora=data_hora_evento
)

0 comments on commit 2ea4d7d

Please sign in to comment.