arreglos refactorizacion

app/db/sync_invoices.py

@@ -6,6 +6,13 @@ from config import load_config
 from decimal import Decimal, ROUND_HALF_UP
 from . import sql_sentences as SQL
 from utils import limpiar_cadena, normalizar_telefono_con_plus, corregir_y_validar_email, normalizar_url_para_insert, map_tax_code, cents, money_round, tax_fraction_from_code
+from striprtf.striprtf import rtf_to_text
+
+# =========================
+# constantes
+# =========================
+# Compañia RODAX
+cte_company_id = '5e4dc5b3-96b9-4968-9490-14bd032fec5f'
 
 
 def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
@@ -25,7 +32,7 @@ def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
         # logging.info(f"Customer invoices rows to be deleted: {len(ids_verifactu_deleted)}")
         # Verificar si hay filas en el resultado
         if ids_verifactu_deleted:
-            eliminar_datos(conn_factuges, ids_verifactu_deleted, config)
+            sync_delete_invoices(conn_factuges, ids_verifactu_deleted, config)
         else:
             logging.info(f"There are customer invoices deleted")
 
@@ -64,13 +71,14 @@ def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
         tuplas_seleccionadas.append(tupla)
     # Verificar si hay filas en el resultado
     if tuplas_seleccionadas:
-        insertar_datos(conn_mysql, tuplas_seleccionadas, conn_factuges, config)
+        sync_invoices_from_FACTUGES(
+            conn_mysql, tuplas_seleccionadas, conn_factuges, config)
     else:
         logging.info(
             "There are no new FACTURAS rows since the last run.")
 
 
-def eliminar_datos(conn_factuges, ids_verifactu_deleted, config):
+def sync_delete_invoices(conn_factuges, ids_verifactu_deleted, config):
     # Eliminamos todos los IDs asociados en FactuGES que han sido eliminados así liberaremos la factura borrador y podermos modificarla de nuevo, para volverla a subir una vez hechos los cambios.
     # VERIFACTU = 0 and ID_VERIFACTU = NULL
     cursor_FactuGES = None
@@ -93,16 +101,13 @@ def eliminar_datos(conn_factuges, ids_verifactu_deleted, config):
             cursor_FactuGES.close()
 
 
-def insertar_datos(conn_mysql, filas, conn_factuges, config):
+def sync_invoices_from_FACTUGES(conn_mysql, filas, conn_factuges, config):
     # Insertaremos cada factura existente en las filas a la nueva estructura de tablas del programa nuevo de facturacion.
     # logging.info(f"FACTURAS_CLIENTE_DETALLE rows to be processed: {len(filas)}")
 
-    # Compañia RODAX
-    cte_company_id = '5e4dc5b3-96b9-4968-9490-14bd032fec5f'
     cursorMySQL = None
     cursor_FactuGES = None
     factuges_id_anterior = None
-    id_customer_invoice = None
     num_fac_procesed = 0
 
     try:
@@ -112,50 +117,19 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
         # Insertar datos en la tabla 'customer_invoices'
         for factura_detalle in filas:
 
-            # Preparamos el tipo de IVA, en FactuGES es único
+            # Preparamos los campos para evitar errores
-            # Map tax code (cabecera)
+            customer_fields = normalize_customer_fields(factura_detalle)
-            tax_code = map_tax_code(str(factura_detalle.get("DES_TIPO_IVA")))
+            header_invoice_fields = normalize_header_invoice_fields(
-
+                factura_detalle)
-            # La cuota de impuestos es el IVA + RE
+            details_invoice_fields = normalize_details_invoice_fields(
-            tax_amount_value = (
+                factura_detalle)
-                (factura_detalle['IMPORTE_IVA'] or 0) + (factura_detalle['IMPORTE_RE'] or 0))*100
+            logging.info(
-
+                f"FACTURAS_CLIENTE DETALLLLLLLLLLESSSSSSS: {details_invoice_fields}")
-            factuges_payment_method_id = str(factura_detalle['ID_FORMA_PAGO'])
-            payment_method_description = str(factura_detalle['DES_FORMA_PAGO'])
-
-            factuges_customer_id = str(factura_detalle['ID_CLIENTE'])
-            customer_tin = limpiar_cadena(str(factura_detalle['NIF_CIF']))
-
-            item_position = int(factura_detalle['POSICION'])
-            item_description = str(factura_detalle['CONCEPTO'])
-            item_quantity_value = None if factura_detalle['CANTIDAD'] is None else (
-                factura_detalle['CANTIDAD'] or 0)*100
-            item_unit_amount_value = None if factura_detalle['IMPORTE_UNIDAD'] is None else (
-                factura_detalle['IMPORTE_UNIDAD'] or 0)*10000
-            Descuento = factura_detalle['DESCUENTO_DET']
-            item_discount_percentage_value = None if Descuento is None else None if Descuento == 0 else (
-                factura_detalle['DESCUENTO_DET'])*100
-            item_total_amount = (factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
-
-            # None if factura_detalle['IMPORTE_TOTAL_DET'] is None else (
-            #    factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
-
-            # campos pendiente de revisar en un futuro
-            # xxxxxxx = str(factura_detalle['ID_EMPRESA'])
-            # xxxxxxx = str(factura_detalle['ID_FORMA_PAGO']) según este id se debe de guardar en la factura los vencimiento asociados a la forma de pago
-            # xxxxxxx = str(factura_detalle['OBSERVACIONES'])
 
             factuges_id = int(factura_detalle['ID_FACTURA'])
-
             if factuges_id_anterior is None or factuges_id_anterior != factuges_id:
-                # Comprobamos si existe el cliente del primer item de la factura
-                # cursorMySQL.execute(SQL.SELECT_CUSTOMER_BY_FACTUGES,
-                #                    (factuges_customer_id, ))
-                # row = cursorMySQL.fetchone()
-                # is_new = (row is None) or (row[0] is None)
 
-                # ---- cliente
+                # Comprobamos si existe el cliente del primer item de la factura
-                customer_fields = normalize_customer_fields(factura_detalle)
                 customer_id = get_or_create_customer(
                     cursorMySQL,
                     cte_company_id,
@@ -169,16 +143,18 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
                     str(factura_detalle["ID_FORMA_PAGO"]),
                     str(factura_detalle["DES_FORMA_PAGO"]),
                 )
+                # campos pendiente de revisar en un futuro
+                # xxxxxxx = str(factura_detalle['ID_FORMA_PAGO']) según este id se debe de guardar en la factura los vencimiento asociados a la forma de pago
 
                 # ---- cabecera factura
-                invoice_id, tax_code = insert_invoice_header(
+                invoice_id = insert_invoice_header(
-                    cursorMySQL, cte_company_id, factura_detalle, customer_id, pm_id, str(
+                    cursorMySQL, cte_company_id, customer_fields, header_invoice_fields, customer_id, pm_id, str(
                         factura_detalle["DES_FORMA_PAGO"])
                 )
 
                 # ---- impuestos cabecera
                 insert_header_taxes_if_any(
-                    cursorMySQL, invoice_id, factura_detalle, tax_code)
+                    cursorMySQL, invoice_id, factura_detalle['IVA'], factura_detalle['RECARGO_EQUIVALENCIA'], header_invoice_fields)
 
                 # Guardamos en Factuges el id de la customer_invoice
                 logging.info(
@@ -190,7 +166,7 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
             # Insertamos detalles y taxes correspondientes siempre
             # Siempre insertamos la línea
             insert_item_and_taxes(cursorMySQL, invoice_id,
-                                  factura_detalle, tax_code)
+                                  details_invoice_fields)
 
             # Asignamos el id factura anterior para no volver a inserta cabecera
             factuges_id_anterior = factuges_id
@@ -211,9 +187,16 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
             cursor_FactuGES.close()
 
 
+def rtf_a_texto_plano(rtf: str) -> str:
+    """
+    Convierte RTF a texto plano usando striprtf.
+    """
+    return rtf_to_text(rtf).strip()
+
+
 def normalize_customer_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
     """
-    Normaliza campos de cliente del registro de origen 'fd' (factura_detalle).
+    Normaliza campos del registro de origen 'fd' (factura).
     """
     # >>> aquí usa tus helpers reales:
     def clean_phone(v): return normalizar_telefono_con_plus(v)
@@ -237,11 +220,80 @@ def normalize_customer_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
         "mobile_secondary": clean_phone(fd.get("MOVIL_2")),
         "email_primary": email1 if email1_ok else None,
         "email_secondary": email2 if email2_ok else None,
-        "website": clean_web(str(fd.get("PAGINA_WEB"))),
+        "website": clean_web(str(fd.get("PAGINA_WEB")))
+    }
+
+
+def normalize_header_invoice_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Normaliza campos del registro de origen 'fd' (factura).
+
+    # campos pendiente de revisar en un futuro
+    # xxxxxxx = str(factura_detalle['ID_EMPRESA'])
+    # xxxxxxx = str(factura_detalle['OBSERVACIONES'])
+    """
+    return {
+        "factuges_id": int(fd['ID_FACTURA']),
+        "reference": str(fd['REFERENCIA']),
+        "invoice_date": str(fd['FECHA_FACTURA']),
+        "operation_date": str(fd['FECHA_FACTURA']),
+        "description": textwrap.shorten(
+            f"{str(fd['REFERENCIA'])} - {str(fd.get('NOMBRE')) or ''}", width=50, placeholder="…"),
+
+        # siempre tendrán 2 decimales
+        'subtotal_amount_value': cents(fd.get('IMPORTE_NETO')),
+        'discount_amount_value': cents(fd.get('IMPORTE_DESCUENTO')),
+        'discount_percentage_val': int((Decimal(str(fd.get('DESCUENTO') or 0)) * 100).to_integral_value())
+        if fd.get('DESCUENTO') is not None else 0,
+        'taxable_amount_value': cents(fd.get('BASE_IMPONIBLE')),
+        'tax_code': map_tax_code(str(fd.get("DES_TIPO_IVA"))),
+        'taxes_amount_value': cents((fd.get('IMPORTE_IVA') or 0) + (fd.get('IMPORTE_RE') or 0)),
+        'total_amount_value': cents(fd.get('IMPORTE_TOTAL')),
+
+        'base': cents(fd.get('BASE_IMPONIBLE')),
+        'iva': cents(fd.get('IMPORTE_IVA')),
+        're': cents(fd.get('IMPORTE_RE'))
+
+    }
+
+
+def normalize_details_invoice_fields(fd: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Normaliza campos del registro de origen 'fd' (factura).
+    """
+    tax_code = map_tax_code(str(fd.get("DES_TIPO_IVA")))
+    disc_pct = fd.get('DESCUENTO_DET')
+
+    # Calcular cuota de IVA de la línea
+    frac = tax_fraction_from_code(tax_code)  # 0.21, 0.10, 0…
+    # Ttotal_det ya incluye descuento
+    line_base = Decimal(str(fd.get('IMPORTE_TOTAL_DET') or 0))
+    tax_amount = int(
+        (money_round(line_base * frac, 2)*100).to_integral_value())
+
+    logging.info(
+        f"FACTURAS_CLIENTE IVAAAAAAAAAAAAAAA {str(frac)} DETALLE: {str((money_round(line_base * frac, 2)*100))}")
+    # Se calcula en el objeto de negocio de nuevo, comprobar si coincide
+    # item_discount_amount = (
+    #    (factura_detalle['IMPORTE_UNIDAD'] or 0)*((factura_detalle['DESCUENTO'] or 0)/100))*100
+
+    return {
+        'tax_code': tax_code,
+        'position': int(fd['POSICION']),
+        'description': rtf_a_texto_plano(str(fd['CONCEPTO'])),
+        'quantity_value': None if fd['CANTIDAD'] is None else int(Decimal(str(fd['CANTIDAD'] or 0)) * 100),
+        'unit_value': None if fd['IMPORTE_UNIDAD'] is None else int(Decimal(str(fd['IMPORTE_UNIDAD'] or 0)) * 10000),
+        'disc_pct': disc_pct,
+        'discount_percentage_value': None if disc_pct in (None, 0) else int(Decimal(str(disc_pct)) * 100),
+        'total_value': cents(fd.get('IMPORTE_TOTAL_DET')),
+        'tax_amount': tax_amount
     }
 
 
 def get_or_create_customer(cur, company_id: str, factuges_customer_id: str, fields: Dict[str, Any]) -> str:
+    """
+    Comprobamos si existe el cliente del primer item de la factura y si no lo creamos
+    """
     cur.execute(SQL.SELECT_CUSTOMER_BY_FACTUGES, (factuges_customer_id,))
     row = cur.fetchone()
     if not row or not row[0]:
@@ -273,6 +325,9 @@ def get_or_create_customer(cur, company_id: str, factuges_customer_id: str, fiel
 
 
 def get_or_create_payment_method(cur, factuges_payment_id: str, description: str) -> str:
+    """
+    En el caso de que la forma de pago no exista la creamos
+    """
     cur.execute(SQL.SELECT_PAYMENT_METHOD_BY_FACTUGES, (factuges_payment_id,))
     row = cur.fetchone()
     if not row or not row[0]:
@@ -287,132 +342,65 @@ def get_or_create_payment_method(cur, factuges_payment_id: str, description: str
     return pm_id
 
 
-def insert_invoice_header(cur, company_id: str, fd: Dict[str, Any], customer_id: str,
+def insert_invoice_header(cur, company_id: str, cf: Dict[str, Any], hif: Dict[str, Any], customer_id: str,
-                          payment_method_id: str, payment_method_description: str) -> Tuple[str, str]:
+                          payment_method_id: str, payment_method_description: str) -> str:
     """
-    Inserta cabecera y devuelve (invoice_id, tax_code_normalized)
+    Inserta cabecera y devuelve invoice_id
     """
     invoice_id = str(uuid7())
 
-    reference = str(fd['REFERENCIA'])
-    invoice_date = str(fd['FECHA_FACTURA'])
-    operation_date = str(fd['FECHA_FACTURA'])
-    description = textwrap.shorten(
-        f"{reference or ''} - {str(fd.get('NOMBRE')) or ''}", width=50, placeholder="…")
-
-    # siempre tendrán 2 decimales
-    subtotal_amount_value = cents(fd.get('IMPORTE_NETO'))
-    discount_amount_value = cents(fd.get('IMPORTE_DESCUENTO'))
-    discount_percentage_val = int((Decimal(str(fd.get('DESCUENTO') or 0)) * 100).to_integral_value()) \
-        if fd.get('DESCUENTO') is not None else 0
-    taxable_amount_value = cents(fd.get('BASE_IMPONIBLE'))
-    taxes_amount_value = cents(
-        (fd.get('IMPORTE_IVA') or 0) + (fd.get('IMPORTE_RE') or 0))
-    total_amount_value = cents(fd.get('IMPORTE_TOTAL'))
-
-    # Mapea tax_code cabecera
-    tax_code = map_tax_code(str(fd.get("DES_TIPO_IVA")))
-
     logging.info("Inserting invoice %s %s %s",
-                 invoice_id, reference, invoice_date)
+                 invoice_id, hif.get('reference'), hif.get('invoice_date'))
     cur.execute(
         SQL.INSERT_INVOICE,
         (
-            invoice_id, company_id, 'draft', 'F25/', reference, invoice_date, operation_date, description,
+            invoice_id, company_id, 'draft', 'F25/', hif.get('reference'), hif.get(
-            subtotal_amount_value, discount_amount_value, discount_percentage_val, taxable_amount_value,
+                'invoice_date'), hif.get('operation_date'), hif.get('description'),
-            taxes_amount_value, total_amount_value,
+            hif.get('subtotal_amount_value'), hif.get('discount_amount_value'), hif.get(
-            customer_id, fd.get('NIF_CIF'), fd.get(
+                'discount_percentage_val'), hif.get('taxable_amount_value'),
-                'NOMBRE'), fd.get('CALLE'), fd.get('POBLACION'),
+            hif.get('taxes_amount_value'), hif.get('total_amount_value'),
-            fd.get('PROVINCIA'), fd.get('CODIGO_POSTAL'), 'es',
+            customer_id, cf.get("tin"), cf.get(
-            payment_method_id, payment_method_description, fd.get(
+                'name'), cf.get('street'), cf.get('city'),
-                'ID_FACTURA'), company_id
+            cf.get('province'), cf.get('postal_code'), 'es',
+            payment_method_id, payment_method_description, hif.get(
+                'factuges_id'), company_id
         ),
     )
-    return invoice_id, tax_code
+    return invoice_id
 
 
-def insert_header_taxes_if_any(cur, invoice_id: str, fd: Dict[str, Any], tax_code: str) -> None:
+def insert_header_taxes_if_any(cur, invoice_id: str, IVA: str, RECARGO: str, hif: Dict[str, Any]) -> None:
-    base = cents(fd.get('BASE_IMPONIBLE'))
+    """
-    iva = cents(fd.get('IMPORTE_IVA'))
+    Inserta impuestos de cabecera
-    re = cents(fd.get('IMPORTE_RE'))
+    """
-
     # IVA (>= 0 acepta 0% también, si no quieres registrar 0, cambia condición a > 0)
-    if (fd.get('IVA') or 0) >= 0:
+    if (IVA or 0) >= 0:
         cur.execute(SQL.INSERT_INVOICE_TAX,
-                    (str(uuid7()), invoice_id, tax_code, base, iva))
+                    (str(uuid7()), invoice_id, hif.get('tax_code'), hif.get('base'), hif.get('iva')))
 
     # Recargo equivalencia
-    if (fd.get('RECARGO_EQUIVALENCIA') or 0) > 0:
+    if (RECARGO or 0) > 0:
         cur.execute(SQL.INSERT_INVOICE_TAX,
-                    (str(uuid7()), invoice_id, 'rec_5_2', base, re))
+                    (str(uuid7()), invoice_id, 'rec_5_2', hif.get('base'), hif.get('re')))
 
 
-def insert_item_and_taxes(cur, invoice_id: str, fd: Dict[str, Any], tax_code: str) -> None:
+def insert_item_and_taxes(cur, invoice_id: str, fields: Dict[str, Any]) -> None:
     """
     Inserta línea y sus impuestos derivados.
     """
     item_id = str(uuid7())
-    position = int(fd['POSICION'])
-    description = str(fd['CONCEPTO'])
-    quantity_value = None if fd['CANTIDAD'] is None else int(
-        Decimal(str(fd['CANTIDAD'] or 0)) * 100)
-    unit_value = None if fd['IMPORTE_UNIDAD'] is None else int(
-        Decimal(str(fd['IMPORTE_UNIDAD'] or 0)) * 10000)
-    disc_pct = fd.get('DESCUENTO_DET')
-    discount_percentage_value = None if disc_pct in (
-        None, 0) else int(Decimal(str(disc_pct)) * 100)
-    total_value = cents(fd.get('IMPORTE_TOTAL_DET'))
 
     logging.info("Inserting item %s pos=%s qty=%s",
-                 item_id, position, quantity_value)
+                 item_id, fields.get('position'), fields.get('quantity_value'))
     cur.execute(
         SQL.INSERT_INVOICE_ITEM,
-        (item_id, invoice_id, position, description, quantity_value,
+        (item_id, invoice_id, fields.get('position'), fields.get('description'), fields.get('quantity_value'),
-         unit_value, discount_percentage_value, None, total_value)
+         fields.get('unit_value'), fields.get('discount_percentage_value'), None, fields.get('total_value'))
     )
-    # Se calcula en el objeto de negocio de nuevo, comprobar si coincide
-    # item_discount_amount = (
-    #    (factura_detalle['IMPORTE_UNIDAD'] or 0)*((factura_detalle['DESCUENTO'] or 0)/100))*100
 
-    # Calcular cuota de IVA de la línea
-    frac = tax_fraction_from_code(tax_code)  # 0.21, 0.10, 0…
-    # si tu total_det ya incluye descuento
-    line_base = Decimal(str(fd.get('IMPORTE_TOTAL_DET') or 0))
-    tax_amount = int((money_round(line_base * frac, 2)
-                     * 100).to_integral_value())
     logging.info("Inserting item tax %s code=%s base=%s tax=%s",
-                 item_id, tax_code, total_value, tax_amount)
+                 item_id, fields.get('tax_code'), fields.get('total_value'), fields.get('tax_amount'))
-
     cur.execute(
-        SQL.INSERT_INVOICE_ITEM_TAX,
+        SQL.INSERT_INVOICE_ITEM_TAX, (str(uuid7()), item_id, fields.get('tax_code'),
-        (str(uuid7()), item_id, tax_code, total_value, tax_amount)
+                                      fields.get('total_value'), fields.get('tax_amount'))
     )
-
-
-"""
-logging.info(
-                f"Inserting customer_invoice_item_taxes {item_id} {item_position} {tax_code} {item_total_amount} {tax_amount_value}")
-            cursorMySQL.execute(SQL.INSERT_INVOICE_ITEM_TAX, (str(uuid7()), item_id, tax_code,
-                                item_total_amount, tax_amount_value))
-
-
-            if tax_code == 'iva_21':
-                tax_amount_value = (
-                    (Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.21')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
-            elif tax_code == 'iva_18':
-                tax_amount_value = (
-                    (Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.18')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
-            elif tax_code == 'iva_16':
-                tax_amount_value = (
-                    (Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.16')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
-            elif tax_code == 'iva_10':
-                tax_amount_value = (
-                    (Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.10')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
-            elif tax_code == 'iva_exenta':
-                tax_amount_value = (
-                    (Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
-            else:
-                tax_amount_value = (
-                    factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
-
-"""

app/utils/tax_catalog_helper.py

@@ -1,3 +1,4 @@
+import logging
 import json
 from decimal import Decimal
 from typing import Any, Dict, Iterable, Optional, Tuple
@@ -43,6 +44,8 @@ def tax_fraction_from_code(tax_code: str) -> Decimal:
     Devuelve la fracción (0.21, 0.18, ...) según el tax_code normalizado.
     Exenta -> 0.
     """
+    logging.info(f"FACTURAS_CLIENTE IVAAAAAAAAAAAAAAA {tax_code}")
+
     # Si ya tienes TaxCatalog, puedes delegarlo allí; lo dejo inline para simplicidad.
     mapping = {
         'iva_21': Decimal('0.21'),
@@ -56,7 +59,7 @@ def tax_fraction_from_code(tax_code: str) -> Decimal:
         'iva_0':  Decimal('0.00'),
         'iva_exenta': Decimal('0.00'),
     }
-    return mapping.get(tax_code or '', Decimal('0.00'))
+    return mapping.get(tax_code or '', Decimal('0.13'))
 
 
 def reduce_scale_pair(value: int, scale: int, *, min_scale: int = 0) -> Tuple[int, int]: