felipe
/
PyNFe
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Adicionando e refatorando documentação e melhorando para PEP8

pull/22/head
Lucas Cardoso 8 years ago
parent
faf354d290
commit
50cbb325ec
1 changed files with 189 additions and 86 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 275
      pynfe/processamento/comunicacao.py

275
pynfe/processamento/comunicacao.py
View File

@ -25,8 +25,10 @@ from .assinatura import AssinaturaA1, AssinaturaA1SignXML
class Comunicacao(object): class Comunicacao(object):
u"""Classe abstrata responsavel por definir os metodos e logica das classes
de comunicação com os webservices da NF-e."""
"""
Classe abstrata responsavel por definir os metodos e logica das classes
de comunicação com os webservices da NF-e.
"""
_ambiente = 1 # 1 = Produção, 2 = Homologação _ambiente = 1 # 1 = Produção, 2 = Homologação
uf = None uf = None
@ -40,52 +42,67 @@ class Comunicacao(object):
self.certificado_senha = certificado_senha self.certificado_senha = certificado_senha
self._ambiente = 2 if homologacao else 1 self._ambiente = 2 if homologacao else 1
class ComunicacaoSefaz(Comunicacao): class ComunicacaoSefaz(Comunicacao):
u"""Classe de comunicação que segue o padrão definido para as SEFAZ dos Estados."""
"""Classe de comunicação que segue o padrão definido para as SEFAZ dos Estados."""
_versao = VERSAO_PADRAO _versao = VERSAO_PADRAO
_assinatura = AssinaturaA1 _assinatura = AssinaturaA1
def autorizacao(self, modelo, nota_fiscal, idlote=1, indSinc=1):
def autorizacao(self, modelo, nota_fiscal, id_lote=1, ind_sinc=1):
"""
Método para realizar autorização da nota de acordo com o modelo
:param modelo: Modelo
:param nota_fiscal: 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 nfe 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 # url do serviço
url = self._get_url(modelo=modelo, consulta='AUTORIZACAO') url = self._get_url(modelo=modelo, consulta='AUTORIZACAO')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('enviNFe', xmlns=NAMESPACE_NFE, versao=VERSAO_PADRAO) raiz = etree.Element('enviNFe', xmlns=NAMESPACE_NFE, versao=VERSAO_PADRAO)
etree.SubElement(raiz, 'idLote').text = str(idlote) # numero autoincremental gerado pelo sistema
etree.SubElement(raiz, 'indSinc').text = str(indSinc) # 0 para assincrono, 1 para sincrono
etree.SubElement(raiz, 'idLote').text = str(id_lote) # numero autoincremental gerado pelo sistema
etree.SubElement(raiz, 'indSinc').text = str(ind_sinc) # 0 para assincrono, 1 para sincrono
raiz.append(nota_fiscal) raiz.append(nota_fiscal)
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeAutorizacao'), metodo='NfeAutorizacao', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeAutorizacao'), metodo='NfeAutorizacao', dados=raiz
)
# Faz request no Servidor da Sefaz # Faz request no Servidor da Sefaz
retorno = self._post(url, xml) retorno = self._post(url, xml)
# Em caso de sucesso, retorna xml com nfe e protocolo de autorização. # Em caso de sucesso, retorna xml com nfe e protocolo de autorização.
# Caso contrário, envia todo o soap de resposta da Sefaz para decisão do usuário. # Caso contrário, envia todo o soap de resposta da Sefaz para decisão do usuário.
if retorno.status_code == 200: if retorno.status_code == 200:
if indSinc == 1:
if ind_sinc == 1:
# Procuta status no xml # Procuta status no xml
ns = {'ns':'http://www.portalfiscal.inf.br/nfe'} # namespace
ns = {'ns': 'http://www.portalfiscal.inf.br/nfe'} # namespace
try: try:
prot = etree.fromstring(retorno.text) prot = etree.fromstring(retorno.text)
except ValueError: except ValueError:
#em SP retorno.text apresenta erro
# em SP retorno.text apresenta erro
prot = etree.fromstring(retorno.content) prot = etree.fromstring(retorno.content)
try: try:
# Protocolo com envio OK # Protocolo com envio OK
infProt = prot[1][0][0][6] # root protNFe
status = infProt.xpath("ns:infProt/ns:cStat", namespaces=ns)[0].text
inf_prot = prot[1][0][0][6] # root protNFe
status = inf_prot.xpath("ns:infProt/ns:cStat", namespaces=ns)[0].text
except IndexError: except IndexError:
# Protocolo com algum erro no Envio # Protocolo com algum erro no Envio
retEnvi = prot[1][0][0] # root retEnvi
status = retEnvi.xpath("ns:cStat", namespaces=ns)[0].text
ret_envi = prot[1][0][0] # root retEnvi
status = ret_envi.xpath("ns:cStat", namespaces=ns)[0].text
if status == '100': if status == '100':
raiz = etree.Element('nfeProc', xmlns=NAMESPACE_NFE, versao=VERSAO_PADRAO) raiz = etree.Element('nfeProc', xmlns=NAMESPACE_NFE, versao=VERSAO_PADRAO)
raiz.append(nota_fiscal) raiz.append(nota_fiscal)
raiz.append(infProt)
raiz.append(inf_prot)
return 0, raiz return 0, raiz
else: else:
# Retorna id do protocolo para posterior consulta em caso de sucesso. # Retorna id do protocolo para posterior consulta em caso de sucesso.
ns = {'ns':'http://www.portalfiscal.inf.br/nfe'} # namespace
ns = {'ns': 'http://www.portalfiscal.inf.br/nfe'} # namespace
rec = etree.fromstring(retorno.text) rec = etree.fromstring(retorno.text)
rec = rec[1][0][0] rec = rec[1][0][0]
status = rec.xpath("ns:cStat", namespaces=ns)[0].text status = rec.xpath("ns:cStat", namespaces=ns)[0].text
@ -97,64 +114,95 @@ class ComunicacaoSefaz(Comunicacao):
def consulta_recibo(self, modelo, numero): def consulta_recibo(self, modelo, numero):
""" """
Este método oferece a consulta do resultado do processamento de um lote de NF-e.
O aplicativo do Contribuinte deve ser construído de forma a aguardar um tempo mínimo de
15 segundos entre o envio do Lote de NF-e para processamento e a consulta do resultado
deste processamento, evitando a obtenção desnecessária do status de erro 105 - "Lote em
Processamento".
Este método oferece a consulta do resultado do processamento de um lote de NF-e.
O aplicativo do Contribuinte deve ser construído de forma a aguardar um tempo mínimo de
15 segundos entre o envio do Lote de NF-e para processamento e a consulta do resultado
deste processamento, evitando a obtenção desnecessária do status de erro 105 - "Lote em
Processamento".
:param modelo: Modelo da nota
:param numero: Número da nota
:return:
""" """
# url do serviço # url do serviço
url = self._get_url(modelo=modelo, consulta='RECIBO') url = self._get_url(modelo=modelo, consulta='RECIBO')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('consReciNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE) raiz = etree.Element('consReciNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE)
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente) etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
etree.SubElement(raiz, 'nRec').text = numero etree.SubElement(raiz, 'nRec').text = numero
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeRetAutorizacao'), metodo='NfeRetAutorizacao', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeRetAutorizacao'), metodo='NfeRetAutorizacao', dados=raiz
)
return self._post(url, xml) return self._post(url, xml)
def consulta_nota(self, modelo, chave): def consulta_nota(self, modelo, chave):
""" """
Este método oferece a consulta da situação da NF-e/NFC-e na Base de Dados do Portal da Secretaria de Fazenda Estadual.
Este método oferece a consulta da situação da NF-e/NFC-e na Base de Dados do Portal
da Secretaria de Fazenda Estadual.
:param modelo: Modelo da nota
:param chave: Chave da nota
:return:
""" """
# url do serviço # url do serviço
url = self._get_url(modelo=modelo, consulta='CHAVE') url = self._get_url(modelo=modelo, consulta='CHAVE')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('consSitNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE) raiz = etree.Element('consSitNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE)
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente) etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
etree.SubElement(raiz, 'xServ').text = 'CONSULTAR' etree.SubElement(raiz, 'xServ').text = 'CONSULTAR'
etree.SubElement(raiz, 'chNFe').text = chave etree.SubElement(raiz, 'chNFe').text = chave
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeConsulta2'), metodo='NfeConsulta2', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeConsulta2'), metodo='NfeConsulta2', dados=raiz
)
return self._post(url, xml) return self._post(url, xml)
def consulta_notas_cnpj(self, cnpj, nsu=0): def consulta_notas_cnpj(self, cnpj, nsu=0):
""" """
“Serviço de Consulta da Relação de Documentos Destinados” para um determinado CNPJ de destinatário informado na NF-e.
“Serviço de Consulta da Relação de Documentos Destinados” para um determinado CNPJ de
destinatário informado na NF-e.
:param cnpj: CNPJ
:param nsu: NSU
:return:
""" """
# url do serviço # url do serviço
url = self._get_url_AN(consulta='DESTINADAS')
url = self._get_url_an(consulta='DESTINADAS')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('consNFeDest', versao='1.01', xmlns=NAMESPACE_NFE) raiz = etree.Element('consNFeDest', versao='1.01', xmlns=NAMESPACE_NFE)
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente) etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
etree.SubElement(raiz, 'xServ').text = 'CONSULTAR NFE DEST' etree.SubElement(raiz, 'xServ').text = 'CONSULTAR NFE DEST'
etree.SubElement(raiz, 'CNPJ').text = cnpj etree.SubElement(raiz, 'CNPJ').text = cnpj
# Indicador de NF-e consultada: # Indicador de NF-e consultada:
# 0=Todas as NF-e;
# 1=Somente as NF-e que ainda não tiveram manifestação do destinatário (Desconhecimento da operação, Operação não Realizada ou Confirmação da Operação);
# 2=Idem anterior, incluindo as NF-e que também não tiveram a Ciência da Operação.
# 0 = Todas as NF-e;
# 1 = Somente as NF-e que ainda não tiveram manifestação do destinatário (Desconhecimento da
# operação, Operação não Realizada ou Confirmação da Operação);
# 2 = Idem anterior, incluindo as NF-e que também não tiveram a Ciência da Operação.
etree.SubElement(raiz, 'indNFe').text = '0' etree.SubElement(raiz, 'indNFe').text = '0'
# Indicador do Emissor da NF-e: # Indicador do Emissor da NF-e:
# 0=Todos os Emitentes / Remetentes;
# 1=Somente as NF-e emitidas por emissores / remetentes que não tenham o mesmo CNPJ-Base do destinatário (para excluir as notas fiscais de transferência entre filiais).
# 0 = Todos os Emitentes / Remetentes;
# 1 = Somente as NF-e emitidas por emissores / remetentes que não tenham o mesmo CNPJ-Base do
# destinatário (para excluir as notas fiscais de transferência entre filiais).
etree.SubElement(raiz, 'indEmi').text = '0' etree.SubElement(raiz, 'indEmi').text = '0'
# Último NSU recebido pela Empresa. Caso seja informado com zero, ou com um NSU muito antigo, a consulta retornará unicamente as notas fiscais que tenham sido recepcionadas nos últimos 15 dias.
# Último NSU recebido pela Empresa. Caso seja informado com zero, ou com um NSU muito antigo, a consulta
# retornará unicamente as notas fiscais que tenham sido recepcionadas nos últimos 15 dias.
etree.SubElement(raiz, 'ultNSU').text = str(nsu) etree.SubElement(raiz, 'ultNSU').text = str(nsu)
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeConsultaDest'), metodo='NfeConsultaDest', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeConsultaDest'), metodo='NfeConsultaDest', dados=raiz
)
return self._post(url, xml) return self._post(url, xml)
@ -162,8 +210,15 @@ class ComunicacaoSefaz(Comunicacao):
pass pass
def consulta_cadastro(self, modelo, cnpj): def consulta_cadastro(self, modelo, cnpj):
"""
Consulta de cadastro
:param modelo: Modelo da nota
:param cnpj: CNPJ da empresa
:return:
"""
# UF que utilizam a SVRS - Sefaz Virtual do RS: Para serviço de Consulta Cadastro: AC, RN, PB, SC # UF que utilizam a SVRS - Sefaz Virtual do RS: Para serviço de Consulta Cadastro: AC, RN, PB, SC
lista_svrs = ['AC','RN','PB','SC']
lista_svrs = ['AC', 'RN', 'PB', 'SC']
# RS implementa um método diferente na consulta de cadastro # RS implementa um método diferente na consulta de cadastro
if self.uf.upper() == 'RS': if self.uf.upper() == 'RS':
url = NFE['RS']['CADASTRO'] url = NFE['RS']['CADASTRO']
@ -179,35 +234,51 @@ class ComunicacaoSefaz(Comunicacao):
etree.SubElement(info, 'xServ').text = 'CONS-CAD' etree.SubElement(info, 'xServ').text = 'CONS-CAD'
etree.SubElement(info, 'UF').text = self.uf.upper() etree.SubElement(info, 'UF').text = self.uf.upper()
etree.SubElement(info, 'CNPJ').text = cnpj etree.SubElement(info, 'CNPJ').text = cnpj
#etree.SubElement(info, 'CPF').text = cpf
# etree.SubElement(info, 'CPF').text = cpf
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='CadConsultaCadastro2'), metodo='CadConsultaCadastro2', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='CadConsultaCadastro2'), metodo='CadConsultaCadastro2', dados=raiz
)
# Chama método que efetua a requisição POST no servidor SOAP # Chama método que efetua a requisição POST no servidor SOAP
return self._post(url, xml) return self._post(url, xml)
def evento(self, modelo, evento, idlote=1):
""" Envia um evento de nota fiscal (cancelamento e carta de correção)"""
def evento(self, modelo, evento, id_lote=1):
"""
Envia um evento de nota fiscal (cancelamento e carta de correção)
:param modelo: Modelo da nota
:param evento: Eventro
:param id_lote: Id do lote
:return:
"""
# url do serviço # url do serviço
try: try:
# manifestacao url é do AN # manifestacao url é do AN
if evento[0][5].text.startswith('2'): if evento[0][5].text.startswith('2'):
url = self._get_url_AN(consulta='EVENTOS')
url = self._get_url_an(consulta='EVENTOS')
else: else:
url = self._get_url(modelo=modelo, consulta='EVENTOS') url = self._get_url(modelo=modelo, consulta='EVENTOS')
except Exception: except Exception:
url = self._get_url(modelo=modelo, consulta='EVENTOS') url = self._get_url(modelo=modelo, consulta='EVENTOS')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('envEvento', versao='1.00', xmlns=NAMESPACE_NFE) raiz = etree.Element('envEvento', versao='1.00', xmlns=NAMESPACE_NFE)
etree.SubElement(raiz, 'idLote').text = str(idlote) # numero autoincremental gerado pelo sistema
etree.SubElement(raiz, 'idLote').text = str(id_lote) # numero autoincremental gerado pelo sistema
raiz.append(evento) raiz.append(evento)
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='RecepcaoEvento'), metodo='RecepcaoEvento', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='RecepcaoEvento'), metodo='RecepcaoEvento', dados=raiz
)
return self._post(url, xml) return self._post(url, xml)
def status_servico(self, modelo): def status_servico(self, modelo):
""" Verifica status do servidor da receita. """
""" modelo é a string com tipo de serviço que deseja consultar
Ex: nfe ou nfce
""" """
Verifica status do servidor da receita.
:param modelo: modelo é a string com tipo de serviço que deseja consultar, Ex: nfe ou nfce
:return:
"""
url = self._get_url(modelo=modelo, consulta='STATUS') url = self._get_url(modelo=modelo, consulta='STATUS')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
@ -215,19 +286,28 @@ class ComunicacaoSefaz(Comunicacao):
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente) etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
etree.SubElement(raiz, 'cUF').text = CODIGOS_ESTADOS[self.uf.upper()] etree.SubElement(raiz, 'cUF').text = CODIGOS_ESTADOS[self.uf.upper()]
etree.SubElement(raiz, 'xServ').text = 'STATUS' etree.SubElement(raiz, 'xServ').text = 'STATUS'
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeStatusServico2'), metodo='NfeStatusServico2', dados=raiz)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeStatusServico2'), metodo='NfeStatusServico2', dados=raiz
)
# Chama método que efetua a requisição POST no servidor SOAP # Chama método que efetua a requisição POST no servidor SOAP
return self._post(url, xml) return self._post(url, xml)
def download(self, cnpj, chave): def download(self, cnpj, chave):
""" """
Metodo para download de NFe por parte de destinatário.
O certificado digital deve ser o mesmo do destinatário da Nfe.
NT 2012/002
Metodo para download de NFe por parte de destinatário.
O certificado digital deve ser o mesmo do destinatário da Nfe.
NT 2012/002
:param cnpj: CNPJ da empresa
:param chave: Chave
:return:
""" """
# url do serviço # url do serviço
url = self._get_url_AN(consulta='DOWNLOAD')
url = self._get_url_an(consulta='DOWNLOAD')
# Monta XML do corpo da requisição # Monta XML do corpo da requisição
raiz = etree.Element('downloadNFe', versao='1.00', xmlns=NAMESPACE_NFE) raiz = etree.Element('downloadNFe', versao='1.00', xmlns=NAMESPACE_NFE)
etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente) etree.SubElement(raiz, 'tpAmb').text = str(self._ambiente)
@ -235,15 +315,29 @@ class ComunicacaoSefaz(Comunicacao):
etree.SubElement(raiz, 'CNPJ').text = str(cnpj) etree.SubElement(raiz, 'CNPJ').text = str(cnpj)
etree.SubElement(raiz, 'chNFe').text = str(chave) etree.SubElement(raiz, 'chNFe').text = str(chave)
# Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeDownloadNF'), metodo='NfeDownloadNF', dados=raiz)
# Monta XML para envio da requisição
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeDownloadNF'), metodo='NfeDownloadNF', dados=raiz
)
return self._post(url, xml) return self._post(url, xml)
def inutilizacao(self, modelo, cnpj, numero_inicial, numero_final, justificativa='', ano=None, serie='1'): def inutilizacao(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. """
"""
Serviço destinado ao atendimento de solicitações de inutilização de numeração.
:param modelo: Modelo da nota
:param cnpj: CNPJda empresa
:param numero_inicial: Número inicial
:param numero_final: Número final
:param justificativa: Justificativa
:param ano: Ano
:param serie: Série
:return:
"""
# url do servico # url do servico
url = self._get_url(modelo=modelo, consulta='INUTILIZACAO') url = self._get_url(modelo=modelo, consulta='INUTILIZACAO')
# Valores default # Valores default
ano = str(ano or datetime.date.today().year)[-2:] ano = str(ano or datetime.date.today().year)[-2:]
uf = CODIGOS_ESTADOS[self.uf.upper()] uf = CODIGOS_ESTADOS[self.uf.upper()]
@ -251,15 +345,15 @@ class ComunicacaoSefaz(Comunicacao):
# Identificador da TAG a ser assinada formada com Código da UF + Ano (2 posições) + # 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” # 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': cnpj,
'modelo': '55',
'serie': serie.zfill(3),
'num_ini': str(numero_inicial).zfill(9),
'num_fin': str(numero_final).zfill(9),
}
id_unico = 'ID%(uf)s%(ano)s%(cnpj)s%(modelo)s%(serie)s%(num_ini)s%(num_fin)s' % {
'uf': uf,
'ano': ano,
'cnpj': cnpj,
'modelo': '55',
'serie': 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 # Monta XML do corpo da requisição # FIXME
raiz = etree.Element('inutNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE) raiz = etree.Element('inutNFe', versao=VERSAO_PADRAO, xmlns=NAMESPACE_NFE)
@ -280,11 +374,14 @@ class ComunicacaoSefaz(Comunicacao):
xml = a1.assinar(raiz) xml = a1.assinar(raiz)
# Monta XML para envio da requisição # Monta XML para envio da requisição
xml = self._construir_xml_status_pr(cabecalho=self._cabecalho_soap(metodo='NfeInutilizacao2'), metodo='NfeInutilizacao2', dados=xml)
xml = self._construir_xml_status_pr(
cabecalho=self._cabecalho_soap(metodo='NfeInutilizacao2'), metodo='NfeInutilizacao2', dados=xml
)
# Faz request no Servidor da Sefaz e retorna resposta # Faz request no Servidor da Sefaz e retorna resposta
return self._post(url, xml) return self._post(url, xml)
def _get_url_AN(self, consulta):
def _get_url_an(self, consulta):
# producao # producao
if self._ambiente == 1: if self._ambiente == 1:
if consulta == 'DISTRIBUICAO': if consulta == 'DISTRIBUICAO':
@ -301,7 +398,7 @@ class ComunicacaoSefaz(Comunicacao):
def _get_url(self, modelo, consulta): def _get_url(self, modelo, consulta):
""" Retorna a url para comunicação com o webservice """ """ Retorna a url para comunicação com o webservice """
# estado que implementam webservices proprios # estado que implementam webservices proprios
lista = ['PR','MS','SP','AM','CE','BA','GO','MG','MT','PE','RS']
lista = ['PR', 'MS', 'SP', 'AM', 'CE', 'BA', 'GO', 'MG', 'MT', 'PE', 'RS']
if self.uf.upper() in lista: if self.uf.upper() in lista:
if self._ambiente == 1: if self._ambiente == 1:
ambiente = 'HTTPS' ambiente = 'HTTPS'
@ -321,7 +418,7 @@ class ComunicacaoSefaz(Comunicacao):
raise Exception('Modelo não encontrado! Defina modelo="nfe" ou "nfce"') raise Exception('Modelo não encontrado! Defina modelo="nfe" ou "nfce"')
# Estados que utilizam outros ambientes # Estados que utilizam outros ambientes
else: else:
lista_svrs = ['AC','RN','PB','SC','SE']
lista_svrs = ['AC', 'RN', 'PB', 'SC', 'SE']
if self.uf.upper() in lista_svrs: if self.uf.upper() in lista_svrs:
if self._ambiente == 1: if self._ambiente == 1:
ambiente = 'HTTPS' ambiente = 'HTTPS'
@ -340,8 +437,8 @@ class ComunicacaoSefaz(Comunicacao):
def _get_url_uf(self, modelo, consulta): def _get_url_uf(self, modelo, consulta):
""" Estados que implementam url diferente do padrão nacional""" """ Estados que implementam url diferente do padrão nacional"""
# estados que implementam webservice SVRS # estados que implementam webservice SVRS
svrs = ['AC','AL','AP','DF','ES','PB','RJ','RN','RO','RR','SC','SE','TO']
svan = ['MA','PA','PI']
svrs = ['AC', 'AL', 'AP', 'DF', 'ES', 'PB', 'RJ', 'RN', 'RO', 'RR', 'SC', 'SE', 'TO']
svan = ['MA', 'PA', 'PI']
# SVRS # SVRS
if self.uf.upper() in svrs: if self.uf.upper() in svrs:
if self._ambiente == 1: if self._ambiente == 1:
@ -398,10 +495,10 @@ class ComunicacaoSefaz(Comunicacao):
def _construir_xml_soap(self, cabecalho, metodo, dados): def _construir_xml_soap(self, cabecalho, metodo, dados):
"""Mota o XML para o envio via SOAP""" """Mota o XML para o envio via SOAP"""
raiz = etree.Element('{%s}Envelope'%NAMESPACE_SOAP, nsmap={'soap12': NAMESPACE_SOAP})
c= etree.SubElement(raiz, '{%s}Header'%NAMESPACE_SOAP)
raiz = etree.Element('{%s}Envelope' % NAMESPACE_SOAP, nsmap={'soap12': NAMESPACE_SOAP})
c = etree.SubElement(raiz, '{%s}Header' % NAMESPACE_SOAP)
c.append(cabecalho) c.append(cabecalho)
body = etree.SubElement(raiz, '{%s}Body'%NAMESPACE_SOAP)
body = etree.SubElement(raiz, '{%s}Body' % NAMESPACE_SOAP)
a = etree.SubElement(body, 'nfeDadosMsg', xmlns=NAMESPACE_METODO+metodo) a = etree.SubElement(body, 'nfeDadosMsg', xmlns=NAMESPACE_METODO+metodo)
a.append(dados) a.append(dados)
return raiz return raiz
@ -409,11 +506,13 @@ class ComunicacaoSefaz(Comunicacao):
def _construir_xml_status_pr(self, cabecalho, metodo, dados): def _construir_xml_status_pr(self, cabecalho, metodo, dados):
u"""Mota o XML para o envio via SOAP""" u"""Mota o XML para o envio via SOAP"""
raiz = etree.Element('{%s}Envelope' % NAMESPACE_SOAP, nsmap={'xsi': NAMESPACE_XSI, 'xsd': NAMESPACE_XSD,'soap': NAMESPACE_SOAP})
raiz = etree.Element('{%s}Envelope' % NAMESPACE_SOAP, nsmap={
'xsi': NAMESPACE_XSI, 'xsd': NAMESPACE_XSD,'soap': NAMESPACE_SOAP
})
c = etree.SubElement(raiz, '{%s}Header' % NAMESPACE_SOAP) c = etree.SubElement(raiz, '{%s}Header' % NAMESPACE_SOAP)
c.append(cabecalho) c.append(cabecalho)
body = etree.SubElement(raiz, '{%s}Body' % NAMESPACE_SOAP) body = etree.SubElement(raiz, '{%s}Body' % NAMESPACE_SOAP)
a = etree.SubElement(body, 'nfeDadosMsg', xmlns=NAMESPACE_METODO+metodo)
a = etree.SubElement(body, 'nfeDadosMsg', xmlns=NAMESPACE_METODO + metodo)
a.append(dados) a.append(dados)
return raiz return raiz
@ -422,19 +521,22 @@ class ComunicacaoSefaz(Comunicacao):
return { return {
u'content-type': u'application/soap+xml; charset=utf-8;', u'content-type': u'application/soap+xml; charset=utf-8;',
u'Accept': u'application/soap+xml; charset=utf-8;', u'Accept': u'application/soap+xml; charset=utf-8;',
}
}
def _post(self, url, xml): def _post(self, url, xml):
certificadoA1 = CertificadoA1(self.certificado)
chave, cert = certificadoA1.separar_arquivo(self.certificado_senha, caminho=True)
certificado_a1 = CertificadoA1(self.certificado)
chave, cert = certificado_a1.separar_arquivo(self.certificado_senha, caminho=True)
chave_cert = (cert, chave) chave_cert = (cert, chave)
# Abre a conexão HTTPS # Abre a conexão HTTPS
try: try:
xml_declaration='<?xml version="1.0" encoding="UTF-8"?>'
xml_declaration = '<?xml version="1.0" encoding="UTF-8"?>'
# limpa xml com caracteres bugados para infNFeSupl em NFC-e # limpa xml com caracteres bugados para infNFeSupl em NFC-e
xml = re.sub('<qrCode>(.*?)</qrCode>',
lambda x:x.group(0).replace('&lt;','<').replace('&gt;','>').replace('amp;',''),
etree.tostring(xml, encoding='unicode').replace('\n',''))
xml = re.sub(
'<qrCode>(.*?)</qrCode>',
lambda x: x.group(0).replace('&lt;', '<').replace('&gt;', '>').replace('amp;', ''),
etree.tostring(xml, encoding='unicode').replace('\n', '')
)
xml = xml_declaration + xml xml = xml_declaration + xml
# adapter para substituir ssl por tls # adapter para substituir ssl por tls
@ -454,7 +556,7 @@ class ComunicacaoSefaz(Comunicacao):
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
raise e raise e
finally: finally:
certificadoA1.excluir()
certificado_a1.excluir()
class ComunicacaoNfse(Comunicacao): class ComunicacaoNfse(Comunicacao):
@ -570,9 +672,10 @@ class ComunicacaoNfse(Comunicacao):
""" Monta o XML do cabeçalho da requisição wsdl """ Monta o XML do cabeçalho da requisição wsdl
Namespaces padrão homologação (Ginfes) """ Namespaces padrão homologação (Ginfes) """
xml_declaration='<?xml version="1.0" encoding="UTF-8"?>'
# cabecalho = '<ns2:cabecalho versao="3" xmlns:ns2="http://www.ginfes.com.br/cabecalho_v03.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><versaoDados>3</versaoDados></ns2:cabecalho>'
# cabecalho
xml_declaration = '<?xml version="1.0" encoding="UTF-8"?>'
# cabecalho = '<ns2:cabecalho versao="3" xmlns:ns2="http://www.ginfes.com.br/cabecalho_v03.xsd"
# xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"><versaoDados>3</versaoDados></ns2:cabecalho>'
# cabecalho
raiz = etree.Element('{%s}cabecalho'%self._namespace, nsmap={'ns2':self._namespace, 'xsi':NAMESPACE_XSI}, versao=self._versao) raiz = etree.Element('{%s}cabecalho'%self._namespace, nsmap={'ns2':self._namespace, 'xsi':NAMESPACE_XSI}, versao=self._versao)
etree.SubElement(raiz, 'versaoDados').text = self._versao etree.SubElement(raiz, 'versaoDados').text = self._versao
@ -587,14 +690,14 @@ class ComunicacaoNfse(Comunicacao):
""" Monta o XML do cabeçalho da requisição wsdl """ Monta o XML do cabeçalho da requisição wsdl
Namespaces que funcionaram em produção (Ginfes)""" Namespaces que funcionaram em produção (Ginfes)"""
xml_declaration='<?xml version="1.0" encoding="UTF-8"?>'
xml_declaration = '<?xml version="1.0" encoding="UTF-8"?>'
# cabecalho # cabecalho
raiz = etree.Element('cabecalho', xmlns=self._namespace, versao=self._versao) raiz = etree.Element('cabecalho', xmlns=self._namespace, versao=self._versao)
etree.SubElement(raiz, 'versaoDados').text = self._versao etree.SubElement(raiz, 'versaoDados').text = self._versao
if retorna_string: if retorna_string:
cabecalho = etree.tostring(raiz, encoding='unicode', pretty_print=False).replace('\n','')
cabecalho = etree.tostring(raiz, encoding='unicode', pretty_print=False).replace('\n', '')
cabecalho = xml_declaration + cabecalho cabecalho = xml_declaration + cabecalho
return cabecalho return cabecalho
else: else:
@ -673,7 +776,7 @@ class ComunicacaoNfse(Comunicacao):
# versão 2 # versão 2
return cliente.service.CancelarNfse(xml) return cliente.service.CancelarNfse(xml)
# versão 3 # versão 3
#return cliente.service.CancelarNfseV3(cabecalho, xml)
# return cliente.service.CancelarNfseV3(cabecalho, xml)
# TODO outros metodos # TODO outros metodos
else: else:
raise Exception('Método não implementado no autorizador.') raise Exception('Método não implementado no autorizador.')

Write Preview
Loading…
Cancel
Save