diff --git a/facho/cli.py b/facho/cli.py index ab5a3b0..91bc360 100644 --- a/facho/cli.py +++ b/facho/cli.py @@ -41,9 +41,26 @@ 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('--password') +@click.argument('filename', required=True) +@click.argument('zipfile', type=click.Path(exists=True)) +def SendTestSetAsync(private_key, public_key, password, filename, zipfile): + from facho.fe.client import dian + + client = dian.DianSignatureClient(private_key, public_key, password=password) + resp = client.request(dian.SendTestSetAsync( + filename, zipfile.read().encode('utf-8') + )) + print(resp) + @click.group() def main(): pass main.add_command(consultaResolucionesFacturacion) +main.add_command(SendTestSetAsync) diff --git a/facho/fe/client/dian.py b/facho/fe/client/dian.py index b2281a8..bbfbc44 100644 --- a/facho/fe/client/dian.py +++ b/facho/fe/client/dian.py @@ -2,7 +2,7 @@ from facho import facho import zeep from zeep.wsse.username import UsernameToken - +from zeep.wsse.signature import Signature import urllib.request from datetime import datetime @@ -90,15 +90,26 @@ class SendBillAsync: def build_response(self, as_dict): return {} - -class DianClient: - def __init__(self, user, password): - self._username = user - self._password = password +@dataclass +class SendTestSetAsync: + fileName: str + contentFile: str + + def get_wsdl(self): + return 'https://colombia-dian-webservices-input-sbx.azurewebsites.net/WcfDianCustomerServices.svc?wsdl' + + def get_service(self): + return 'SendTestSetAsync' + + def build_response(self, as_dict): + return {} + + +class DianGateway: def _open(self, service): - return zeep.Client(service.get_wsdl(), wsse=UsernameToken(self._username, self._password)) + raise NotImplementedError() def _remote_service(self, conn, service): return conn.service[service.get_service()] @@ -117,5 +128,26 @@ class DianClient: return service.build_response(resp) + +class DianClient(DianGateway): + + def __init__(self, user, password): + self._username = user + self._password = password + + def _open(self, service): + return zeep.Client(service.get_wsdl(), wsse=UsernameToken(self._username, self._password)) + + +class DianSignatureClient(DianGateway): + + def __init__(self, private_key_path, public_key_path, password=None): + self.private_key_path = private_key_path + self.public_key_path = public_key_path + self.password = password + + def _open(self, service): + return zeep.Client(service.get_wsdl(), wsse=Signature( + self.private_key_path, self.public_key_path, self.password))