wscdc.py

#!/usr/bin/python
# -*- coding: utf8 -*-
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation; either version 3, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY
# or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License
# for more details.

"Módulo para utilizar el servicio web Constatación de Comprobantes de AFIP"
from __future__ import print_function
from __future__ import absolute_import

# Información adicional y documentación:
# http://www.sistemasagiles.com.ar/trac/wiki/ConstatacionComprobantes

from future import standard_library

standard_library.install_aliases()
from builtins import str

__author__ = "Mariano Reingart (reingart@gmail.com)"
__copyright__ = "Copyright (C) 2013-2021 Mariano Reingart"
__license__ = "LGPL-3.0-or-later"
__version__ = "3.03a"

import sys, os, time
from configparser import SafeConfigParser
from pyafipws.utils import inicializar_y_capturar_excepciones, BaseWS, get_install_dir
from pyafipws.utils import leer, escribir, leer_dbf, guardar_dbf, N, A, I, json


# Constantes (si se usa el script de linea de comandos)
WSDL = "https://wswhomo.afip.gov.ar/WSCDC/service.asmx?WSDL"
HOMO = False
CONFIG_FILE = "rece.ini"

# No debería ser necesario modificar nada despues de esta linea

# definición del formato del archivo de intercambio (sólo para linea de comandos):

ENCABEZADO = [
    ("tipo_reg", 1, A, u"0: encabezado"),
    ("cbte_modo", 4, A, u"Modalidad de autorización (CAI, CAE, CAEA)"),
    ("cuit_emisor", 11, A, u"CUIT del emisor del comprobante"),
    ("pto_vta", 4, N, u"Punto de Venta del comprobante"),
    ("cbte_tipo", 3, N, u"Tipo de comprobante"),
    ("cbte_nro", 8, N, u"Número de comprobante"),
    ("cbte_fch", 8, A, u"Fecha en formato AAAAMMDD"),
    ("imp_total", 15, I, u"Importe total Double (13 + 2)"),
    ("cod_autorizacion", 14, A, u"Número de CAI, CAE, CAEA"),
    ("doc_tipo_receptor", 2, A, u"Tipo de documento del receptor"),
    ("doc_nro_receptor", 20, A, u"N° de documento del receptor"),
    # campos devueltos por AFIP (respuesta)
    ("resultado", 1, A, u"Resultado (A: Aprobado, O: Observado, R: rechazado)"),
    ("fch_proceso", 14, A, u"Fecha y hora de procesamiento"),
]

OBSERVACION = [
    ("tipo_reg", 1, A, u"O: observaciones devueltas por AFIP"),
    ("code", 5, N, u"Código de Observación / Error / Evento"),
    ("msg", 255, A, u"Mensaje"),
]

EVENTO = ERROR = OBSERVACION  # misma estructura, cambia tipo de registro


class WSCDC(BaseWS):
    "Interfaz para el WebService de Constatación de Comprobantes"
    _public_methods_ = [
        "Conectar",
        "SetTicketAcceso",
        "DebugLog",
        "AnalizarXml",
        "ObtenerTagXml",
        "LoadTestXML",
        "ConstatarComprobante",
        "Dummy",
        "ConsultarModalidadComprobantes",
        "ConsultarTipoComprobantes",
        "ConsultarTipoDocumentos",
        "ConsultarTipoOpcionales",
        "SetParametros",
        "SetParametro",
        "GetParametro",
    ]
    _public_attrs_ = [
        "Token",
        "Sign",
        "Cuit",
        "ExpirationTime",
        "Version",
        "XmlRequest",
        "XmlResponse",
        "Observaciones",
        "Errores",
        "InstallDir",
        "Traceback",
        "Excepcion",
        "ErrMsg",
        "Obs",
        "AppServerStatus",
        "DbServerStatus",
        "AuthServerStatus",
        "Resultado",
        "FchProceso",
        "Observaciones",
        "Obs",
        "FechaCbte",
        "CbteNro",
        "PuntoVenta",
        "ImpTotal",
        "EmisionTipo",
        "CAE",
        "CAEA",
        "CAI",
        "DocTipo",
        "DocNro",
        "SoapFault",
        "LanzarExcepciones",
    ]
    _readonly_attrs_ = _public_attrs_[3:-1]
    _reg_progid_ = "WSCDC"
    _reg_clsid_ = "{D1B97BDD-A78C-4D51-8999-1D9A5034EC10}"

    # Variables globales para BaseWS:
    HOMO = HOMO
    WSDL = WSDL
    Version = "%s %s" % (__version__, HOMO and "Homologación" or "")

    def inicializar(self):
        BaseWS.inicializar(self)
        self.AppServerStatus = self.DbServerStatus = self.AuthServerStatus = None
        self.Resultado = self.EmisionTipo = ""
        self.CAI = self.CAE = self.CAEA = self.Vencimiento = ""
        self.CbteNro = self.PuntoVenta = self.ImpTotal = None

    def __analizar_errores(self, ret):
        "Comprueba y extrae errores si existen en la respuesta XML"
        if "Errors" in ret:
            errores = ret["Errors"]
            for error in errores:
                self.Errores.append(
                    "%s: %s"
                    % (
                        error["Err"]["Code"],
                        error["Err"]["Msg"],
                    )
                )
            self.errores = [
                {
                    "code": err["Err"]["Code"],
                    "msg": err["Err"]["Msg"].replace("\n", "").replace("\r", ""),
                }
                for err in errores
            ]
            self.ErrCode = " ".join([str(error["Err"]["Code"]) for error in errores])
            self.ErrMsg = "\n".join(self.Errores)

    @inicializar_y_capturar_excepciones
    def Dummy(self):
        "Método Dummy para verificación de funcionamiento de infraestructura"
        result = self.client.ComprobanteDummy()["ComprobanteDummyResult"]
        self.AppServerStatus = result["AppServer"]
        self.DbServerStatus = result["DbServer"]
        self.AuthServerStatus = result["AuthServer"]
        self.__analizar_errores(result)
        return True

    @inicializar_y_capturar_excepciones
    def ConstatarComprobante(
        self,
        cbte_modo,
        cuit_emisor,
        pto_vta,
        cbte_tipo,
        cbte_nro,
        cbte_fch,
        imp_total,
        cod_autorizacion,
        doc_tipo_receptor=None,
        doc_nro_receptor=None,
        **kwargs
    ):
        "Método de Constatación de Comprobantes"
        response = self.client.ComprobanteConstatar(
            Auth={"Token": self.Token, "Sign": self.Sign, "Cuit": self.Cuit},
            CmpReq={
                "CbteModo": cbte_modo,
                "CuitEmisor": cuit_emisor,
                "PtoVta": pto_vta,
                "CbteTipo": cbte_tipo,
                "CbteNro": cbte_nro,
                "CbteFch": cbte_fch,
                "ImpTotal": imp_total,
                "CodAutorizacion": cod_autorizacion,
                "DocTipoReceptor": doc_tipo_receptor,
                "DocNroReceptor": doc_nro_receptor,
            },
        )
        result = response["ComprobanteConstatarResult"]
        self.__analizar_errores(result)
        if "CmpResp" in result:
            resp = result["CmpResp"]
            self.Resultado = result["Resultado"]
            self.FchProceso = result.get("FchProceso", "")
            self.observaciones = []
            for obs in result.get("Observaciones", []):
                self.Observaciones.append("%(Code)s: %(Msg)s" % (obs["Obs"]))
                self.observaciones.append(
                    {
                        "code": obs["Obs"]["Code"],
                        "msg": obs["Obs"]["Msg"].replace("\n", "").replace("\r", ""),
                    }
                )
            self.Obs = "\n".join(self.Observaciones)
            self.FechaCbte = resp.get("CbteFch", "")  # .strftime("%Y/%m/%d")
            self.CbteNro = resp.get("CbteNro", 0)  # 1L
            self.PuntoVenta = resp.get("PtoVta", 0)  # 4000
            self.ImpTotal = str(resp["ImpTotal"])
            self.EmisionTipo = resp["CbteModo"]
            self.DocTipo = resp.get("DocTipoReceptor", "")
            self.DocNro = resp.get("DocNroReceptor", "")
            cod_aut = str(resp.get("CodAutorizacion", ""))  # 60423794871430L
            if self.EmisionTipo == "CAE":
                self.CAE = cod_aut
            elif self.EmisionTipo == "CAEA":
                self.CAEA = cod_aut
            elif self.EmisionTipo == "CAI":
                self.CAI = cod_aut
        return True

    @inicializar_y_capturar_excepciones
    def ConsultarModalidadComprobantes(self, sep="|"):
        "Recuperador de modalidades de autorización de comprobantes"
        response = self.client.ComprobantesModalidadConsultar(
            Auth={"Token": self.Token, "Sign": self.Sign, "Cuit": self.Cuit},
        )
        result = response["ComprobantesModalidadConsultarResult"]
        self.__analizar_errores(result)
        return [
            (u"\t%(Cod)s\t%(Desc)s\t" % p["FacModTipo"]).replace("\t", sep)
            for p in result["ResultGet"]
        ]

    @inicializar_y_capturar_excepciones
    def ConsultarTipoComprobantes(self, sep="|"):
        "Recuperador de valores referenciales de códigos de Tipos de comprobante"
        response = self.client.ComprobantesTipoConsultar(
            Auth={"Token": self.Token, "Sign": self.Sign, "Cuit": self.Cuit},
        )
        result = response["ComprobantesTipoConsultarResult"]
        self.__analizar_errores(result)
        return [
            (u"\t%(Id)s\t%(Desc)s\t" % p["CbteTipo"]).replace("\t", sep)
            for p in result["ResultGet"]
        ]

    @inicializar_y_capturar_excepciones
    def ConsultarTipoDocumentos(self, sep="|"):
        "Recuperador de valores referenciales de códigos de Tipos de Documentos"
        response = self.client.DocumentosTipoConsultar(
            Auth={"Token": self.Token, "Sign": self.Sign, "Cuit": self.Cuit},
        )
        result = response["DocumentosTipoConsultarResult"]
        self.__analizar_errores(result)
        return [
            (u"\t%(Id)s\t%(Desc)s\t" % p["DocTipo"]).replace("\t", sep)
            for p in result["ResultGet"]
        ]

    @inicializar_y_capturar_excepciones
    def ConsultarTipoOpcionales(self, sep="|"):
        "Recuperador de valores referenciales de códigos de Tipos de datos Opcionales"
        response = self.client.OpcionalesTipoConsultar(
            Auth={"Token": self.Token, "Sign": self.Sign, "Cuit": self.Cuit},
        )
        result = response["OpcionalesTipoConsultarResult"]
        res = result["ResultGet"] if "ResultGet" in result else []
        self.__analizar_errores(result)
        return [
            (u"\t%(Id)s\t%(Desc)s\t" % p["OpcionalTipo"]).replace("\t", sep)
            for p in res
        ]


# busco el directorio de instalación (global para que no cambie si usan otra dll)
INSTALL_DIR = WSCDC.InstallDir = get_install_dir()


def escribir_archivo(dic, nombre_archivo, agrega=True):
    archivo = open(nombre_archivo, agrega and "a" or "w")
    formatos = [
        ("Encabezado", ENCABEZADO, [dic], 0),
        ("Observacion", OBSERVACION, dic.get("observaciones", []), "O"),
        ("Eventos", ERROR, dic.get("eventos", []), "V"),
        ("Error", ERROR, dic.get("errores", []), "E"),
    ]
    if "--json" in sys.argv:
        json.dump(dic, archivo, sort_keys=True, indent=4)
    elif "--dbf" in sys.argv:
        guardar_dbf(formatos, agrega, conf_dbf)
    else:
        for nombre, formato, registros, tipo_reg in formatos:
            for it in registros:
                it["tipo_reg"] = tipo_reg
                archivo.write(escribir(it, formato))
    archivo.close()


def leer_archivo(nombre_archivo):
    archivo = open(nombre_archivo, "r")
    if "--json" in sys.argv:
        dic = json.load(archivo)
    elif "--dbf" in sys.argv:
        dic = {}
        formatos = [
            ("Encabezado", ENCABEZADO, dic),
        ]
        leer_dbf(formatos, conf_dbf)
    else:
        dic = {}
        for linea in archivo:
            if str(linea[0]) == "0":
                d = leer(linea, ENCABEZADO)
                dic.update(d)
            else:
                print("Tipo de registro incorrecto:", linea[0])
    archivo.close()

    if not "cod_autorizacion" in dic:
        raise RuntimeError(
            "Archivo de entrada invalido, revise campos y lineas en blanco"
        )

    return dic


def main():
    "Funcion principal para utilizar la interfaz por linea de comando"

    if "--formato" in sys.argv:
        print("Formato:")
        for msg, formato in [
            ("Encabezado", ENCABEZADO),
            ("Observacion", OBSERVACION),
            ("Evento", EVENTO),
            ("Error", ERROR),
        ]:
            comienzo = 1
            print("=== %s ===" % msg)
            print(
                "|| %-20s || %8s || %9s || %-12s || %-20s ||"
                % ("Campo", "Posición", "Longitud", "Tipo", "Descripción")
            )
            for fmt in formato:
                clave, longitud, tipo, desc = fmt
                print(
                    "|| %-20s || %8d || %9d || %-12s || %-20s ||"
                    % (clave, comienzo, longitud, tipo, desc.encode("latin1"))
                )
                comienzo += longitud
        return

    # leer configuracion
    global CONFIG_FILE
    if len(sys.argv) > 1 and sys.argv[1][0] not in "-/":
        CONFIG_FILE = sys.argv.pop(1)
    config = SafeConfigParser()
    if sys.version_info[0] < 3 :
        config.read(CONFIG_FILE, encoding="latin-1")
    else:
        config.read(CONFIG_FILE)
    crt = config.get("WSAA", "CERT")
    key = config.get("WSAA", "PRIVATEKEY")
    cuit = config.get("WSCDC", "CUIT")
    url_wsaa = config.get("WSAA", "URL") if config.has_option("WSAA", "URL") else ""
    url_wscdc = config.get("WSCDC", "URL") if config.has_option("WSCDC", "URL") else ""

    # leo configuración de archivos de intercambio
    ENTRADA = config.get("WSCDC", "ENTRADA")
    SALIDA = config.get("WSCDC", "SALIDA")
    if config.has_section("DBF"):
        conf_dbf = dict(config.items("DBF"))
    else:
        conf_dbf = {}

    if config.has_section("PROXY") and not HOMO:
        proxy_dict = dict(("proxy_%s" % k, v) for k, v in config.items("PROXY"))
        proxy_dict["proxy_port"] = int(proxy_dict["proxy_port"])
    else:
        proxy_dict = {}

    # instanciar la interfaz con el webservice
    wscdc = WSCDC()
    ok = wscdc.Conectar("", url_wscdc, proxy=proxy_dict)

    if "--dummy" in sys.argv:
        # print wscdc.client.help("ComprobanteDummy")
        wscdc.Dummy()
        print("AppServerStatus", wscdc.AppServerStatus)
        print("DbServerStatus", wscdc.DbServerStatus)
        print("AuthServerStatus", wscdc.AuthServerStatus)
        return

    # Gestionar credenciales de acceso con AFIP:
    from pyafipws.wsaa import WSAA

    wsaa = WSAA()
    ta = wsaa.Autenticar("wscdc", crt, key, url_wsaa, proxy=proxy_dict)
    if not ta:
        sys.exit("Imposible autenticar con WSAA: %s" % wsaa.Excepcion)
    wscdc.SetTicketAcceso(ta)
    wscdc.Cuit = cuit

    if "--constatar" in sys.argv:
        if len(sys.argv) < 8:
            if "--prueba" in sys.argv:
                dic = dict(
                    cbte_modo="CAE",
                    cuit_emisor="20267565393",
                    pto_vta=3,
                    cbte_tipo=6,
                    cbte_nro=1,
                    cbte_fch="20131231",
                    imp_total="1.21",
                    cod_autorizacion="63533727749637",
                    doc_tipo_receptor=99,
                    doc_nro_receptor=0,
                )
                # escribir archivo de intercambio con datos de prueba:
                escribir_archivo(dic, ENTRADA)
            else:
                # leer archivo de intercambio:
                dic = leer_archivo(ENTRADA)
            # constatar el comprobante
            wscdc.ConstatarComprobante(**dic)
            # actualizar el diccionario con los datos de devueltos por AFIP
            dic.update(
                {
                    "resultado": wscdc.Resultado,
                    "fch_proceso": wscdc.FchProceso,
                }
            )
            dic["observaciones"] = wscdc.observaciones
            dic["errores"] = wscdc.errores
            escribir_archivo(dic, SALIDA)
        else:
            # usar los datos pasados por linea de comandos:
            wscdc.ConstatarComprobante(*sys.argv[sys.argv.index("--constatar") + 1 :])

        print("Resultado:", wscdc.Resultado)
        print("Mensaje de Error:", wscdc.ErrMsg)
        print("Observaciones:", wscdc.Obs)

    if "--params" in sys.argv:

        print("=== Modalidad Comprobantes ===")
        print(u"\n".join(wscdc.ConsultarModalidadComprobantes("||")))
        print("=== Tipo Comprobantes ===")
        print(u"\n".join(wscdc.ConsultarTipoComprobantes("||")))
        print("=== Tipo Documentos ===")
        print(u"\n".join(wscdc.ConsultarTipoDocumentos("||")))
        print("=== Tipo Opcionales ===")
        print(u"\n".join(wscdc.ConsultarTipoOpcionales("||")))
        print("Mensaje de Error:", wscdc.ErrMsg)


if __name__ == "__main__":

    if "--register" in sys.argv or "--unregister" in sys.argv:
        import pythoncom
        import win32com.server.register

        win32com.server.register.UseCommandLine(WSCDC)
    elif "/Automate" in sys.argv:
        # MS seems to like /automate to run the class factories.
        import win32com.server.localserver

        # win32com.server.localserver.main()
        # start the server.
        win32com.server.localserver.serve([WSCDC._reg_clsid_])
    else:
        main()