diff --git a/pynfe/processamento/comunicacao.py b/pynfe/processamento/comunicacao.py index 6a22b747..b5f466c8 100644 --- a/pynfe/processamento/comunicacao.py +++ b/pynfe/processamento/comunicacao.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- import datetime import re +import pytz import requests @@ -54,7 +55,7 @@ class ComunicacaoSefaz(Comunicacao): _assinatura = AssinaturaA1 def autorizacao( - self, modelo, nota_fiscal, id_lote=1, ind_sinc=1, contingencia=False, timeout=None + self, modelo, nota_fiscal, id_lote=1, ind_sinc=1, contingencia=False, timeout=None ): """ Método para realizar autorização da nota de acordo com o modelo @@ -183,7 +184,7 @@ def consulta_nota(self, modelo, chave, contingencia=False): return self._post(url, xml) def consulta_distribuicao( - self, cnpj=None, cpf=None, chave=None, nsu=0, consulta_nsu_especifico=False + self, cnpj=None, cpf=None, chave=None, nsu=0, consulta_nsu_especifico=False ): """ O XML do pedido de distribuição suporta três tipos de consultas @@ -274,10 +275,10 @@ def consulta_cadastro(self, modelo, documento, tipo='CNPJ', uf=None): info = etree.SubElement(raiz, "infCons") etree.SubElement(info, "xServ").text = "CONS-CAD" etree.SubElement(info, "UF").text = uf.upper() - + # Monta tipo de documento CNPJ, CPF ou IE etree.SubElement(info, tipo.upper()).text = documento - + # etree.SubElement(info, 'CPF').text = cpf # Monta XML para envio da requisição @@ -329,14 +330,14 @@ def status_servico(self, modelo, timeout=None): return self._post(url, xml, timeout) def inutilizacao( - self, - modelo, - cnpj, - numero_inicial, - numero_final, - justificativa="", - ano=None, - serie="1", + self, + modelo, + cnpj, + numero_inicial, + numero_final, + justificativa="", + ano=None, + serie="1", ): """ Serviço destinado ao atendimento de solicitações de inutilização de numeração. @@ -366,16 +367,16 @@ def inutilizacao( # Identificador da TAG a ser assinada formada com Código da UF + Ano (2 posições) + # CNPJ + modelo + série + nro inicial e nro final precedida do literal “ID” id_unico = ( - "ID%(uf)s%(ano)s%(cnpj)s%(modelo)s%(serie)s%(num_ini)s%(num_fin)s" - % { - "uf": uf, - "ano": ano, - "cnpj": cnpjcpf_chaveacesso, - "modelo": "55" if modelo == "nfe" else "65", # 55=NF-e; 65=NFC-e; - "serie": str(serie).zfill(3), - "num_ini": str(numero_inicial).zfill(9), - "num_fin": str(numero_final).zfill(9), - } + "ID%(uf)s%(ano)s%(cnpj)s%(modelo)s%(serie)s%(num_ini)s%(num_fin)s" + % { + "uf": uf, + "ano": ano, + "cnpj": cnpjcpf_chaveacesso, + "modelo": "55" if modelo == "nfe" else "65", # 55=NF-e; 65=NFC-e; + "serie": str(serie).zfill(3), + "num_ini": str(numero_inicial).zfill(9), + "num_fin": str(numero_final).zfill(9), + } ) # Monta XML do corpo da requisição # FIXME @@ -491,7 +492,7 @@ def _get_url(self, modelo, consulta, contingencia=False): else: # nfe Ex: https://nfe.fazenda.pr.gov.br/nfe/NFeStatusServico3 self.url = ( - NFE[self.uf.upper()][ambiente] + NFE[self.uf.upper()][consulta] + NFE[self.uf.upper()][ambiente] + NFE[self.uf.upper()][consulta] ) elif modelo == "nfce": # PE e BA são as únicas UF'sque possuem NFE proprio e SVRS para NFCe @@ -500,8 +501,8 @@ def _get_url(self, modelo, consulta, contingencia=False): else: # nfce Ex: https://homologacao.nfce.fazenda.pr.gov.br/nfce/NFeStatusServico3 self.url = ( - NFCE[self.uf.upper()][ambiente] - + NFCE[self.uf.upper()][consulta] + NFCE[self.uf.upper()][ambiente] + + NFCE[self.uf.upper()][consulta] ) else: raise Exception('Modelo não encontrado! Defina modelo="nfe" ou "nfce"') @@ -965,7 +966,7 @@ def autorizacao(self, manifesto, id_lote=1, ind_sinc=1): # autorizado uso do MDF-e # retorna xml final (protMDFe + MDFe) if ( - status in self._edoc_situacao_ja_enviado + status in self._edoc_situacao_ja_enviado ): # if status == '100': raiz = etree.Element( "mdfeProc", xmlns=NAMESPACE_MDFE, versao=VERSAO_MDFE @@ -982,8 +983,8 @@ def autorizacao(self, manifesto, id_lote=1, ind_sinc=1): status = rec.xpath("ns:retEnviMDFe/ns:cStat", namespaces=ns)[0].text # Lote Recebido com Sucesso! if status in ( - self._edoc_situacao_arquivo_recebido_com_sucesso, - self._edoc_situacao_em_processamento, + self._edoc_situacao_arquivo_recebido_com_sucesso, + self._edoc_situacao_em_processamento, ): nrec = rec.xpath("ns:retEnviMDFe/ns:infRec/ns:nRec", namespaces=ns)[ 0 @@ -1094,7 +1095,7 @@ def _post_header(self, soap_webservice_method=False): # PE é a únca UF que exige SOAPAction no header if soap_webservice_method: header[b"SOAPAction"] = ( - self._namespace_metodo + soap_webservice_method + self._namespace_metodo + soap_webservice_method ).encode("utf-8") if self._accept: @@ -1247,7 +1248,7 @@ def consulta_distribuicao(self, cnpj=None, cpf=None, chave=None, nsu=0, consulta # Monta XML para envio da requisição xml = self._construir_xml_soap("CTeDistribuicaoDFe", raiz) return self._post(url, xml) - + def consulta(self, chave): url = self._get_url("CONSULTA") # Monta XML do corpo da requisição @@ -1257,7 +1258,53 @@ def consulta(self, chave): etree.SubElement(raiz, "chCTe").text = chave # Monta XML para envio da requisição xml = self._construir_xml_soap("cteConsultaCT", raiz) - return self._post(url, xml) + + # Monta a SOAPAction com o namespace correto + soap_action = f"{self._namespace_metodo}ConsultaV4/cteConsultaCT" + + return self._post(url, xml, soap_action) + + def prestacao_em_desacordo(self, chave_cte, cnpj, id_lote=1): + codigo_estado = chave_cte[0:2] + + fuso_horario = pytz.timezone('America/Sao_Paulo') + dt_com_fuso = datetime.datetime.now(fuso_horario) + data_hora_evento_str = dt_com_fuso.strftime('%Y-%m-%dT%H:%M:%S%z') + data_hora = data_hora_evento_str[:-2] + ':' + data_hora_evento_str[-2:] + id_evento = f"ID610110{chave_cte}{str('1').zfill(3)}" # 610110 é o código do evento de prestacao de servico em desacordo + + evento_root = etree.Element("eventoCTe", versao="4.00", xmlns=NAMESPACE_CTE) + inf_evento = etree.SubElement(evento_root, "infEvento", Id=id_evento) + + etree.SubElement(inf_evento, "cOrgao").text = CODIGOS_ESTADOS[self.uf.upper()] # Código IBGE da UF + etree.SubElement(inf_evento, "tpAmb").text = str(self._ambiente) + etree.SubElement(inf_evento, "CNPJ").text = cnpj # CNPJ do autor do evento + etree.SubElement(inf_evento, "chCTe").text = chave_cte + etree.SubElement(inf_evento, + "dhEvento").text = data_hora + etree.SubElement(inf_evento, "tpEvento").text = "610110" # Código do evento + etree.SubElement(inf_evento, "nSeqEvento").text = "001" # Sequencial do evento para a mesma chave + + det_evento = etree.SubElement(inf_evento, "detEvento", versaoEvento="4.00") + + ev_prest_desacordo = etree.SubElement(det_evento, "evPrestDesacordo", xmlns=NAMESPACE_CTE) + etree.SubElement(ev_prest_desacordo, "descEvento").text = "Prestacao do Servico em Desacordo" + etree.SubElement(ev_prest_desacordo, "indDesacordoOper").text = "1" + etree.SubElement(ev_prest_desacordo, + "xObs").text = "O servico de transporte foi prestado com avarias na mercadoria." + a1 = AssinaturaA1(self.certificado, self.certificado_senha) + xml = a1.assinar(evento_root) + + return self.evento(xml, id_lote, codigo_estado) + + def evento(self, evento_xml, id_lote=1, codigo_estado='43'): + self.uf = [key for key, value in CODIGOS_ESTADOS.items() if value == codigo_estado][0] + url = self._get_url("RECEPCAO_EVENTO") + raiz = evento_xml + xml_soap = self._construir_xml_soap("CTeRecepcaoEventoV4", raiz, False, "4.00") + soap_action = f"{self._namespace_metodo}CTeRecepcaoEventoV4/cteRecepcaoEvento" + + return self._post(url, xml_soap, soap_action) def _get_url_an(self, consulta): ambiente = "https://www1." # produção @@ -1267,12 +1314,12 @@ def _get_url_an(self, consulta): self.url = ambiente + CTE["AN"][consulta] return self.url - def _cabecalho_soap(self, metodo): + def _cabecalho_soap(self, metodo, versao_dados): """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" + etree.SubElement(raiz, "versaoDados").text = versao_dados return raiz def _get_url(self, consulta): @@ -1329,7 +1376,7 @@ def _get_url(self, consulta): raise Exception(f"Url não encontrada para {consulta} {self.uf.upper()}") return self.url - def _construir_xml_soap(self, metodo, dados, cabecalho=False): + def _construir_xml_soap(self, metodo, dados, cabecalho=False, versao_dados="3.00"): """Monta o XML para o envio via SOAP""" raiz = etree.Element( @@ -1338,7 +1385,7 @@ def _construir_xml_soap(self, metodo, dados, cabecalho=False): ) if self._header: - cabecalho = self._cabecalho_soap(metodo) + cabecalho = self._cabecalho_soap(metodo, versao_dados) c = etree.SubElement(raiz, "{%s}Header" % self._namespace_soap) c.append(cabecalho) @@ -1349,6 +1396,10 @@ def _construir_xml_soap(self, metodo, dados, cabecalho=False): body, "cteDistDFeInteresse", xmlns=NAMESPACE_CTE_METODO + metodo ) a = etree.SubElement(x, "cteDadosMsg") + elif metodo == 'CTeRecepcaoEventoV4': + a = etree.SubElement( + body, "cteDadosMsg", xmlns=NAMESPACE_CTE_METODO + metodo + ) else: a = etree.SubElement( body, "cteDadosMsg", xmlns=NAMESPACE_CTE_METODO + metodo @@ -1356,21 +1407,23 @@ def _construir_xml_soap(self, metodo, dados, cabecalho=False): a.append(dados) return raiz - def _post_header(self): - """Retorna um dicionário com os atributos para o cabeçalho da requisição HTTP""" - response = { - "content-type": "application/soap+xml; charset=utf-8;", - "Accept": "application/soap+xml; charset=utf-8;", + def _post_header(self, soap_action=None): + headers = { + 'Content-Type': 'application/soap+xml; charset=utf-8;', + 'Accept': 'application/soap+xml; charset=utf-8;', } - response["SOAPAction"] = "" - return response + if soap_action: + headers['Content-Type'] = headers.get('Content-Type') + f'action={soap_action}' - def _post(self, url, xml): + return headers + + def _post(self, url, xml, soap_action=None): certificado_a1 = CertificadoA1(self.certificado) chave, cert = certificado_a1.separar_arquivo( self.certificado_senha, caminho=True ) chave_cert = (cert, chave) + request_headers = self._post_header(soap_action) # Abre a conexão HTTPS try: xml_declaration = '' @@ -1390,7 +1443,7 @@ def _post(self, url, xml): result = requests.post( url, xml, - headers=self._post_header(), + headers=request_headers, cert=chave_cert, verify=False, timeout=300, @@ -1402,4 +1455,4 @@ def _post(self, url, xml): except requests.exceptions.RequestException as e: raise e finally: - certificado_a1.excluir() + certificado_a1.excluir() \ No newline at end of file diff --git a/pynfe/utils/webservices.py b/pynfe/utils/webservices.py index 8d2a1d3e..bb451e7b 100644 --- a/pynfe/utils/webservices.py +++ b/pynfe/utils/webservices.py @@ -553,24 +553,28 @@ "MT": { "STATUS": "sefaz.mt.gov.br/ctews/services/CteStatusServico", "CONSULTA": "sefaz.mt.gov.br/ctews2/services/CTeConsultaV4?wsdl", + "RECEPCAO_EVENTO": "sefaz.mt.gov.br/ctews2/services/CTeRecepcaoEventoV4?wsdl", "HTTPS": "https://cte.", "HOMOLOGACAO": "https://homologacao.", }, "MS": { "STATUS": "cte.ms.gov.br/ws/CteStatusServico", "CONSULTA": "cte.ms.gov.br/ws/CTeConsultaV4", + "RECEPCAO_EVENTO": "cte.ms.gov.br/ws/CTeRecepcaoEventoV4", "HTTPS": "https://producao.", "HOMOLOGACAO": "https://homologacao.", }, "MG": { "STATUS": "fazenda.mg.gov.br/cte/services/CteStatusServico", "CONSULTA": "fazenda.mg.gov.br/cte/services/CTeConsultaV4", + "RECEPCAO_EVENTO": "fazenda.mg.gov.br/cte/services/CTeRecepcaoEventoV4", "HTTPS": "https://cte.", "HOMOLOGACAO": "https://hcte.", }, "PR": { "STATUS": "fazenda.pr.gov.br/cte/CteStatusServico?wsdl", "CONSULTA": "fazenda.pr.gov.br/cte4/CTeConsultaV4?wsdl", + "RECEPCAO_EVENTO": "fazenda.pr.gov.br/cte4/CTeRecepcaoEventoV4?wsdl", "HTTPS": "https://cte.", "HOMOLOGACAO": "https://homologacao.", }, @@ -578,23 +582,27 @@ "STATUS": "svrs.rs.gov.br/ws/ctestatusservico/CteStatusServico.asmx", "CONSULTA": "svrs.rs.gov.br/ws/CTeConsultaV4/CTeConsultaV4.asmx", "HTTPS": "https://cte.", + "RECEPCAO_EVENTO": "svrs.rs.gov.br/ws/CTeRecepcaoEventoV4/CTeRecepcaoEventoV4.asmx", "HOMOLOGACAO": "https://cte-homologacao.", }, "SP": { "STATUS": "fazenda.sp.gov.br/cteWEB/services/cteStatusServico.asmx", "CONSULTA": "fazenda.sp.gov.br/CTeWS/WS/CTeConsultaV4.asmx", + "RECEPCAO_EVENTO": "fazenda.sp.gov.br/CTeWS/WS/CTeRecepcaoEventoV4.asmx", "HTTPS": "https://nfe.", "HOMOLOGACAO": "https://homologacao.nfe.", }, "SVRS": { "STATUS": "svrs.rs.gov.br/ws/ctestatusservico/CteStatusServico.asmx", "CONSULTA": "svrs.rs.gov.br/ws/CTeConsultaV4/CTeConsultaV4.asmx", + "RECEPCAO_EVENTO": "svrs.rs.gov.br/ws/CTeRecepcaoEventoV4/CTeRecepcaoEventoV4.asmx", "HTTPS": "https://cte.", "HOMOLOGACAO": "https://cte-homologacao.", }, "SVSP": { "STATUS": "fazenda.sp.gov.br/cteWEB/services/CteStatusServico.asmx", "CONSULTA": "fazenda.sp.gov.br/CTeWS/WS/CTeConsultaV4.asmx", + "RECEPCAO_EVENTO": "fazenda.sp.gov.br/CTeWS/WS/CTeRecepcaoEventoV4.asmx", "HTTPS": "https://nfe.", "HOMOLOGACAO": "https://homologacao.nfe.", }, diff --git a/requirements.txt b/requirements.txt index 711dbba4..71dd232d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,5 +3,6 @@ requests lxml signxml cryptography +pytz # Opcional para NFS-e #-r requirements-nfse.txt diff --git a/setup.py b/setup.py index a2ac4f8f..07ed8c3d 100644 --- a/setup.py +++ b/setup.py @@ -34,6 +34,7 @@ "requests", "lxml", "signxml", + "pytz" ], extras_require={ "nfse": [