import logging import textwrap from typing import Dict, Any, Optional, Tuple from uuid6 import uuid7 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 def sync_invoices(conn_factuges, conn_mysql, last_execution_date): config = load_config() # LIMPIAMOS LAS FACTURAS DE FACTUGES QUE HAYAN SIDO ELIMINADAS DEL PROGRAMA NUEVO DE FACTURACION, PARA QUE PUEDAN SER MODIFICADAS # Crear un cursor para ejecutar consultas SQL cursor_mysql = None try: cursor_mysql = conn_mysql.cursor() cursor_mysql.execute(SQL.SELECT_INVOICES_DELETED) filas = cursor_mysql.fetchall() cursor_mysql.close() # Crear un conjunto con los IDs [0] de los customer_inovices que debo liberar en FactuGES, porque han sido eliminadas en programa de facturación nuevo ids_verifactu_deleted = {str(fila[0]) for fila in filas} # 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) else: logging.info(f"There are customer invoices deleted") except Exception as e: 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(e) raise e # BUSCAMOS FACTURAS ENVIADAS A VERIFACTU EN FACTUGES, PARA SUBIRLAS AL NUEVO PROGRAMA DE FACTURACIÓN # Crear un cursor para ejecutar consultas SQL cursor_FactuGES = None try: cursor_FactuGES = conn_factuges.cursor() # Ejecutar la consulta de FACTURAS_CLIENTE cursor_FactuGES.execute(SQL.SELECT_FACTUGES_FACTURAS_CLIENTE) filas = cursor_FactuGES.fetchall() except Exception as e: if cursor_FactuGES is not None: cursor_FactuGES.close() logging.error(f"(ERROR) Failed to fetch from database:{ config['FACTUGES_DATABASE']} - using user:{config['FACTUGES_USER']}") logging.error(e) raise e # Obtener los nombres de las columnas columnas = [desc[0] for desc in cursor_FactuGES.description] cursor_FactuGES.close() # 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) # Verificar si hay filas en el resultado if tuplas_seleccionadas: insertar_datos(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): # 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 try: cursor_FactuGES = conn_factuges.cursor() if ids_verifactu_deleted: logging.info(f"Liberate factures: {ids_verifactu_deleted}") cursor_FactuGES.executemany(SQL.LIMPIAR_FACTUGES_LINK, [( id_verifactu,) for id_verifactu in ids_verifactu_deleted]) else: logging.info("No articles to delete.") 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 finally: # Cerrar la conexión if cursor_FactuGES is not None: cursor_FactuGES.close() def insertar_datos(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: cursorMySQL = conn_mysql.cursor() cursor_FactuGES = conn_factuges.cursor() # 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']) 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) customer_id = get_or_create_customer( cursorMySQL, cte_company_id, str(factura_detalle["ID_CLIENTE"]), customer_fields, ) # ---- forma de pago pm_id = get_or_create_payment_method( cursorMySQL, str(factura_detalle["ID_FORMA_PAGO"]), str(factura_detalle["DES_FORMA_PAGO"]), ) # ---- cabecera factura invoice_id, tax_code = insert_invoice_header( cursorMySQL, cte_company_id, factura_detalle, customer_id, pm_id, str( factura_detalle["DES_FORMA_PAGO"]) ) # ---- impuestos cabecera insert_header_taxes_if_any( cursorMySQL, invoice_id, factura_detalle, tax_code) # Guardamos en Factuges el id de la customer_invoice logging.info( f"Updating FACTURAS_CLIENTE {invoice_id} {factuges_id}") cursor_FactuGES.execute( SQL.UPDATE_FACTUGES_LINK, (invoice_id, factuges_id)) num_fac_procesed += 1 # Insertamos detalles y taxes correspondientes siempre # Siempre insertamos la línea insert_item_and_taxes(cursorMySQL, invoice_id, factura_detalle, tax_code) # Asignamos el id factura anterior para no volver a inserta cabecera factuges_id_anterior = factuges_id logging.info( f"FACTURAS_CLIENTE rows to be processed: {str(num_fac_procesed)}") 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 finally: # Cerrar la conexión if cursorMySQL is not None: cursorMySQL.close() if cursor_FactuGES is not None: cursor_FactuGES.close() def normalize_customer_fields(fd: Dict[str, Any]) -> Dict[str, Any]: """ Normaliza campos de cliente del registro de origen 'fd' (factura_detalle). """ # >>> aquí usa tus helpers reales: def clean_phone(v): return normalizar_telefono_con_plus(v) def clean_web(v): return normalizar_url_para_insert(v) def clean_tin(v): return limpiar_cadena(v) email1_ok, email1 = corregir_y_validar_email(fd.get("EMAIL_1")) email2_ok, email2 = corregir_y_validar_email(fd.get("EMAIL_2")) return { "tin": clean_tin(str(fd.get("NIF_CIF"))), "name": str(fd.get("NOMBRE")), "street": str(fd.get("CALLE")), "city": str(fd.get("POBLACION")), "province": str(fd.get("PROVINCIA")), "postal_code": str(fd.get("CODIGO_POSTAL")), "country": "es", "phone_primary": clean_phone(fd.get("TELEFONO_1")), "phone_secondary": clean_phone(fd.get("TELEFONO_2")), "mobile_primary": clean_phone(fd.get("MOVIL_1")), "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"))), } def get_or_create_customer(cur, company_id: str, factuges_customer_id: str, fields: Dict[str, Any]) -> str: cur.execute(SQL.SELECT_CUSTOMER_BY_FACTUGES, (factuges_customer_id,)) row = cur.fetchone() if not row or not row[0]: customer_id = str(uuid7()) logging.info("Inserting customer %s %s %s", factuges_customer_id, fields["tin"], fields["name"]) cur.execute( SQL.INSERT_CUSTOMER, ( customer_id, fields["name"], fields["tin"], fields["street"], fields["city"], fields["province"], fields["postal_code"], fields["country"], fields["phone_primary"], fields["phone_secondary"], fields["mobile_primary"], fields["mobile_secondary"], fields["email_primary"], fields["email_secondary"], fields["website"], factuges_customer_id, company_id, ), ) return customer_id customer_id = str(row[0]) logging.info("Updating customer %s %s", factuges_customer_id, customer_id) cur.execute( SQL.UPDATE_CUSTOMER, ( fields["name"], fields["tin"], fields["street"], fields["city"], fields["province"], fields["postal_code"], fields["country"], fields["phone_primary"], fields["phone_secondary"], fields["mobile_primary"], fields["mobile_secondary"], fields["email_primary"], fields["email_secondary"], fields["website"], customer_id, ), ) return customer_id def get_or_create_payment_method(cur, factuges_payment_id: str, description: str) -> str: cur.execute(SQL.SELECT_PAYMENT_METHOD_BY_FACTUGES, (factuges_payment_id,)) row = cur.fetchone() if not row or not row[0]: pm_id = str(uuid7()) logging.info("Inserting payment method %s %s %s", factuges_payment_id, pm_id, description) cur.execute(SQL.INSERT_PAYMENT_METHOD, (pm_id, description, factuges_payment_id)) return pm_id pm_id = str(row[0]) logging.info("Payment method exists %s -> %s", factuges_payment_id, pm_id) 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]: """ Inserta cabecera y devuelve (invoice_id, tax_code_normalized) """ 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) 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 ), ) return invoice_id, tax_code 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')) # IVA (>= 0 acepta 0% también, si no quieres registrar 0, cambia condición a > 0) if (fd.get('IVA') or 0) >= 0: cur.execute(SQL.INSERT_INVOICE_TAX, (str(uuid7()), invoice_id, tax_code, base, iva)) # Recargo equivalencia if (fd.get('RECARGO_EQUIVALENCIA') or 0) > 0: cur.execute(SQL.INSERT_INVOICE_TAX, (str(uuid7()), invoice_id, 'rec_5_2', base, re)) def insert_item_and_taxes(cur, invoice_id: str, fd: Dict[str, Any], tax_code: str) -> 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) cur.execute( SQL.INSERT_INVOICE_ITEM, (item_id, invoice_id, position, description, quantity_value, unit_value, discount_percentage_value, None, 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) cur.execute( SQL.INSERT_INVOICE_ITEM_TAX, (str(uuid7()), item_id, tax_code, total_value, 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 """