import sys import base64 import warnings import click import logging.config logging.config.dictConfig({ 'version': 1, 'formatters': { 'verbose': { 'format': '%(name)s: %(message)s' } }, 'handlers': { 'console': { 'level': 'DEBUG', 'class': 'logging.StreamHandler', 'formatter': 'verbose', }, }, 'loggers': { 'zeep.transports': { 'level': 'DEBUG', 'propagate': True, 'handlers': ['console'], }, } }) # MACHETE se corrige # lxml.etree.DocumentInvalid: Element '{http://www.w3.org/2000/09/xmldsig#}X509SerialNumber': '34255301462796514282327995225552892834' is not a valid value of the atomic type 'xs:integer'. import xmlsig, os.path path_schema = os.path.join(os.path.dirname(xmlsig.__file__), 'data', 'xmldsig-core-schema.xsd') with open(path_schema, 'r+') as f: content = f.read().replace('', '') f.seek(0) f.write(content) warnings.warn("!!MACHETE fix xmlsig X509SerialNumber type") @click.command() @click.option('--nit', required=True) @click.option('--nit-proveedor', required=True) @click.option('--id-software', required=True) @click.option('--username', required=True) @click.option('--password', required=True) def consultaResolucionesFacturacion(nit, nit_proveedor, id_software, username, password): from facho.fe.client import dian client_dian = dian.DianClient(username, password) resp = client_dian.request(dian.ConsultaResolucionesFacturacionPeticion( nit, nit_proveedor, id_software )) 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) @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_sync(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.SendBillSync if habilitacion: req = dian.Habilitacion.SendBillSync 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) @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("StatusCode:", resp.StatusCode) print("StatusDescription:", resp.StatusDescription) print("===ERRORES NOTIFICADOS====") for error_msg in resp.ErrorMessage: print("*", error_msg) @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(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.GetStatus if habilitacion: req = dian.Habilitacion.GetStatus resp = client.request(req( trackId = track_id )) 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('--nit', required=True) @click.option('--nit-proveedor', required=True) @click.option('--id-software', required=True) def soap_get_numbering_range(private_key, public_key, habilitacion, password, nit, nit_proveedor, id_software): from facho.fe.client import dian client = dian.DianSignatureClient(private_key, public_key, password=password) req = dian.GetNumberingRange if habilitacion: req = dian.Habilitacion.GetNumberingRange resp = client.request(req( nit, nit_proveedor, id_software )) print(resp) @click.command() @click.argument('invoice_path') def validate_invoice(invoice_path): from facho.fe.data.dian import XSD content = open(invoice_path, 'r').read() # TODO donde ubicar esta responsabilidad? # esto es requerido por el XSD de la DIAN content = content.replace( 'xmlns:fe="http://www.dian.gov.co/contratos/facturaelectronica/v1"', 'xmlns:fe="urn:oasis:names:specification:ubl:schema:xsd:Invoice-2"' ) XSD.validate(content, XSD.UBLInvoice) @click.command() @click.option('--private-key', type=click.Path(exists=True)) @click.option('--generate/--validate', default=False) @click.option('--passphrase') @click.option('--ssl/--no-ssl', default=False) @click.argument('scriptname', type=click.Path(exists=True), required=True) def generate_invoice(private_key, passphrase, scriptname, generate=False, ssl=True): """ imprime xml en pantalla. SCRIPTNAME espera def invoice() -> form.Invoice def extensions(form.Invoice): -> List[facho.FachoXMLExtension] """ # MACHETE if not ssl: import ssl if getattr(ssl, '_create_unverified_context', None): ssl._create_default_https_context = ssl._create_unverified_context warnings.warn("be sure!! ssl disable") else: warnings.warn("can't disable ssl") import importlib.util spec = importlib.util.spec_from_file_location('invoice', scriptname) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) import facho.fe.form as form from facho import fe invoice = module.invoice() invoice.calculate() validator = form.DianResolucion0001Validator() if not validator.validate(invoice): for error in validator.errors: print("ERROR:", error) if generate: xml = form.DIANInvoiceXML(invoice) extensions = module.extensions(invoice) for extension in extensions: xml.add_extension(extension) if private_key: signer = fe.DianXMLExtensionSigner(private_key, passphrase=passphrase) xml.add_extension(signer) print(xml.tostring(xml_declaration=True)) @click.group() 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(soap_get_numbering_range) main.add_command(generate_invoice) main.add_command(validate_invoice)