You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
286 lines
11 KiB
286 lines
11 KiB
# -*- coding: utf-8 -*-
|
|
import time
|
|
import re
|
|
import requests
|
|
from io import StringIO
|
|
import base64
|
|
|
|
from pynfe.utils.flags import (
|
|
NAMESPACE_MDFE,
|
|
MODELO_MDFE,
|
|
VERSAO_MDFE,
|
|
NAMESPACE_MDFE_METODO,
|
|
NAMESPACE_SOAP,
|
|
NAMESPACE_XSI,
|
|
NAMESPACE_XSD,
|
|
CODIGOS_ESTADOS
|
|
)
|
|
from pynfe.utils.webservices import MDFE
|
|
from pynfe.entidades.certificado import CertificadoA1
|
|
from pynfe.utils import etree, extrai_id_srtxml
|
|
from .comunicacao import Comunicacao
|
|
from .resposta import analisar_retorno
|
|
|
|
MDFE_SITUACAO_JA_ENVIADO = ('100', '101', '132')
|
|
|
|
|
|
class ComunicacaoMDFe(Comunicacao):
|
|
|
|
_modelo = MODELO_MDFE
|
|
_namespace = NAMESPACE_MDFE
|
|
_versao = VERSAO_MDFE
|
|
_header = 'mdfeCabecMsg'
|
|
_envio_mensagem = 'mdfeDadosMsg'
|
|
_retorno_mensagem = 'mdfeRecepcaoResult'
|
|
_namespace_metodo = NAMESPACE_MDFE_METODO
|
|
|
|
_accept = True
|
|
_soap_action = False
|
|
_namespace_soap = NAMESPACE_SOAP
|
|
_namespace_xsi = NAMESPACE_XSI
|
|
_namespace_xsd = NAMESPACE_XSD
|
|
_soap_version = 'soap12'
|
|
_edoc_situacao_ja_enviado = MDFE_SITUACAO_JA_ENVIADO
|
|
_edoc_situacao_arquivo_recebido_com_sucesso = '103'
|
|
_edoc_situacao_lote_processado = '104'
|
|
_edoc_situacao_em_processamento = '105'
|
|
_edoc_situacao_servico_em_operacao = '107'
|
|
|
|
consulta_servico_ao_enviar = True
|
|
maximo_tentativas_consulta_recibo = 5
|
|
|
|
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
|
|
: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.
|
|
"""
|
|
# 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 f'ind_sinc deve ser 0=assincrono ou 1=sincrono'
|
|
|
|
# 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
|
|
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)
|
|
|
|
# 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}
|
|
# Procuta status no xml
|
|
try:
|
|
prot = etree.fromstring(retorno.text)
|
|
except ValueError:
|
|
# 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
|
|
|
|
# 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
|
|
|
|
# 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 == self._edoc_situacao_arquivo_recebido_com_sucesso:
|
|
nrec = rec.xpath("ns:retEnviMDFe/ns:infRec/ns:nRec", namespaces=ns)[0].text
|
|
return 0, nrec, manifesto
|
|
return 1, retorno, manifesto
|
|
|
|
def status_servico(self):
|
|
url = self._get_url('STATUS')
|
|
# Monta XML do corpo da requisição
|
|
raiz = etree.Element('consStatServMDFe', versao=self._versao, xmlns=NAMESPACE_MDFE)
|
|
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
|
|
etree.SubElement(raiz, 'xServ').text = 'STATUS'
|
|
xml = self._construir_xml_soap('MDFeStatusServico', raiz)
|
|
return self._post(url, xml)
|
|
|
|
def consulta(self, chave):
|
|
url = self._get_url('CONSULTA')
|
|
# Monta XML do corpo da requisição
|
|
raiz = etree.Element('consSitMDFe', versao=self._versao, xmlns=NAMESPACE_MDFE)
|
|
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
|
|
etree.SubElement(raiz, 'xServ').text = 'CONSULTAR'
|
|
etree.SubElement(raiz, 'chMDFe').text = chave
|
|
# Monta XML para envio da requisição
|
|
xml = self._construir_xml_soap('MDFeConsulta', raiz)
|
|
return self._post(url, xml)
|
|
|
|
def consulta_nao_encerrados(self, cpfcnpj):
|
|
url = self._get_url('NAO_ENCERRADOS')
|
|
# Monta XML do corpo da requisição
|
|
raiz = etree.Element('consMDFeNaoEnc', xmlns=NAMESPACE_MDFE, versao=self._versao)
|
|
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
|
|
etree.SubElement(raiz, 'xServ').text = 'CONSULTAR NÃO ENCERRADOS'
|
|
if len(cpfcnpj) == 11:
|
|
etree.SubElement(raiz, 'CPF').text = cpfcnpj.zfill(11)
|
|
else:
|
|
etree.SubElement(raiz, 'CNPJ').text = cpfcnpj.zfill(14)
|
|
# Monta XML para envio da requisição
|
|
xml = self._construir_xml_soap('MDFeConsNaoEnc', raiz)
|
|
return self._post(url, xml)
|
|
|
|
def consulta_recibo(self, numero):
|
|
url = self._get_url('RET_RECEPCAO')
|
|
# Monta XML do corpo da requisição
|
|
raiz = etree.Element('consReciMDFe', versao=self._versao, xmlns=NAMESPACE_MDFE)
|
|
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
|
|
etree.SubElement(raiz, 'nRec').text = numero.zfill(15)
|
|
# Monta XML para envio da requisição
|
|
xml = self._construir_xml_soap('MDFeRetRecepcao', raiz)
|
|
return self._post(url, xml)
|
|
|
|
def evento(self, evento):
|
|
"""
|
|
Envia eventos do MDFe como:
|
|
Encerramento
|
|
Cancelamento
|
|
Inclusao Condutor
|
|
Inclusao DF-e
|
|
Pagamento Operacao MDF-e
|
|
:param evento: Nome do Evento
|
|
:return:
|
|
"""
|
|
# url do serviço
|
|
url = self._get_url('EVENTOS')
|
|
# Monta XML do corpo da requisição
|
|
xml = self._construir_xml_soap('MDFeRecepcaoEvento', evento)
|
|
return self._post(url, xml)
|
|
|
|
def _construir_xml_soap(self, metodo, dados):
|
|
"""Mota o XML para o envio via SOAP"""
|
|
|
|
raiz = etree.Element(
|
|
'{%s}Envelope' % NAMESPACE_SOAP,
|
|
nsmap={
|
|
'xsi': self._namespace_xsi,
|
|
'xsd': self._namespace_xsd,
|
|
self._soap_version: self._namespace_soap
|
|
}
|
|
)
|
|
|
|
if self._header:
|
|
cabecalho = self._cabecalho_soap(metodo)
|
|
c = etree.SubElement(raiz, '{%s}Header' % self._namespace_soap)
|
|
c.append(cabecalho)
|
|
|
|
body = etree.SubElement(raiz, '{%s}Body' % self._namespace_soap)
|
|
|
|
a = etree.SubElement(
|
|
body,
|
|
self._envio_mensagem,
|
|
xmlns=self._namespace_metodo+metodo
|
|
)
|
|
|
|
# if metodo == 'MDFeRecepcaoSinc':
|
|
# body_base64 = base64.b16encode(a).decode()
|
|
|
|
a.append(dados)
|
|
return raiz
|
|
|
|
def _post_header(self, soap_webservice_method=False):
|
|
"""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;',
|
|
}
|
|
|
|
# PE é a únca UF que exige SOAPAction no header
|
|
if soap_webservice_method:
|
|
header[b'SOAPAction'] = \
|
|
(self._namespace_metodo + soap_webservice_method).encode('utf-8')
|
|
|
|
if self._accept:
|
|
header[b'Accept'] = b'application/soap+xml; charset=utf-8;'
|
|
|
|
return header
|
|
|
|
def _post(self, url, xml):
|
|
certificado_a1 = CertificadoA1(self.certificado)
|
|
chave, cert = certificado_a1.separar_arquivo(self.certificado_senha, caminho=True)
|
|
chave_cert = (cert, chave)
|
|
# Abre a conexão HTTPS
|
|
try:
|
|
xml_declaration = '<?xml version="1.0" encoding="UTF-8"?>'
|
|
|
|
# limpa xml com caracteres bugados para infMDFeSupl em NFC-e
|
|
xml = re.sub(
|
|
'<qrCodMDFe>(.*?)</qrCodMDFe>',
|
|
lambda x: x.group(0).replace('<', '<').replace('>', '>').replace('amp;', ''),
|
|
etree.tostring(xml, encoding='unicode').replace('\n', '')
|
|
)
|
|
xml = xml_declaration + xml
|
|
xml = xml.encode('utf8') # necessário para o evento "CONSULTAR NÃO ENCERRADOS"
|
|
|
|
# Faz o request com o servidor
|
|
result = requests.post(
|
|
url,
|
|
xml,
|
|
headers=self._post_header(),
|
|
cert=chave_cert,
|
|
verify=False
|
|
)
|
|
result.encoding = 'utf-8'
|
|
return result
|
|
except requests.exceptions.RequestException as e:
|
|
raise e
|
|
finally:
|
|
certificado_a1.excluir()
|
|
|
|
def _cabecalho_soap(self, metodo):
|
|
"""Monta o XML do cabeçalho da requisição SOAP"""
|
|
|
|
raiz = etree.Element(
|
|
self._header,
|
|
xmlns=self._namespace_metodo + metodo
|
|
)
|
|
etree.SubElement(raiz, 'cUF').text = CODIGOS_ESTADOS[self.uf.upper()]
|
|
etree.SubElement(raiz, 'versaoDados').text = '3.00'
|
|
return raiz
|
|
|
|
def _get_url(self, consulta):
|
|
# producao
|
|
if self._ambiente == 1:
|
|
ambiente = MDFE['SVRS']['HTTPS']
|
|
# homologacao
|
|
else:
|
|
ambiente = MDFE['SVRS']['HOMOLOGACAO']
|
|
|
|
self.url = ambiente + MDFE['SVRS'][consulta]
|
|
return self.url
|