From 6da0b3bd81652f486908d3bc78be233d9b89361a Mon Sep 17 00:00:00 2001 From: david Date: Fri, 7 Nov 2025 19:22:36 +0100 Subject: [PATCH] arreglos refactorizacion --- app/db/sync_invoices.py | 288 +++++++++++++++----------------- app/utils/tax_catalog_helper.py | 5 +- 2 files changed, 142 insertions(+), 151 deletions(-) diff --git a/app/db/sync_invoices.py b/app/db/sync_invoices.py index a7cffe6..f73a2fd 100644 --- a/app/db/sync_invoices.py +++ b/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 - # Map tax code (cabecera) - tax_code = map_tax_code(str(factura_detalle.get("DES_TIPO_IVA"))) - - # La cuota de impuestos es el IVA + RE - tax_amount_value = ( - (factura_detalle['IMPORTE_IVA'] or 0) + (factura_detalle['IMPORTE_RE'] or 0))*100 - - 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']) + # Preparamos los campos para evitar errores + customer_fields = normalize_customer_fields(factura_detalle) + header_invoice_fields = normalize_header_invoice_fields( + factura_detalle) + details_invoice_fields = normalize_details_invoice_fields( + factura_detalle) + logging.info( + f"FACTURAS_CLIENTE DETALLLLLLLLLLESSSSSSS: {details_invoice_fields}") 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 - customer_fields = normalize_customer_fields(factura_detalle) + # Comprobamos si existe el cliente del primer item de la factura 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( - cursorMySQL, cte_company_id, factura_detalle, customer_id, pm_id, str( + invoice_id = insert_invoice_header( + 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, - payment_method_id: str, payment_method_description: str) -> Tuple[str, 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) -> 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, - subtotal_amount_value, discount_amount_value, discount_percentage_val, taxable_amount_value, - taxes_amount_value, total_amount_value, - customer_id, fd.get('NIF_CIF'), fd.get( - 'NOMBRE'), fd.get('CALLE'), fd.get('POBLACION'), - fd.get('PROVINCIA'), fd.get('CODIGO_POSTAL'), 'es', - payment_method_id, payment_method_description, fd.get( - 'ID_FACTURA'), company_id + invoice_id, company_id, 'draft', 'F25/', hif.get('reference'), hif.get( + 'invoice_date'), hif.get('operation_date'), hif.get('description'), + hif.get('subtotal_amount_value'), hif.get('discount_amount_value'), hif.get( + 'discount_percentage_val'), hif.get('taxable_amount_value'), + hif.get('taxes_amount_value'), hif.get('total_amount_value'), + customer_id, cf.get("tin"), cf.get( + 'name'), cf.get('street'), cf.get('city'), + 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: - base = cents(fd.get('BASE_IMPONIBLE')) - iva = cents(fd.get('IMPORTE_IVA')) - re = cents(fd.get('IMPORTE_RE')) - +def insert_header_taxes_if_any(cur, invoice_id: str, IVA: str, RECARGO: str, hif: Dict[str, Any]) -> None: + """ + Inserta impuestos de cabecera + """ # 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, - unit_value, discount_percentage_value, None, total_value) + (item_id, invoice_id, fields.get('position'), fields.get('description'), fields.get('quantity_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, - (str(uuid7()), item_id, tax_code, total_value, tax_amount) + SQL.INSERT_INVOICE_ITEM_TAX, (str(uuid7()), item_id, fields.get('tax_code'), + 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 - -""" diff --git a/app/utils/tax_catalog_helper.py b/app/utils/tax_catalog_helper.py index 21ee5cc..80ef093 100644 --- a/app/utils/tax_catalog_helper.py +++ b/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]: