padron.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.

"Herramienta para procesar y consultar el Padrón Unico de Contribuyentes AFIP"
from __future__ import division
from __future__ import print_function
from __future__ import absolute_import

# Documentación e información adicional:
#    http://www.sistemasagiles.com.ar/trac/wiki/PadronContribuyentesAFIP

from future import standard_library

standard_library.install_aliases()
from builtins import next
from builtins import str
from builtins import range
from past.builtins import basestring
from builtins import object
from past.utils import old_div

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


import csv
import json
import os
import shelve
import socket
import sqlite3
import urllib.request, urllib.error, urllib.parse
import zipfile
from email.utils import formatdate
import sys
import warnings
from pyafipws.utils import (
    leer,
    escribir,
    N,
    A,
    I,
    get_install_dir,
    safe_console,
    inicializar_y_capturar_excepciones_simple,
    WebClient,
    norm,
    exception_info,
)


# formato y ubicación archivo completo de la condición tributaria según RG 1817

FORMATO = [
    ("nro_doc", 11, N, ""),
    ("denominacion", 30, A, ""),
    ("imp_ganancias", 2, A, "'NI', 'AC','EX', 'NC'"),
    ("imp_iva", 2, A, "'NI' , 'AC','EX','NA','XN','AN'"),
    ("monotributo", 2, A, "'NI', 'Codigo categoria tributaria'"),
    ("integrante_soc", 1, A, "'N' , 'S'"),
    ("empleador", 1, A, "'N', 'S'"),
    ("actividad_monotributo", 2, A, ""),
    ("tipo_doc", 2, N, "80: CUIT, 96: DNI, etc."),
    ("cat_iva", 2, N, "1: RI, 4: EX, 5: CF, 6: MT, etc"),
    ("email", 250, A, ""),
]

# Mapeos constantes:

PROVINCIAS = {
    0: "CIUDAD AUTONOMA BUENOS AIRES",
    1: "BUENOS AIRES",
    2: "CATAMARCA",
    3: "CORDOBA",
    4: "CORRIENTES",
    5: "ENTRE RIOS",
    6: "JUJUY",
    7: "MENDOZA",
    8: "LA RIOJA",
    9: "SALTA",
    10: "SAN JUAN",
    11: "SAN LUIS",
    12: "SANTA FE",
    13: "SANTIAGO DEL ESTERO",
    14: "TUCUMAN",
    16: "CHACO",
    17: "CHUBUT",
    18: "FORMOSA",
    19: "MISIONES",
    20: "NEUQUEN",
    21: "LA PAMPA",
    22: "RIO NEGRO",
    23: "SANTA CRUZ",
    24: "TIERRA DEL FUEGO",
}

TIPO_CLAVE = {"CUIT": 80, "CUIL": 86, "CDI": 86, "DNI": 96, "Otro": 99}

DEBUG = True

URL = "http://www.afip.gob.ar/genericos/cInscripcion/archivos/apellidoNombreDenominacion.zip"
URL_API = "https://soa.afip.gob.ar/"


class PadronAFIP(object):
    "Interfaz para consultar situación tributaria (Constancia de Inscripcion)"

    _public_methods_ = [
        "Buscar",
        "Descargar",
        "Procesar",
        "Guardar",
        "ConsultarDomicilios",
        "Consultar",
        "Conectar",
        "DescargarConstancia",
        "MostrarPDF",
        "BuscarCUIT",
        "ObtenerTablaParametros",
    ]
    _public_attrs_ = [
        "InstallDir",
        "Traceback",
        "Excepcion",
        "Version",
        "cuit",
        "dni",
        "denominacion",
        "imp_ganancias",
        "imp_iva",
        "monotributo",
        "integrante_soc",
        "empleador",
        "actividad_monotributo",
        "cat_iva",
        "domicilios",
        "tipo_doc",
        "nro_doc",
        "LanzarExcepciones",
        "tipo_persona",
        "estado",
        "impuestos",
        "actividades",
        "direccion",
        "localidad",
        "provincia",
        "cod_postal",
        "data",
        "response",
    ]
    _readonly_attrs_ = _public_attrs_[3:-1]
    _reg_progid_ = "PadronAFIP"
    _reg_clsid_ = "{6206DF5E-3EEF-47E9-A532-CD81EBBAF3AA}"

    def __init__(self):
        self.db_path = os.path.join(self.InstallDir, "padron.db")
        self.Version = __version__
        # Abrir la base de datos
        self.db = sqlite3.connect(self.db_path)
        self.db.row_factory = sqlite3.Row
        self.cursor = self.db.cursor()
        self.LanzarExcepciones = False
        self.inicializar()
        self.client = None

    def inicializar(self):
        self.Excepcion = self.Traceback = ""
        self.cuit = self.dni = 0
        self.tipo_persona = ""  # FISICA o JURIDICA
        self.tipo_doc = 0
        self.estado = ""  # ACTIVO
        self.denominacion = ""
        self.direccion = self.localidad = self.provincia = self.cod_postal = ""
        self.domicilios = []
        self.impuestos = []
        self.actividades = []
        self.imp_iva = self.empleador = self.integrante_soc = self.cat_iva = ""
        self.monotributo = self.actividad_monotributo = ""
        self.data = {}
        self.response = ""

    @inicializar_y_capturar_excepciones_simple
    def Conectar(self, url=URL_API, proxy="", wrapper=None, cacert=None, trace=False):
        self.client = WebClient(location=url, trace=trace, cacert=cacert)
        self.client.method = "GET"  # metodo RESTful predeterminado
        self.client.enctype = None  # no enviar body
        return True

    @inicializar_y_capturar_excepciones_simple
    def Descargar(self, url=URL, filename="padron.txt", proxy=None):
        "Descarga el archivo de AFIP, devuelve 200 o 304 si no fue modificado"
        proxies = {}
        if proxy:
            proxies["http"] = proxy
            proxies["https"] = proxy
            proxy_handler = urllib.request.ProxyHandler(proxies)
        print("Abriendo URL %s ..." % url)
        req = urllib.request.Request(url)
        if os.path.exists(filename):
            http_date = formatdate(
                timeval=os.path.getmtime(filename), localtime=False, usegmt=True
            )
            req.add_header("If-Modified-Since", http_date)
        try:
            web = urllib.request.urlopen(req)
        except urllib.error.HTTPError as e:
            if e.code == 304:
                print("No modificado desde", http_date)
                return 304
            else:
                raise
        # leer info del request:
        meta = web.info()
        lenght = float(meta["Content-Length"])
        date = meta["Last-Modified"]
        tmp = open(filename + ".zip", "wb")
        print("Guardando")
        size = 0
        p0 = None
        while True:
            p = int(old_div(size, lenght) * 100)
            if p0 is None or p > p0:
                print("Leyendo ... %0d %%" % p)
                p0 = p
            data = web.read(1024 * 100)
            size = size + len(data)
            if not data:
                print("Descarga Terminada!")
                break
            tmp.write(data)
        print("Abriendo ZIP...")
        tmp.close()
        web.close()
        uf = open(filename + ".zip", "rb")
        zf = zipfile.ZipFile(uf)
        for fn in zf.namelist():
            print("descomprimiendo", fn)
            tf = open(filename, "wb")
            tf.write(zf.read(fn))
            tf.close()
        return 200

    @inicializar_y_capturar_excepciones_simple
    def Procesar(self, filename="padron.txt", borrar=False):
        "Analiza y crea la base de datos interna sqlite para consultas"
        f = open(filename, "r")
        keys = [k for k, l, t, d in FORMATO]
        # conversion a planilla csv (no usado)
        if False and not os.path.exists("padron.csv"):
            csvfile = open("padron.csv", "wb")
            import csv

            wr = csv.writer(
                csvfile, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL
            )
            for i, l in enumerate(f):
                if i % 100000 == 0:
                    print("Progreso: %d registros" % i)
                r = leer(l, FORMATO)
                row = [r[k] for k in keys]
                wr.writerow(row)
            csvfile.close()
        f.seek(0)
        if os.path.exists(self.db_path) and borrar:
            os.remove(self.db_path)
        if True:
            db = db = sqlite3.connect(self.db_path)
            c = db.cursor()
            c.execute(
                "CREATE TABLE padron ("
                "nro_doc INTEGER, "
                "denominacion VARCHAR(30), "
                "imp_ganancias VARCHAR(2), "
                "imp_iva VARCHAR(2), "
                "monotributo VARCHAR(1), "
                "integrante_soc VARCHAR(1), "
                "empleador VARCHAR(1), "
                "actividad_monotributo VARCHAR(2), "
                "tipo_doc INTEGER, "
                "cat_iva INTEGER DEFAULT NULL, "
                "email VARCHAR(250), "
                "PRIMARY KEY (tipo_doc, nro_doc)"
                ");"
            )
            c.execute(
                "CREATE TABLE domicilio ("
                "id INTEGER PRIMARY KEY AUTOINCREMENT, "
                "tipo_doc INTEGER, "
                "nro_doc INTEGER, "
                "direccion TEXT, "
                "FOREIGN KEY (tipo_doc, nro_doc) REFERENCES padron "
                ");"
            )
            # importar los datos a la base sqlite
            for i, l in enumerate(f):
                if i % 10000 == 0:
                    print(i)
                l = l.strip("\x00")
                r = leer(l, FORMATO)
                params = [r[k] for k in keys]
                params[8] = 80  # agrego tipo_doc = CUIT
                params[9] = None  # cat_iva no viene de AFIP
                placeholders = ", ".join(["?"] * len(params))
                c.execute("INSERT INTO padron VALUES (%s)" % placeholders, params)
            db.commit()
            c.close()
            db.close()

    @inicializar_y_capturar_excepciones_simple
    def Buscar(self, nro_doc, tipo_doc=80):
        "Devuelve True si fue encontrado y establece atributos con datos"
        # cuit: codigo único de identificación tributaria del contribuyente
        #       (sin guiones)
        self.cursor.execute(
            "SELECT * FROM padron WHERE " " tipo_doc=? AND nro_doc=?",
            [tipo_doc, nro_doc],
        )
        row = self.cursor.fetchone()
        for key in [k for k, l, t, d in FORMATO]:
            if row:
                val = row[key]
                if not isinstance(val, basestring):
                    val = str(row[key])
                setattr(self, key, val)
            else:
                setattr(self, key, "")
        if self.tipo_doc == 80:
            self.cuit = self.nro_doc
        elif self.tipo_doc == 96:
            self.dni = self.nro_doc
        # determinar categoría de IVA (tentativa)
        try:
            cat_iva = int(self.cat_iva)
        except ValueError:
            cat_iva = None
        if cat_iva:
            pass
        elif self.imp_iva in ("AC", "S"):
            self.cat_iva = 1  # RI
        elif self.imp_iva == "EX":
            self.cat_iva = 4  # EX
        elif self.monotributo:
            self.cat_iva = 6  # MT
        else:
            self.cat_iva = 5  # CF
        return True if row else False

    @inicializar_y_capturar_excepciones_simple
    def ConsultarDomicilios(self, nro_doc, tipo_doc=80, cat_iva=None):
        "Busca los domicilios, devuelve la cantidad y establece la lista"
        self.cursor.execute(
            "SELECT direccion FROM domicilio WHERE "
            " tipo_doc=? AND nro_doc=? ORDER BY id ",
            [tipo_doc, nro_doc],
        )
        filas = self.cursor.fetchall()
        self.domicilios = [fila["direccion"] for fila in filas]
        return len(filas)

    @inicializar_y_capturar_excepciones_simple
    def BuscarCUIT(self, denominacion, limite=10):
        "Busca por nombre, devuelve una lista con tipo y nro de documento"
        sql = (
            "SELECT tipo_doc, nro_doc, denominacion "
            "FROM padron_fts WHERE denominacion MATCH ? "
            "LIMIT ?"
        )
        denominacion = denominacion.replace(" ", "*")
        self.cursor.execute(sql, ["*{}*".format(denominacion), limite])
        filas = self.cursor.fetchall()
        return [dict(fila) for fila in filas]

    @inicializar_y_capturar_excepciones_simple
    def Guardar(self, tipo_doc, nro_doc, denominacion, cat_iva, direccion, email):
        "Agregar o actualizar los datos del cliente"
        if self.Buscar(nro_doc, tipo_doc):
            sql = (
                "UPDATE padron SET denominacion=?, cat_iva=?, email=? "
                "WHERE tipo_doc=? AND nro_doc=?"
            )
            params = [denominacion, cat_iva, email, tipo_doc, nro_doc]
        else:
            sql = (
                "INSERT INTO padron (tipo_doc, nro_doc, denominacion, "
                "cat_iva, email) VALUES (?, ?, ?, ?, ?)"
            )
            params = [tipo_doc, nro_doc, denominacion, cat_iva, email]
        self.cursor.execute(sql, params)
        # agregar el domicilio solo si no existe:
        if direccion:
            self.cursor.execute(
                "SELECT * FROM domicilio WHERE direccion=? "
                "AND tipo_doc=? AND nro_doc=?",
                [direccion, tipo_doc, nro_doc],
            )
            if self.cursor.rowcount < 0:
                sql = (
                    "INSERT INTO domicilio (nro_doc, tipo_doc, direccion)"
                    "VALUES (?, ?, ?)"
                )
                self.cursor.execute(sql, [nro_doc, tipo_doc, direccion])
        self.db.commit()
        return True

    @inicializar_y_capturar_excepciones_simple
    def Consultar(self, nro_doc):
        "Llama a la API pública de AFIP para obtener los datos de una persona"
        n = 0
        while n <= 4:
            n += 1  # reintentar 3 veces
            try:
                if not self.client:
                    if DEBUG:
                        warnings.warn("reconectando intento [%d]..." % n)
                    self.Conectar()
                self.response = self.client("sr-padron", "v2", "persona", str(nro_doc))
            except Exception as e:
                self.client = None
                ex = exception_info()
                self.Traceback = ex.get("tb", "")
                try:
                    self.Excepcion = norm(ex.get("msg", "").replace("\n", ""))
                except:
                    self.Excepcion = "<no disponible>"
                if DEBUG:
                    warnings.warn("Error %s [%d]" % (self.Excepcion, n))
            else:
                break
        else:
            return False
        result = json.loads(self.response)
        if result["success"]:
            data = result["data"]
            # extraigo datos generales del contribuyente:
            self.cuit = data["idPersona"]
            self.tipo_persona = data["tipoPersona"]
            self.tipo_doc = TIPO_CLAVE.get(data["tipoClave"])
            self.dni = data.get("numeroDocumento")
            self.estado = data.get("estadoClave")
            self.denominacion = data.get("nombre")
            # analizo el domicilio
            domicilio = data.get("domicilioFiscal")
            if domicilio:
                self.direccion = domicilio.get("direccion", "")
                self.localidad = domicilio.get("localidad", "")  # no usado en CABA
                self.provincia = PROVINCIAS.get(domicilio.get("idProvincia"), "")
                self.cod_postal = domicilio.get("codPostal")
            else:
                self.direccion = self.localidad = self.provincia = ""
                self.cod_postal = ""
            # retrocompatibilidad:
            self.domicilios = [
                "%s - %s (%s) - %s"
                % (
                    self.direccion,
                    self.localidad,
                    self.cod_postal,
                    self.provincia,
                )
            ]
            # analizo impuestos:
            self.impuestos = data.get("impuestos", [])
            self.actividades = data.get("actividades", [])
            if 32 in self.impuestos:
                self.imp_iva = "EX"
            elif 33 in self.impuestos:
                self.imp_iva = "NI"
            elif 34 in self.impuestos:
                self.imp_iva = "NA"
            else:
                self.imp_iva = "S" if 30 in self.impuestos else "N"
            mt = data.get("categoriasMonotributo", {})
            self.monotributo = "S" if mt else "N"
            self.actividad_monotributo = ""  # TODO: mt[0].get("idCategoria")
            self.integrante_soc = ""
            self.empleador = "S" if 301 in self.impuestos else "N"
            self.cat_iva = ""
            self.data = data
        else:
            error = result["error"]
            self.Excepcion = error["mensaje"]
        return True

    @inicializar_y_capturar_excepciones_simple
    def DescargarConstancia(self, nro_doc, filename="constancia.pdf"):
        "Llama a la API para descargar una constancia de inscripcion (PDF)"
        if not self.client:
            self.Conectar()
        self.response = self.client("sr-padron", "v1", "constancia", str(nro_doc))
        if self.response.startswith("{"):
            result = json.loads(self.response)
            assert not result["success"]
            self.Excepcion = result["error"]["mensaje"]
            return False
        else:
            with open(filename, "wb") as f:
                f.write(self.response)
            return True

    @inicializar_y_capturar_excepciones_simple
    def MostrarPDF(self, archivo, imprimir=False):
        if sys.platform.startswith(("linux2", "java")):
            os.system("evince " "%s" "" % archivo)
        else:
            operation = imprimir and "print" or ""
            os.startfile(archivo, operation)
        return True

    @inicializar_y_capturar_excepciones_simple
    def ObtenerTablaParametros(self, tipo_recurso, sep="||"):
        "Devuelve un array de elementos que tienen id y descripción"
        if not self.client:
            self.Conectar()
        self.response = self.client("parametros", "v1", tipo_recurso)
        result = json.loads(self.response)
        ret = {}
        if result["success"]:
            data = result["data"]
            # armo un diccionario con los datos devueltos:
            key = [k for k in list(data[0].keys()) if k.startswith("id")][0]
            val = [k for k in list(data[0].keys()) if k.startswith("desc")][0]
            for it in data:
                ret[it[key]] = it[val]
            self.data = data
        else:
            error = result["error"]
            self.Excepcion = error["mensaje"]
        if sep:
            return ["%s%%s%s%%s%s" % (sep, sep, sep) % it for it in sorted(ret.items())]
        else:
            return ret


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

def main():
    global DEBUG
    safe_console()

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

        win32com.server.register.UseCommandLine(PadronAFIP)
    else:
        padron = PadronAFIP()
        padron.LanzarExcepciones = True
        import time

        t0 = time.time()
        if "--descargar" in sys.argv:
            padron.Descargar()
        if "--procesar" in sys.argv:
            padron.Procesar(borrar="--borrar" in sys.argv)
        if "--parametros" in sys.argv:
            import codecs, locale, traceback

            if sys.stdout.encoding is None:
                sys.stdout = codecs.getwriter(locale.getpreferredencoding())(
                    sys.stdout, "replace"
                )
                sys.stderr = codecs.getwriter(locale.getpreferredencoding())(
                    sys.stderr, "replace"
                )
            print("=== Impuestos ===")
            print(u"\n".join(padron.ObtenerTablaParametros("impuestos")))
            print("=== Conceptos ===")
            print(u"\n".join(padron.ObtenerTablaParametros("conceptos")))
            print("=== Actividades ===")
            print(u"\n".join(padron.ObtenerTablaParametros("actividades")))
            print("=== Caracterizaciones ===")
            print(u"\n".join(padron.ObtenerTablaParametros("caracterizaciones")))
            print("=== Categorias Monotributo ===")
            print(u"\n".join(padron.ObtenerTablaParametros("categoriasMonotributo")))
            print("=== Categorias Autonomos ===")
            print(u"\n".join(padron.ObtenerTablaParametros("categoriasAutonomo")))

        if "--csv" in sys.argv:
            csv_reader = csv.reader(
                open("entrada.csv", "rU"), dialect="excel", delimiter=","
            )
            csv_writer = csv.writer(
                open("salida.csv", "w"), dialect="excel", delimiter=","
            )
            encabezado = next(csv_reader)
            columnas = [
                "cuit",
                "denominacion",
                "estado",
                "direccion",
                "localidad",
                "provincia",
                "cod_postal",
                "impuestos",
                "actividades",
                "imp_iva",
                "monotributo",
                "actividad_monotributo",
                "empleador",
                "imp_ganancias",
                "integrante_soc",
            ]
            csv_writer.writerow(columnas)

            for fila in csv_reader:
                cuit = (fila[0] if fila else "").replace("-", "")
                if cuit.isdigit():
                    if "--online" in sys.argv:
                        padron.Conectar(trace="--trace" in sys.argv)
                        print("Consultando AFIP online...", cuit, end=" ")
                        ok = padron.Consultar(cuit)
                    else:
                        print("Consultando AFIP local...", cuit, end=" ")
                        ok = padron.Buscar(cuit)
                    print("ok" if ok else "error", padron.Excepcion)
                    # domicilio posiblemente esté en Latin1, normalizar
                    csv_writer.writerow(
                        [norm(getattr(padron, campo, "")) for campo in columnas]
                    )
        elif "--reconex" in sys.argv:
            padron.Conectar(trace="--trace" in sys.argv)
            cuit = 20267565393
            for i in range(10000):
                t0 = time.time()
                ok = padron.Consultar(cuit)
                t1 = time.time()
                print("%2.4f" % (t1 - t0))
        else:
            cuit = len(sys.argv) > 1 and sys.argv[1] or "20267565393"
            # consultar un cuit:
            if "--online" in sys.argv:
                padron.Conectar(trace="--trace" in sys.argv)
                print("Consultando AFIP online...", end=" ")
                ok = padron.Consultar(cuit)
                print("ok" if ok else "error", padron.Excepcion)
                print("Denominacion:", padron.denominacion)
                print("CUIT:", padron.cuit)
                print("Tipo:", padron.tipo_persona, padron.tipo_doc, padron.dni)
                print("Estado:", padron.estado)
                print("Direccion:", padron.direccion)
                print("Localidad:", padron.localidad)
                print("Provincia:", padron.provincia)
                print("Codigo Postal:", padron.cod_postal)
                print("Impuestos:", padron.impuestos)
                print("Actividades:", padron.actividades)
                print("IVA", padron.imp_iva)
                print("MT", padron.monotributo, padron.actividad_monotributo)
                print("Empleador", padron.empleador)
            elif "--constancia" in sys.argv:
                filename = sys.argv[2]
                print("Descargando constancia AFIP online...", cuit, filename)
                ok = padron.DescargarConstancia(cuit, filename)
                print("ok" if ok else "error", padron.Excepcion)
                if "--mostrar" in sys.argv:
                    padron.MostrarPDF(
                        archivo=filename, imprimir="--imprimir" in sys.argv
                    )
            else:
                ok = padron.Buscar(cuit)
                if ok:
                    print("Denominacion:", padron.denominacion)
                    print("IVA:", padron.imp_iva)
                    print("Ganancias:", padron.imp_ganancias)
                    print("Monotributo:", padron.monotributo)
                    print("Integrante Soc.:", padron.integrante_soc)
                    print("Empleador", padron.empleador)
                    print("Actividad Monotributo:", padron.actividad_monotributo)
                    print("Categoria IVA:", padron.cat_iva)
                    padron.ConsultarDomicilios(cuit)
                    for dom in padron.domicilios:
                        print(dom)
                else:
                    print(padron.Excepcion)
                    print(padron.Traceback)
        t1 = time.time()
        if "--trace" in sys.argv:
            print("tiempo", t1 - t0)

if __name__ == "__main__":
    main()