oc-facho/facho/cli.py

# -*- coding: utf-8 -*-
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"],
            },
        },
    }
)


def disable_ssl():
    # MACHETE
    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")


@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):
    warnings.warn("!! NO APROBADO FUNCIONAMIENTO")

    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.argument("nomina_path")
def validate_nominaindividual(nomina_path):
    from facho.fe.data.dian import XSD

    content = open(nomina_path, "r").read()
    content = content.replace(
        'xmlns="http://www.dian.gov.co/contratos/facturaelectronica/v1"',
        'xmlns="dian:gov:co:facturaelectronica:NominaIndividual"',
    )
    XSD.validate(content, XSD.NominaIndividual)


@click.command()
@click.option("--private-key", type=click.Path(exists=True))
@click.option("--passphrase")
@click.option("--ssl/--no-ssl", default=False)
@click.option("--use-cache-policy/--no-use-cache-policy", default=False)
@click.argument("xmlfile", type=click.Path(exists=True), required=True)
@click.argument("output", required=True)
def sign_xml(
    private_key,
    passphrase,
    xmlfile,
    ssl=True,
    use_cache_policy=False,
    output=None,
):
    if not ssl:
        disable_ssl()

    from facho import fe

    if use_cache_policy:
        warnings.warn("xades using cache policy")

    signer = fe.DianXMLExtensionSigner(
        private_key, passphrase=passphrase, localpolicy=use_cache_policy
    )
    document = open(xmlfile, "r").read().encode("utf-8")
    with open(output, "w") as f:
        f.write(signer.sign_xml_string(document))


@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.option("--sign/--no-sign", default=False)
@click.option("--use-cache-policy/--no-use-cache-policy", default=False)
@click.argument("scriptname", type=click.Path(exists=True), required=True)
@click.argument("output", required=True)
def generate_invoice(
    private_key,
    passphrase,
    scriptname,
    generate=False,
    ssl=True,
    sign=False,
    use_cache_policy=False,
    output=None,
):
    """
    imprime xml en pantalla.
    SCRIPTNAME espera
     def invoice() -> form.Invoice
     def extensions(form.Invoice): -> List[facho.FachoXMLExtension]
    """

    if not ssl:
        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)

    from facho.fe.form_xml import (
        DIANWriteSigned,
        DIANWrite,
        DIANSupportDocumentXML,
    )

    try:
        invoice_xml = module.document_xml()
    except AttributeError:
        # invoice_xml = DIANInvoiceXML
        invoice_xml = DIANSupportDocumentXML
    print("Using document xml:", invoice_xml)
    invoice = module.invoice()
    invoice.calculate()

    if generate:
        xml = invoice_xml(invoice)

        extensions = module.extensions(invoice)
        for extension in extensions:
            xml.add_extension(extension)

        if sign:
            DIANWriteSigned(
                xml, output, private_key, passphrase, use_cache_policy
            )
        else:
            DIANWrite(xml, output)


@click.command()
@click.option("--private-key", type=click.Path(exists=True))
@click.option("--passphrase")
@click.option("--ssl/--no-ssl", default=False)
@click.option("--sign/--no-sign", default=False)
@click.option("--use-cache-policy/--no-use-cache-policy", default=False)
@click.argument("scriptname", type=click.Path(exists=True), required=True)
@click.argument("output", required=True)
def generate_nomina(
    private_key,
    passphrase,
    scriptname,
    ssl=True,
    sign=False,
    use_cache_policy=False,
    output=None,
):
    """
    imprime xml en pantalla.
    SCRIPTNAME espera
    def nomina() -> (
    fe.nomina.NominaIndividual
    )
    def extensions(
    fe.nomina.NominaIndividual
    ): -> List[facho.FachoXMLExtension]
    """

    if not ssl:
        disable_ssl()

    import importlib.util

    spec = importlib.util.spec_from_file_location("nomina", scriptname)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    from facho.fe.form_xml import DIANWriteSigned, DIANWrite
    import facho.fe

    nomina = module.nomina()

    xml = nomina.toFachoXML()

    extensions = module.extensions(nomina)
    for extension in extensions:
        xml.add_extension(extension)

    if sign:
        DIANWriteSigned(
            xml,
            output,
            private_key,
            passphrase,
            use_cache_policy,
            dian_signer=facho.fe.nomina.DianXMLExtensionSigner,
        )
    else:
        DIANWrite(xml, output)


@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_nomina_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.SendNominaSync
    if habilitacion:
        req = dian.Habilitacion.SendNominaSync
    resp = client.request(req(open(zipfile, "rb").read()))
    print(resp)


@click.command()
@click.option("--private-key", type=click.Path(exists=True))
@click.option("--passphrase")
@click.option("--ssl/--no-ssl", default=False)
@click.option("--use-cache-policy/--no-use-cache-policy", default=False)
@click.argument("xmlfile", type=click.Path(exists=True), required=True)
def sign_verify_xml(
    private_key,
    passphrase,
    xmlfile,
    ssl=True,
    use_cache_policy=False,
    output=None,
):
    if not ssl:
        disable_ssl()

    from facho.fe import fe

    if use_cache_policy:
        warnings.warn("xades using cache policy")

    print("THIS ONLY WORKS FOR DOCUMENTS GENERATE WITH FACHO")
    signer = fe.DianXMLExtensionSignerVerifier(
        private_key, passphrase=passphrase, localpolicy=use_cache_policy
    )
    document = open(xmlfile, "r").read().encode("utf-8")

    if signer.verify_string(document):
        print("+OK")
    else:
        print("-INVALID")


@click.command()
@click.option("--software-id")
@click.option("--software-pin")
@click.option("--nit")
@click.option("--dv")
@click.option("--output-zippath")
def generate_nomina_habilitacion(
    software_id, software_pin, nit, dv, output_zippath
):
    from facho import fe

    generador = fe.nomina.habilitacion.Habilitacion(
        fe.nomina.habilitacion.Habilitacion.Metadata(
            software_id=software_id,
            software_pin=software_pin,
            nit=nit,
            dv=dv,
        )
    )
    generador.generar(output_zippath)


@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)
main.add_command(sign_xml)
main.add_command(sign_verify_xml)
main.add_command(generate_nomina)
main.add_command(soap_send_nomina_sync)
main.add_command(validate_nominaindividual)
main.add_command(generate_nomina_habilitacion)