# -*- coding: utf-8 -*- import time import re import requests import collections from io import StringIO 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_em_processamento = '105' _edoc_situacao_servico_em_operacao = '107' consulta_servico_ao_enviar = True maximo_tentativas_consulta_recibo = 5 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) attr = collections.OrderedDict() attr['xmlns'] = NAMESPACE_MDFE attr['versao'] = self._versao raiz = etree.Element('consMDFeNaoEnc', attr) 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""" ns = collections.OrderedDict() ns['xsi'] = self._namespace_xsi ns['xsd'] = self._namespace_xsd ns[self._soap_version] = self._namespace_soap raiz = etree.Element( '{%s}Envelope' % self._namespace_soap, nsmap=ns ) 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 ) 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 = '' # limpa xml com caracteres bugados para infNFeSupl em NFC-e xml = re.sub( '(.*?)', lambda x: x.group(0).replace('<', '<').replace('>', '>').replace('&', ''), etree.tostring(xml, encoding='unicode').replace('\n', '') ) xml = xml_declaration + xml # print(xml) # print('-' * 20) # 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