|
|
@ -6,6 +6,7 @@ |
|
|
import os |
|
|
import os |
|
|
import requests |
|
|
import requests |
|
|
from lxml import etree |
|
|
from lxml import etree |
|
|
|
|
|
from .patch import nfeInutilizacaoCE, has_patch |
|
|
from .assinatura import Assinatura |
|
|
from .assinatura import Assinatura |
|
|
from pytrustnfe.xml import render_xml, sanitize_response |
|
|
from pytrustnfe.xml import render_xml, sanitize_response |
|
|
from pytrustnfe.utils import gerar_chave, ChaveNFe |
|
|
from pytrustnfe.utils import gerar_chave, ChaveNFe |
|
|
@ -65,11 +66,7 @@ def _render(certificado, method, sign, **kwargs): |
|
|
return xml_send |
|
|
return xml_send |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _send(certificado, method, **kwargs): |
|
|
|
|
|
xml_send = kwargs["xml"] |
|
|
|
|
|
base_url = localizar_url( |
|
|
|
|
|
method, kwargs['estado'], kwargs['modelo'], kwargs['ambiente']) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_session(certificado): |
|
|
cert, key = extract_cert_and_key_from_pfx( |
|
|
cert, key = extract_cert_and_key_from_pfx( |
|
|
certificado.pfx, certificado.password) |
|
|
certificado.pfx, certificado.password) |
|
|
cert, key = save_cert_key(cert, key) |
|
|
cert, key = save_cert_key(cert, key) |
|
|
@ -77,16 +74,32 @@ def _send(certificado, method, **kwargs): |
|
|
session = Session() |
|
|
session = Session() |
|
|
session.cert = (cert, key) |
|
|
session.cert = (cert, key) |
|
|
session.verify = False |
|
|
session.verify = False |
|
|
transport = Transport(session=session) |
|
|
|
|
|
|
|
|
return session |
|
|
|
|
|
|
|
|
parser = etree.XMLParser(strip_cdata=False) |
|
|
|
|
|
xml = etree.fromstring(xml_send, parser=parser) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _get_client(base_url, transport): |
|
|
client = Client(base_url, transport=transport) |
|
|
client = Client(base_url, transport=transport) |
|
|
|
|
|
|
|
|
port = next(iter(client.wsdl.port_types)) |
|
|
port = next(iter(client.wsdl.port_types)) |
|
|
first_operation = [x for x in iter( |
|
|
first_operation = [x for x in iter( |
|
|
client.wsdl.port_types[port].operations) if "zip" not in x.lower()][0] |
|
|
client.wsdl.port_types[port].operations) if "zip" not in x.lower()][0] |
|
|
|
|
|
return first_operation, client |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _send(certificado, method, **kwargs): |
|
|
|
|
|
xml_send = kwargs["xml"] |
|
|
|
|
|
base_url = localizar_url( |
|
|
|
|
|
method, kwargs['estado'], kwargs['modelo'], kwargs['ambiente']) |
|
|
|
|
|
session = _get_session(certificado) |
|
|
|
|
|
if has_patch: |
|
|
|
|
|
return nfeInutilizacaoCE(session, xml_send) |
|
|
|
|
|
transport = Transport(session=session) |
|
|
|
|
|
first_op, client = _get_client(base_url, transport) |
|
|
|
|
|
return _send_zeep(first_op, client, xml_send) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _send_zeep(first_operation, client, xml_send): |
|
|
|
|
|
parser = etree.XMLParser(strip_cdata=False) |
|
|
|
|
|
xml = etree.fromstring(xml_send, parser=parser) |
|
|
|
|
|
|
|
|
namespaceNFe = xml.find(".//{http://www.portalfiscal.inf.br/nfe}NFe") |
|
|
namespaceNFe = xml.find(".//{http://www.portalfiscal.inf.br/nfe}NFe") |
|
|
if namespaceNFe is not None: |
|
|
if namespaceNFe is not None: |
|
|
|