Uecko_ERP_FactuGES_sync/app/db/sync_invoices_verifactu.py

import logging
from uuid import uuid4
from typing import Dict, Any, Tuple, Optional, List, Iterable
from config import load_config
from decimal import Decimal
from utils import validar_nif, estado_factura, crear_factura, TaxCatalog, unscale_to_str


def sync_invoices_verifactu(conn_mysql, last_execution_date):
    config = load_config()

# OPCION A SACAMOS EL RESUMEN DE LA TAXES DE LA CABECERA
# SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value,
# cit.tax_code, sum(cit.taxable_amount_value), sum(cit.taxes_amount_value)
# FROM customer_invoices as ci
# LEFT JOIN customer_invoice_taxes cit on (ci.id = cit.invoice_id)
# WHERE (ci.is_proforma = 0) AND (ci.status= 'draft')
# group by 1,2,3,4,5,6,7,8,9


# OPCION B SACAMOS LOS IVAS DE LOS DETALLES DE LOS ITEM
# SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value,
# ciit.tax_code, sum(ciit.taxable_amount_value), sum(ciit.taxes_amount_value)
# FROM customer_invoices as ci
# LEFT JOIN customer_invoice_items cii on (ci.id = cii.invoice_id)
# LEFT JOIN customer_invoice_item_taxes ciit on (cii.item_id = ciit.item_id)
# WHERE (ci.is_proforma = 0) AND (ci.status= 'issued')
# group by 1,2,3,4,5,6,7,8,9

    # Recorrer todas las facturas emitidas para madarlas o refrescar los campos
    consulta_sql_customer_invoices_issue = (
        f"SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value, ci.total_amount_scale, ci.reference, "
        f"cit.taxable_amount_scale, cit.taxes_amount_scale, cit.tax_code, sum(cit.taxable_amount_value) as taxable_amount_value, sum(cit.taxes_amount_value) as taxes_amount_value, "
        f"vr.id as vrId, vr.uuid, vr.estado "
        f"FROM customer_invoices as ci "
        f"LEFT JOIN customer_invoice_taxes cit on (ci.id = cit.invoice_id) "
        f"LEFT JOIN verifactu_records vr on (ci.id = vr.invoice_id) "
        f"WHERE (ci.is_proforma = 0) AND (ci.status= 'issued') "
        f"AND ((vr.estado is null) OR (vr.estado <> 'Correcto')) "
        f"group by 1,2,3,4,5,6,7,8,9,10,11 "
        f"order by reference"
    )

    # Crear un cursor para ejecutar consultas SQL
    cursor_mysql = None
    try:
        cursor_mysql = conn_mysql.cursor()
    # Ejecutar la consulta de FACTURAS_CLIENTE
        cursor_mysql.execute(consulta_sql_customer_invoices_issue)
        filas = cursor_mysql.fetchall()

        # Obtener los nombres de las columnas
        columnas = [desc[0] for desc in cursor_mysql.description]
        # Convertir las filas en diccionarios con nombres de columnas como claves
        tuplas_seleccionadas = []
        for fila in filas:
            tupla = dict(zip(columnas, fila))
            tuplas_seleccionadas.append(tupla)

        # invoices_to_verifactu = {str(fila[0]) for fila in filas}
        logging.info(
            f"Customer invoices rows to be send Verifactu: {len(tuplas_seleccionadas)}")

        # Verificar si hay filas en el resultado
        if tuplas_seleccionadas:
            enviar_datos(tuplas_seleccionadas, cursor_mysql, config)
            logging.info(f"Ha ido bien enviar_datos")
        else:
            logging.info(f"There are no rows to send")

    except Exception as error:
        if cursor_mysql is not None:
            cursor_mysql.close()
        logging.error(
            f"(ERROR) Failed to fetch from database:{config['UECKO_MYSQL_DATABASE']} - using user:{config['UECKO_MYSQL_USER']}")
        logging.error(error)
        raise error


def enviar_datos(invoices_to_verifactu, cursor_mysql, config):

    # Recorrer todas las facturas para crear json de envio
    try:
        invoice_id = None
        factura = None
        for fila in invoices_to_verifactu:
            # Si los ids de factura anterior y actual no coinciden o empezamos factura nueva, miramos si ya existe una factura si es así la mandamos AEAT
            # y creamos la cabecera de la factura siguiente, si no existe factura solo la creamos
            if invoice_id != str(fila['id']):

                procesar_factura_verifactu(factura, cursor_mysql, config)

                # preparamos nueva factura
                ok, respuesta = preparar_factura(fila)
                if not ok:
                    logging.info(
                        f">>> Factura {fila['reference']} no cumple requisitos para ser mandada a Verifactu:")
                    logging.info(
                        f">>>>>> Faltan campos requeridos: {respuesta}")
                    factura = None
                    continue

                factura = respuesta
                # Validamos que el cif de la factura exista en la AEAT si no es así no se hace el envío
                if not validar_nif(factura.get('nif'), factura.get('nombre'), config):
                    logging.info(
                        f">>> Factura {factura.get('reference')} no cumple requisitos para ser mandada a Verifactu:")
                    logging.info(
                        f">>>>>> El cif de la factura no existe en AEAT: {factura.get('nif')}")
                    continue

            ok, linea = preparar_linea(fila)
            if not ok:
                logging.info(
                    f">>> Factura {factura.get('reference')} no cumple requisitos para ser mandada a Verifactu:")
                logging.info(f">>>>>> Faltan campos requeridos: {linea}")
                factura = None
            else:
                factura["lineas"].append(linea)

        procesar_factura_verifactu(factura, cursor_mysql, config)

    except Exception as e:
        # Escribir el error en el archivo de errores
        logging.error(str(e))
        raise e  # Re-lanzar la excepción para detener el procesamiento


def procesar_factura_verifactu(
    factura: Optional[Dict[str, Any]],
    cursor_mysql,
    config: Dict[str, Any]
) -> bool:

    insert_verifactu_records = ("INSERT INTO verifactu_records (id, invoice_id, estado, uuid, url, qr, created_at, updated_at) "
                                "VALUES (%s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) "
                                )

    update_verifactu_records = ("UPDATE verifactu_records "
                                "set estado = %s,"
                                "operacion = %s, "
                                "updated_at = CURRENT_TIMESTAMP "
                                "WHERE uuid = %s "
                                )

    if factura != None:
        # Creamos registro de factura en verifactu
        if factura.get('uuid') is None:
            # logging.info(f"Send to create Verifactu: {factura}")
            respuesta = crear_factura(factura, config)
            if respuesta.get("status") == 200 and respuesta.get("ok"):
                data = respuesta.get("data")
                cursor_mysql.execute(insert_verifactu_records, (str(uuid4()), factura.get(
                    "id"), data.get("estado"), data.get("uuid"), data.get("url"), data.get("qr")))
                logging.info(
                    f">>> Factura {factura.get("reference")} registrada en Verifactu")
                return True
            else:
                logging.info(
                    f">>> Factura {factura.get("reference")} enviada a Verifactu con error {respuesta}")
                return False
        # Actualizamos registro de factura en verifactu
        else:
            # logging.info(f"Send to update Verifactu: {factura}")
            respuesta = estado_factura(factura.get('uuid'), config)
            if respuesta.get("status") == 200 and respuesta.get("ok"):
                data = respuesta.get("data")
                cursor_mysql.execute(update_verifactu_records, (data.get(
                    'estado'), data.get('operacion'), factura.get('uuid')))
                logging.info(
                    f">>> Factura {factura.get("reference")} actualizado registro de Verifactu")
                return True
            else:
                return False


def preparar_factura(fila: Dict[str, Any]) -> Tuple[bool, Dict[str, Any] | list]:
    """
    Prepara el JSON de factura para Verifactu a partir de 'fila'.
    """
    campos_requeridos = ("series", "invoice_number",
                         "invoice_date", "description", "total_amount_value")
    ok, missing = validar_requeridos(fila, campos_requeridos)
    if not ok:
        return False, missing

    factura = {
        "nif": fila['customer_tin'],
        "nombre": fila['customer_name'],
        "serie": fila['series'],
        "numero": fila['invoice_number'],
        # convertimos la fecha al formato requerido en el api
        "fecha_expedicion": fila['invoice_date'].strftime("%d-%m-%Y"),
        # F1: Factura (Art. 6, 7.2 Y 7.3 del RD 1619/2012)
        "tipo_factura": "F1",
        "descripcion": fila['description'],
        # desescalamos el importe para dar su importe real
        "importe_total": unscale_to_str(str(fila['total_amount_value']), str(fila['total_amount_scale'])),
        "lineas": [],

        # CAMPOS PARA LOGICA NUESTRA
        "id": str(fila['id']),
        "reference": str(fila['reference']),
        "uuid": fila['uuid'],
    }

    return True, factura


def preparar_linea(fila: Dict[str, Any]) -> Tuple[bool, Dict[str, Any] | list]:
    """
    Prepara el JSON de línea para Verifactu a partir de 'fila'.
    """
    campos_requeridos = ("taxable_amount_value",
                         "taxable_amount_scale")
    ok, missing = validar_requeridos(fila, campos_requeridos)
    if not ok:
        return False, missing

    catalog = TaxCatalog.create()

    base_imponible = unscale_to_str(
        str(fila['taxable_amount_value']), str(fila['taxable_amount_scale']))

    # Si el tipo impositivo es exento
    if catalog.is_non_exempt(fila['tax_code']):
        calificacion_operacion = "S1"
        # FALTA COMPROBAR SI IMPUESTO IGIC (03) o IPSI (02)
        impuesto = "01"
        tipo_impositivo = str(catalog.get_percent_reduced(fila['tax_code']))
        cuota_repercutida = unscale_to_str(
            str(fila['taxes_amount_value']), str(fila['taxes_amount_scale']))

        linea = {
            "base_imponible": base_imponible,
            "impuesto": impuesto,
            "tipo_impositivo": tipo_impositivo,
            "cuota_repercutida": cuota_repercutida,
        }

    else:
        # FALTA REVISAR DIFERENCIAS ENTRE E2,E3,E4...
        operacion_exenta = "E1"

        linea = {
            "base_imponible": base_imponible,
            "operacion_exenta": operacion_exenta,
        }

    return True, linea


def validar_requeridos(fila: Dict[str, Any], campos: Iterable[str]) -> Tuple[bool, List[str]]:
    missing = [k for k in campos if fila.get(k) is None]
    return (len(missing) == 0), missing