OneTeam/oc-facho
15
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

se adiciona comandos soap send-bill-async,send-bill-sync,get-status-zip

Browse Source 
FossilOrigin-Name: e3ab64b71d8bfb3acc50f813aec9b97c45cd80c2c4f38e23804d1f17f058ffdf
This commit is contained in:
bit4bit@riseup.net 2020-05-29 18:32:31 +00:00
parent 8c53f91940
commit b06537ac75
6 changed files with 108 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
facho/cli.py
View File

@ -44,6 +44,48 @@ def consultaResolucionesFacturacion(nit, nit_proveedor, id_software, username, p
print(str(resp))
@click.command()
@click.option('--private-key', required=True)
@click.option('--public-key', required=True)
@click.option('--habilitacion/--produccion', default=False)
@click.option('--password')
@click.option('--test-setid', required=True)
@click.argument('filename', required=True)
@click.argument('zipfile', type=click.Path(exists=True))
def soap_send_test_set_async(private_key, public_key, habilitacion, password, test_setid, filename, zipfile):
from facho.fe.client import dian
client = dian.DianSignatureClient(private_key, public_key, password=password)
req = dian.SendTestSetAsync
if habilitacion:
req = dian.Habilitacion.SendTestSetAsync
resp = client.request(req(
filename,
open(zipfile, 'rb').read(),
test_setid,
))
print(resp)
@click.command()
@click.option('--private-key', required=True)
@click.option('--public-key', required=True)
@click.option('--habilitacion/--produccion', default=False)
@click.option('--password')
@click.argument('filename', required=True)
@click.argument('zipfile', type=click.Path(exists=True))
def soap_send_bill_async(private_key, public_key, habilitacion, password, filename, zipfile):
from facho.fe.client import dian
client = dian.DianSignatureClient(private_key, public_key, password=password)
req = dian.SendBillAsync
if habilitacion:
req = dian.Habilitacion.SendBillAsync
resp = client.request(req(
filename,
open(zipfile, 'rb').read()
))
print(resp)
@click.command()
@click.option('--private-key', required=True)
@click.option('--public-key', required=True)
@ -64,6 +106,24 @@ def soap_send_bill_sync(private_key, public_key, habilitacion, password, filenam
))
print(resp)
@click.command()
@click.option('--private-key', required=True)
@click.option('--public-key', required=True)
@click.option('--habilitacion/--produccion', default=False)
@click.option('--password')
@click.option('--track-id', required=True)
def soap_get_status_zip(private_key, public_key, habilitacion, password, track_id):
from facho.fe.client import dian
client = dian.DianSignatureClient(private_key, public_key, password=password)
req = dian.GetStatusZip
if habilitacion:
req = dian.Habilitacion.GetStatusZip
resp = client.request(req(
trackId = track_id
))
print(resp)
@click.command()
@click.option('--private-key', required=True)
@click.option('--public-key', required=True)
@ -113,7 +173,7 @@ def generate_invoice(private_key, passphrase, scriptname):
if private_key:
signer = fe.DianXMLExtensionSigner(private_key, passphrase=passphrase)
xml.add_extension(signer)
print(str(xml))
print(xml.tostring(xml_declaration=True))
@click.group()
@ -121,6 +181,9 @@ def main():
pass
main.add_command(consultaResolucionesFacturacion)
main.add_command(soap_send_test_set_async)
main.add_command(soap_send_bill_async)
main.add_command(soap_send_bill_sync)
main.add_command(soap_get_status)
main.add_command(soap_get_status_zip)
main.add_command(generate_invoice)

10
facho/facho.py
View File

@ -98,8 +98,10 @@ class LXMLBuilder:
def set_attribute(self, elem, key, value):
elem.attrib[key] = value
def tostring(self, elem):
return tostring(elem).decode('utf-8')
def tostring(self, elem, **attrs):
attrs['pretty_print'] = attrs.pop('pretty_print', True)
attrs['encoding'] = attrs.pop('encoding', 'UTF-8')
return tostring(elem, **attrs).decode('utf-8')
class FachoXML:
@ -207,8 +209,8 @@ class FachoXML:
text = self.builder.get_text(elem)
return format_(text)
def tostring(self):
return self.builder.tostring(self.root)
def tostring(self, **kw):
return self.builder.tostring(self.root, **kw)
def __str__(self):
return self.tostring()

44
facho/fe/client/dian.py
View File

@ -80,7 +80,7 @@ class ConsultaResolucionesFacturacionPeticion(SOAPService):
return ConsultaResolucionesFacturacionRespuesta.fromdict(as_dict)
@dataclass
class SendBillAsync:
class SendBillAsync(SOAPService):
fileName: str
contentFile: str
@ -91,7 +91,7 @@ class SendBillAsync:
return 'SendBillAsync'
def build_response(self, as_dict):
return {}
return as_dict
@ -108,7 +108,7 @@ class SendTestSetAsync(SOAPService):
return 'SendTestSetAsync'
def build_response(self, as_dict):
return {}
return as_dict
@dataclass
class SendBillSync(SOAPService):
@ -122,7 +122,7 @@ class SendBillSync(SOAPService):
return 'SendBillSync'
def build_response(self, as_dict):
return {}
return as_dict
@dataclass
@ -136,12 +136,31 @@ class GetStatus(SOAPService):
return 'GetStatus'
def build_response(self, as_dict):
return {}
return as_dict
@dataclass
class GetStatusZip(SOAPService):
trackId: bytes
def get_wsdl(self):
return 'https://colombia-dian-webservices-input-sbx.azurewebsites.net/WcfDianCustomerServices.svc?wsdl'
def get_service(self):
return 'GetStatusZip'
def build_response(self, as_dict):
return as_dict
class Habilitacion:
WSDL = 'https://vpfe-hab.dian.gov.co/WcfDianCustomerServices.svc?wsdl'
class SendBillAsync(SendBillAsync):
def get_wsdl(self):
return Habilitacion.WSDL
class SendBillSync(SendBillSync):
def get_wsdl(self):
return Habilitacion.WSDL
@ -153,6 +172,10 @@ class Habilitacion:
class GetStatus(GetStatus):
def get_wsdl(self):
return Habilitacion.WSDL
class GetStatusZip(GetStatusZip):
def get_wsdl(self):
return Habilitacion.WSDL
class DianGateway:
@ -200,12 +223,11 @@ class DianSignatureClient(DianGateway):
from zeep.wsse import utils
client = zeep.Client(service.get_wsdl(), wsse=
[
BinarySignature(
self.private_key_path, self.public_key_path, self.password,
signature_method=xmlsec.Transform.RSA_SHA256,
digest_method=xmlsec.Transform.SHA256)
],
BinarySignature(
self.private_key_path, self.public_key_path, self.password,
signature_method=xmlsec.Transform.RSA_SHA256,
digest_method=xmlsec.Transform.SHA256)
,
)
return client

3
facho/fe/client/wsse/signature.py
View File

@ -346,6 +346,9 @@ def _verify_envelope_with_key(envelope, key):
security = header.find(QName(ns.WSSE, "Security"))
signature = security.find(QName(ns.DS, "Signature"))
# la DIAN no cumple a cabalidad token-profile 1.0
if signature is None:
return SignatureVerificationFailed()
ctx = xmlsec.SignatureContext()

1
facho/fe/fe.py
View File

@ -55,6 +55,7 @@ class DianXMLExtensionCUFE(FachoXMLExtension):
def build(self, fachoxml):
cufe = self._generate_cufe(self.invoice, fachoxml)
fachoxml.set_element('/fe:Invoice/cbc:UUID[schemaName="CUFE-SHA384"]', cufe)
fachoxml.set_element('/fe:Invoice/cbc:ProfileID', 'DIAN 2.1')
fachoxml.set_element('/fe:Invoice/cbc:ProfileExecutionID', self._tipo_ambiente())
return '', []

1
facho/fe/form.py
View File

@ -181,6 +181,7 @@ class DIANInvoiceXML(fe.FeXML):
invoice.calculate()
fexml.set_element('/fe:Invoice/cbc:UBLVersionID', 'UBL 2.1')
fexml.set_element('/fe:Invoice/cbc:ID', invoice.invoice_ident)
fexml.set_element('/fe:Invoice/cbc:IssueDate', invoice.invoice_issue.strftime('%Y-%m-%d'))
fexml.set_element('/fe:Invoice/cbc:IssueTime', invoice.invoice_issue.strftime('%H:%M:%S%z'))