diff --git a/.env.development b/.env.development index 4885917..654a293 100644 --- a/.env.development +++ b/.env.development @@ -36,3 +36,7 @@ BREVO_EMAIL_TEMPLATE = 1 MAIL_FROM = 'no-reply@presupuestos.uecko.com' MAIL_TO = 'soporte@rodax-software.com' +VERIFACTU_BASE_URL = https://api.verifacti.com/ +VERIFACTU_API_KEY = vf_test_kY9FoI86dH+g1a5hmEnb/0YcLTlMFlu+tpp9iMZp020= +VERIFACTU_NIFS_API_KEY = vfn_osYpNdqSzAdTAHpazXG2anz4F3o0gfbSb5FFrCBZcno= + diff --git a/app/config/settings.py b/app/config/settings.py index cf70f3e..fc3b950 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -43,4 +43,8 @@ def load_config(): 'BREVO_EMAIL_TEMPLATE': os.getenv("BREVO_EMAIL_TEMPLATE"), 'MAIL_FROM': os.getenv('MAIL_FROM'), 'MAIL_TO': os.getenv('MAIL_TO'), + + 'VERIFACTU_BASE_URL': os.getenv('VERIFACTU_BASE_URL'), + 'VERIFACTU_API_KEY': os.getenv('VERIFACTU_API_KEY'), + 'VERIFACTU_NIFS_API_KEY': os.getenv('VERIFACTU_NIFS_API_KEY'), } diff --git a/app/db/sync_invoices.py b/app/db/sync_invoices.py index dcddf3e..2287e67 100644 --- a/app/db/sync_invoices.py +++ b/app/db/sync_invoices.py @@ -1,7 +1,9 @@ import logging +import textwrap from uuid import uuid4 from config import load_config from decimal import Decimal, ROUND_HALF_UP +from utils import limpiar_cadena def sync_invoices(conn_factuges, conn_mysql, last_execution_date): @@ -173,13 +175,13 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): ) insert_customer_invoices_query = ( - "INSERT INTO customer_invoices (id, company_id, status, series, reference, invoice_date, operation_date, " + "INSERT INTO customer_invoices (id, company_id, status, series, reference, invoice_date, operation_date, description, " "subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, taxes_amount_value, total_amount_value, " "customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, " "payment_method_id, payment_method_description, " "subtotal_amount_scale, discount_amount_scale, discount_percentage_scale, taxable_amount_scale, taxes_amount_scale, total_amount_scale, " "language_code, currency_code, created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 2, 2, 2, 2, 2, 'es', 'EUR', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" + "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 2, 2, 2, 2, 2, 'es', 'EUR', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" ) insert_customer_invoices_taxes_query = ( @@ -229,7 +231,7 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): factuges_id = int(factura_detalle['ID_FACTURA']) invoice_status = str('draft') invoice_series = str('A') - invoice_number = str(factura_detalle['REFERENCIA']) + reference = str(factura_detalle['REFERENCIA']) invoice_date = str(factura_detalle['FECHA_FACTURA']) operation_date = str(factura_detalle['FECHA_FACTURA']) # siempre tendrán 2 decimales @@ -249,6 +251,8 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): tax_code = 'iva_16' elif tax_code == 'IVA10': tax_code = 'iva_10' + elif tax_code == 'EXENTO': + tax_code = 'iva_exenta' else: tax_code = '' # La cuota de impuestos es el IVA + RE @@ -263,7 +267,7 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): customer_id = str(uuid4()) factuges_customer_id = str(factura_detalle['ID_CLIENTE']) - customer_tin = str(factura_detalle['NIF_CIF']) + customer_tin = limpiar_cadena(str(factura_detalle['NIF_CIF'])) customer_name = str(factura_detalle['NOMBRE']) customer_street = str(factura_detalle['CALLE']) customer_city = str(factura_detalle['POBLACION']) @@ -277,6 +281,8 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): customer_email_secondary = factura_detalle['EMAIL_2'] customer_webside = str(factura_detalle['PAGINA_WEB']) customer_country = 'es' + description = textwrap.shorten( + f"{reference or ''} - {customer_name or ''}", width=50, placeholder="…") item_position = int(factura_detalle['POSICION']) item_description = str(factura_detalle['CONCEPTO']) @@ -343,14 +349,14 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): # Generar un ID único para la tabla customer_invoices id_customer_invoice = str(uuid4()) logging.info( - f"Inserting customer_invoice {id_customer_invoice} {invoice_number} {invoice_date}") - cursorMySQL.execute(insert_customer_invoices_query, (id_customer_invoice, cte_company_id, invoice_status, invoice_series, invoice_number, invoice_date, operation_date, + f"Inserting customer_invoice {id_customer_invoice} {reference} {invoice_date}") + cursorMySQL.execute(insert_customer_invoices_query, (id_customer_invoice, cte_company_id, invoice_status, invoice_series, reference, invoice_date, operation_date, description, subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, tax_amount_value, total_amount_value, customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, payment_method_id, payment_method_description)) # Insertamos el IVA y RE si viene - if (factura_detalle['IVA'] > 0): + if (factura_detalle['IVA'] >= 0): taxable_amount_value = ( factura_detalle['BASE_IMPONIBLE'])*100 tax_amount_value = (factura_detalle['IMPORTE_IVA'])*100 @@ -392,6 +398,9 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config): 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/db/sync_invoices_verifactu.py b/app/db/sync_invoices_verifactu.py index 5cef70a..17f188c 100644 --- a/app/db/sync_invoices_verifactu.py +++ b/app/db/sync_invoices_verifactu.py @@ -1,7 +1,9 @@ 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): @@ -22,18 +24,21 @@ def sync_invoices_verifactu(conn_mysql, last_execution_date): # 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= 'draft') +# 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, " - f"cit.tax_code, sum(cit.taxable_amount_value), sum(cit.taxes_amount_value) " + 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"WHERE " - f"(ci.is_proforma = 0) AND (ci.status= 'draft')" - f"group by 1,2,3,4,5,6,7,8,9" + 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 @@ -46,60 +51,74 @@ def sync_invoices_verifactu(conn_mysql, last_execution_date): # Obtener los nombres de las columnas columnas = [desc[0] for desc in cursor_mysql.description] - cursor_mysql.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) - # Crear un conjunto con los IDs [0] de los customer_inovices que debo liberar en FactuGES # invoices_to_verifactu = {str(fila[0]) for fila in filas} - logging.info(f"Customer invoices rows to be send Verifactu: { - len(tuplas_seleccionadas)}") + 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, config) + 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 e: + 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(e) - raise e + 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, config): +def enviar_datos(invoices_to_verifactu, cursor_mysql, config): + # Recorrer todas las facturas para crear json de envio try: - logging.info(f"Send to Verifactu") + invoice_id = None + factura = None for fila in invoices_to_verifactu: - factura = { - # REQUERIDOS - "id": str(fila['id']), - "serie": fila['series'], - "numero": fila['invoice_number'], - "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'], - "nif": fila['customer_tin'] - # "tin": fila[5], - # "name": fila[6] - # }, - # "totals": { - # "total_amount_value": float(fila[7]) if isinstance(fila[7], Decimal) else fila[7], - # "taxable_amount_value": float(fila[9]) if fila[9] is not None else 0, - # "taxes_amount_value": float(fila[10]) if fila[10] is not None else 0, - # }, - # "tax_code": fila[8] - } - logging.info(f"Send to Verifactu: {factura}") + # 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 @@ -107,261 +126,130 @@ def enviar_datos(invoices_to_verifactu, config): raise e # Re-lanzar la excepción para detener el procesamiento -def insertar_datos(conn_mysql, filas, conn_factuges, config): - cte_company_id = '5e4dc5b3-96b9-4968-9490-14bd032fec5f' - insert_customer_query = ( - "INSERT INTO customers (id, name, tin, street, city, province, postal_code, country, phone_primary, phone_secondary, mobile_primary, mobile_secondary, " - "email_primary, email_secondary, website, factuges_id, company_id, is_company, language_code, currency_code, status, created_at, updated_at ) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 1, 'es', 'EUR', 'active', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) " - ) +def procesar_factura_verifactu( + factura: Optional[Dict[str, Any]], + cursor_mysql, + config: Dict[str, Any] +) -> bool: - insert_payment_methods_query = ( - "INSERT INTO payment_methods (id, description, factuges_id, created_at, updated_at ) " - "VALUES (%s, %s, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) " - ) + 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) " + ) - insert_customer_invoices_query = ( - "INSERT INTO customer_invoices (id, company_id, status, series, reference, invoice_date, operation_date, " - "subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, taxes_amount_value, total_amount_value, " - "customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, " - "payment_method_id, payment_method_description, " - "subtotal_amount_scale, discount_amount_scale, discount_percentage_scale, taxable_amount_scale, taxes_amount_scale, total_amount_scale, " - "language_code, currency_code, created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 2, 2, 2, 2, 2, 'es', 'EUR', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" - ) + update_verifactu_records = ("UPDATE verifactu_records " + "set estado = %s," + "operacion = %s, " + "updated_at = CURRENT_TIMESTAMP " + "WHERE uuid = %s " + ) - insert_customer_invoices_taxes_query = ( - "INSERT INTO customer_invoice_taxes (tax_id, invoice_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, " - "created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, 2, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" - ) - - insert_customer_invoice_items_query = ( - "INSERT INTO customer_invoice_items (item_id, invoice_id, position, description, quantity_value, unit_amount_value, discount_percentage_value, discount_amount_value, total_amount_value, " - "quantity_scale, unit_amount_scale, discount_amount_scale, total_amount_scale, discount_percentage_scale, created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, 2, 4, 2, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" - ) - - insert_customer_invoice_item_taxes_query = ( - "INSERT INTO customer_invoice_item_taxes (tax_id, item_id, tax_code, taxable_amount_value, taxes_amount_value, taxable_amount_scale, taxes_amount_scale, " - "created_at, updated_at) " - "VALUES (%s, %s, %s, %s, %s, 4, 2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)" - ) - - update_FACTURAS_CLIENTE_query = ( - "UPDATE FACTURAS_CLIENTE set ID_VERIFACTU = ? WHERE ID = ?") - - select_customer_query = ( - "SELECT customers.id " - "FROM customers " - "WHERE customers.factuges_id = %s" - ) - - select_payment_method_query = ( - "SELECT payment_methods.id " - "FROM payment_methods " - "WHERE payment_methods.factuges_id = %s" - ) - - 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: - - factuges_id = int(factura_detalle['ID_FACTURA']) - invoice_status = str('draft') - invoice_series = str('A') - invoice_number = str(factura_detalle['REFERENCIA']) - invoice_date = str(factura_detalle['FECHA_FACTURA']) - operation_date = str(factura_detalle['FECHA_FACTURA']) - # siempre tendrán 2 decimales - subtotal_amount_value = (factura_detalle['IMPORTE_NETO'] or 0)*100 - discount_amount_value = ( - factura_detalle['IMPORTE_DESCUENTO'] or 0)*100 - discount_percentage_value = ( - factura_detalle['DESCUENTO'] or 0)*100 if (factura_detalle['DESCUENTO']) is not None else 0 - taxable_amount_value = (factura_detalle['BASE_IMPONIBLE'] or 0)*100 - # Preparamos el tipo de IVA, en FactuGES es único - tax_code = str(factura_detalle['DES_TIPO_IVA']) - if tax_code == 'IVA21': - tax_code = 'iva_21' - elif tax_code == 'IVA18': - tax_code = 'iva_18' - elif tax_code == 'IVA16': - tax_code = 'iva_16' - elif tax_code == 'IVA10': - tax_code = 'iva_10' - else: - tax_code = '' - # La cuota de impuestos es el IVA + RE - tax_amount_value = ( - (factura_detalle['IMPORTE_IVA'] or 0) + (factura_detalle['IMPORTE_RE'] or 0))*100 - - total_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100 - - payment_method_id = str(uuid4()) - factuges_payment_method_id = str(factura_detalle['ID_FORMA_PAGO']) - payment_method_description = str(factura_detalle['DES_FORMA_PAGO']) - - customer_id = str(uuid4()) - factuges_customer_id = str(factura_detalle['ID_CLIENTE']) - customer_tin = str(factura_detalle['NIF_CIF']) - customer_name = str(factura_detalle['NOMBRE']) - customer_street = str(factura_detalle['CALLE']) - customer_city = str(factura_detalle['POBLACION']) - customer_province = str(factura_detalle['PROVINCIA']) - customer_postal_code = str(factura_detalle['CODIGO_POSTAL']) - customer_phone_primary = factura_detalle['TELEFONO_1'] - customer_phone_secondary = factura_detalle['TELEFONO_2'] - customer_mobile_primary = factura_detalle['MOVIL_1'] - customer_mobile_secondary = factura_detalle['MOVIL_2'] - customer_email_primary = factura_detalle['EMAIL_1'] - customer_email_secondary = factura_detalle['EMAIL_2'] - customer_webside = str(factura_detalle['PAGINA_WEB']) - customer_country = 'es' - - 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'] - item_discount_percentage_value = None if Descuento is None else None if Descuento == 0 else ( - factura_detalle['DESCUENTO'])*100 - item_discount_amount = None - # 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 - 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']) - - 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(select_customer_query, - (factuges_customer_id, )) - row = cursorMySQL.fetchone() - is_new = (row is None) or (row[0] is None) - - if is_new: - logging.info( - f"Inserting customer {factuges_customer_id} {customer_tin} {customer_name}") - cursorMySQL.execute(insert_customer_query, (customer_id, customer_name, customer_tin, customer_street, customer_city, customer_province, - customer_postal_code, customer_country, customer_phone_primary, customer_phone_secondary, customer_mobile_primary, - customer_mobile_secondary, customer_email_primary, customer_email_secondary, customer_webside, factuges_customer_id, cte_company_id)) - else: - # Si ya exite ponemos el id del customer correspondiente - customer_id = str(row[0]) - logging.info( - f"Updating customer {factuges_customer_id} {customer_id}") - # cursorMySQL.execute(update_customer_query, .....) - - # Comprobamos si existe la forma de pago del primer item de la factura - cursorMySQL.execute(select_payment_method_query, - (factuges_payment_method_id, )) - row = cursorMySQL.fetchone() - is_new = (row is None) or (row[0] is None) - - if is_new: - logging.info( - f"Inserting cuspayment method {factuges_payment_method_id} {payment_method_id} {payment_method_description}") - cursorMySQL.execute(insert_payment_methods_query, ( - payment_method_id, payment_method_description, factuges_payment_method_id)) - else: - # Si ya exite ponemos el id del customer correspondiente - payment_method_id = str(row[0]) - logging.info( - f"Updating customer {factuges_payment_method_id} {payment_method_id}") - # cursorMySQL.execute(update_customer_query, .....) - - # Insertamos cabecera de la factura - # Generar un ID único para la tabla customer_invoices - id_customer_invoice = str(uuid4()) + 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"Inserting customer_invoice {id_customer_invoice} {invoice_number} {invoice_date}") - cursorMySQL.execute(insert_customer_invoices_query, (id_customer_invoice, cte_company_id, invoice_status, invoice_series, invoice_number, invoice_date, operation_date, - subtotal_amount_value, discount_amount_value, discount_percentage_value, taxable_amount_value, tax_amount_value, total_amount_value, - customer_id, customer_tin, customer_name, customer_street, customer_city, customer_province, customer_postal_code, customer_country, - payment_method_id, payment_method_description)) - - # Insertamos el IVA y RE si viene - if (factura_detalle['IVA'] > 0): - taxable_amount_value = ( - factura_detalle['BASE_IMPONIBLE'])*100 - tax_amount_value = (factura_detalle['IMPORTE_IVA'])*100 - cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()), - id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value)) - - if (factura_detalle['RECARGO_EQUIVALENCIA'] > 0): - tax_code = 're_5_2' - taxable_amount_value = ( - factura_detalle['BASE_IMPONIBLE'])*100 - tax_amount_value = (factura_detalle['IMPORTE_RE'])*100 - cursorMySQL.execute(insert_customer_invoices_taxes_query, (str(uuid4()), - id_customer_invoice, tax_code, taxable_amount_value, tax_amount_value)) - - # Guardamos en Factuges el id de la customer_invoice - logging.info( - f"Updating FACTURAS_CLIENTE {id_customer_invoice} {factuges_id}") - cursor_FactuGES.execute( - update_FACTURAS_CLIENTE_query, (id_customer_invoice, factuges_id)) - num_fac_procesed += 1 - - # Insertamos detalles y taxes correspondientes siempre - # Generar un ID único para la tabla customer_invoice_items - item_id = str(uuid4()) - logging.info( - f"Inserting customer_invoice_items {id_customer_invoice} {item_position} {item_quantity_value}") - cursorMySQL.execute(insert_customer_invoice_items_query, (item_id, id_customer_invoice, item_position, item_description, - item_quantity_value, item_unit_amount_value, item_discount_percentage_value, item_discount_amount, item_total_amount)) - - if tax_code == 'IVA21': - tax_amount_value = ( - (factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.21))*100 - elif tax_code == 'IVA18': - tax_amount_value = ( - (factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.18))*100 - elif tax_code == 'IVA16': - tax_amount_value = ( - (factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.16))*100 - elif tax_code == 'IVA10': - tax_amount_value = ( - (factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.10))*100 + f">>> Factura {factura.get("reference")} registrada en Verifactu") + return True else: - tax_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100 + 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 - logging.info( - f"Inserting customer_invoice_item_taxes {item_id} {item_position} {tax_code} {item_total_amount} {tax_amount_value}") - cursorMySQL.execute(insert_customer_invoice_item_taxes_query, (str(uuid4()), item_id, tax_code, - item_total_amount, tax_amount_value)) - # Asignamos el id factura anterior para no volver a inserta cabecera - factuges_id_anterior = factuges_id +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 - logging.info( - f"FACTURAS_CLIENTE rows to be processed: {str(num_fac_procesed)}") + 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": [], - 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 + # CAMPOS PARA LOGICA NUESTRA + "id": str(fila['id']), + "reference": str(fila['reference']), + "uuid": fila['uuid'], + } - finally: - # Cerrar la conexión - if cursorMySQL is not None: - cursorMySQL.close() - if cursor_FactuGES is not None: - cursor_FactuGES.close() + 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 diff --git a/app/main.py b/app/main.py index 63850a3..f98d534 100644 --- a/app/main.py +++ b/app/main.py @@ -6,7 +6,7 @@ from datetime import datetime from dateutil import tz from config import setup_logging, load_config from db import get_mysql_connection, get_factuges_connection, sync_invoices, sync_invoices_verifactu -from utils import obtener_fecha_ultima_ejecucion, actualizar_fecha_ultima_ejecucion, log_system_metrics, send_orders_mail +from utils import obtener_fecha_ultima_ejecucion, actualizar_fecha_ultima_ejecucion, log_system_metrics, send_orders_mail, limpiar_cadena def main(): @@ -44,15 +44,24 @@ def main(): logging.info(f"Sync invoices") sync_invoices(conn_factuges, conn_mysql, last_execution_date_local_tz) - # Sync Verifactu - logging.info(f"Sync Verifactu") - sync_invoices_verifactu(conn_mysql, last_execution_date_local_tz) - - # actualizar_fecha_ultima_ejecucion() - # Confirmar los cambios conn_mysql.commit() conn_factuges.commit() + conn_factuges.close() + conn_mysql.close() + + # ESTO OTRO DEBERIA SER OTRA TRANSACCION POR LO QUE HACEMOS NUEVA CONEXION + # Vamos que deberia ir en otro lado + conn_mysql = get_mysql_connection(config) + + # Sync Verifactu + logging.info(f"Sync Verifactu") + sync_invoices_verifactu(conn_mysql, last_execution_date_local_tz) + logging.info(f"SALGO Sync Verifactu") + conn_mysql.commit() + conn_mysql.close() + + # actualizar_fecha_ultima_ejecucion() # Enviar email # send_orders_mail(inserted_orders) diff --git a/app/utils/__init__.py b/app/utils/__init__.py index d21f958..86f4473 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -2,5 +2,7 @@ from .last_execution_helper import actualizar_fecha_ultima_ejecucion, obtener_fe from .log_system_metrics import log_system_metrics from .password import hashPassword from .send_orders_mail import send_orders_mail -from .text_converter import text_converter -from .send_rest_api import send_rest_api +from .text_converter import text_converter, limpiar_cadena +from .send_rest_api import validar_nif, estado_factura, crear_factura +from .tax_catalog_helper import TaxCatalog, get_default_tax_catalog +from .importes_helper import unscale_to_str, unscale_to_decimal diff --git a/app/utils/importes_helper.py b/app/utils/importes_helper.py new file mode 100644 index 0000000..08a13a4 --- /dev/null +++ b/app/utils/importes_helper.py @@ -0,0 +1,37 @@ +from decimal import Decimal, ROUND_HALF_UP +from typing import Any, Optional + + +def unscale_to_decimal(value: Any, scale: Any) -> Decimal: + """ + Convierte un valor escalado (p. ej. 24200 con scale=2) a su valor en unidades (242). + No redondea; sólo mueve el punto decimal. + """ + if value is None: + return Decimal("0") + d = Decimal(str(value)) + s = int(scale or 0) + return d.scaleb(-s) # divide por 10**s + + +def unscale_to_str( + value: Any, + scale: Any, + *, + decimals: Optional[int] = None, + strip_trailing_zeros: bool = True +) -> str: + """ + Igual que unscale_to_decimal, pero devuelve str. + - decimals: fija nº de decimales (p. ej. 2). Si None, no fuerza decimales. + - strip_trailing_zeros: si True, quita ceros y el punto sobrantes. + """ + d = unscale_to_decimal(value, scale) + if decimals is not None: + q = Decimal("1").scaleb(-decimals) # p.ej. 2 -> Decimal('0.01') + d = d.quantize(q, rounding=ROUND_HALF_UP) + + s = format(d, "f") + if strip_trailing_zeros and "." in s: + s = s.rstrip("0").rstrip(".") + return s diff --git a/app/utils/send_rest_api.py b/app/utils/send_rest_api.py index 8825c92..2bc5775 100644 --- a/app/utils/send_rest_api.py +++ b/app/utils/send_rest_api.py @@ -3,13 +3,120 @@ import logging from typing import Optional, Dict, Any, Tuple +def estado_factura(uuid_str: str, + config, + ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]: + """ + Llama al endpoint de Verifacti para cosultar el estado de registro. + + Retorna: + (ok, data, error) + - ok: True si la llamada fue exitosa y resultado == 'IDENTIFICADO' + - data: dict con la respuesta JSON completa + - error: mensaje de error si algo falló + """ + url = config['VERIFACTU_BASE_URL'] + "/verifactu/status" + timeout: int = 10 + headers = {"Content-Type": "application/json", + "Accept": "application/json"} + headers["Authorization"] = "Bearer " + config['VERIFACTU_API_KEY'] + params = {"uuid": uuid_str} + + try: + resp = requests.get( + url, headers=headers, params=params, timeout=timeout) + + if resp.status_code == 200: + try: + data = resp.json() + except ValueError: + return {"ok": False, "status": 200, "error": "Respuesta 200 sin JSON válido", "raw": resp.text} + return {"ok": True, "status": 200, "data": data} + + if resp.status_code == 400: + try: + body = resp.json() + msg = body.get( + "error") or "Error de validación (400) sin detalle" + except ValueError: + msg = f"Error de validación (400): {resp.text}" + return {"ok": False, "status": 400, "error": msg} + + # Otros códigos: devuelve mensaje genérico, intenta extraer JSON si existe + try: + body = resp.json() + msg = body.get("error") or body + return {"ok": False, "status": resp.status_code, "error": str(msg)} + except ValueError: + return {"ok": False, "status": resp.status_code, "error": f"HTTP {resp.status_code}", "raw": resp.text} + + except requests.RequestException as e: + logging.error("Error de conexión con la API Verifacti: %s", e) + return False, None, str(e) + except ValueError as e: + logging.error("Respuesta no es JSON válido: %s", e) + return False, None, "Respuesta no es JSON válido" + + +def crear_factura(payload, + config, + ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]: + """ + Llama al endpoint de Verifacti para crear una factura. + + Retorna: + (ok, data, error) + - ok: True si la llamada fue exitosa y resultado == 'IDENTIFICADO' + - data: dict con la respuesta JSON completa + - error: mensaje de error si algo falló + """ + url = config['VERIFACTU_BASE_URL'] + "/verifactu/create" + timeout: int = 10 + headers = {"Content-Type": "application/json", + "Accept": "application/json"} + headers["Authorization"] = "Bearer " + config['VERIFACTU_API_KEY'] + + try: + resp = requests.post( + url, json=payload, headers=headers, timeout=timeout) + + if resp.status_code == 200: + try: + data = resp.json() + logging.info(data) + except ValueError: + return {"ok": False, "status": 200, "error": "Respuesta 200 sin JSON válido", "raw": resp.text} + return {"ok": True, "status": 200, "data": data} + + if resp.status_code == 400: + try: + body = resp.json() + msg = body.get( + "error") or "Error de validación (400) sin detalle" + except ValueError: + msg = f"Error de validación (400): {resp.text}" + return {"ok": False, "status": 400, "error": msg} + + # Otros códigos: devuelve mensaje genérico, intenta extraer JSON si existe + try: + body = resp.json() + msg = body.get("error") or body + return {"ok": False, "status": resp.status_code, "error": str(msg)} + except ValueError: + return {"ok": False, "status": resp.status_code, "error": f"HTTP {resp.status_code}", "raw": resp.text} + + except requests.RequestException as e: + logging.error("Error de conexión con la API Verifacti: %s", e) + return False, None, str(e) + except ValueError as e: + logging.error("Respuesta no es JSON válido: %s", e) + return False, None, "Respuesta no es JSON válido" + + def validar_nif( nif: str, nombre: str, - *, - timeout: int = 10, - api_key: Optional[str] = None, - bearer_token: Optional[str] = None, + config, ) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]: """ Llama al endpoint de Verifacti para validar un NIF. @@ -20,32 +127,27 @@ def validar_nif( - data: dict con la respuesta JSON completa - error: mensaje de error si algo falló """ - url = "https://api.verifacti.com/nifs/validar" + url = config['VERIFACTU_BASE_URL'] + "/nifs/validar" + timeout: int = 10 headers = {"Content-Type": "application/json", "Accept": "application/json"} - - if api_key: - headers["X-API-KEY"] = api_key - if bearer_token: - headers["Authorization"] = f"Bearer {bearer_token}" - - payload = { - "nif": nif, - "nombre": nombre, - } + headers["Authorization"] = "Bearer " + config['VERIFACTU_NIFS_API_KEY'] + payload = {"nif": nif, + "nombre": nombre, + } try: resp = requests.post( url, json=payload, headers=headers, timeout=timeout) if resp.status_code != 200: - return False, None, f"HTTP {resp.status_code}: {resp.text}" + logging.info(f"ERRRRRROOOOOOORRRRR LLAMADA REST API") + # return False, None, f"HTTP {resp.status_code}: {resp.text}" data = resp.json() - resultado = data.get("resultado") + resultado = data.get("resultado", "NO IDENTIFICADO") + logging.info(f"Resultado Verifacti: {resultado}") - # La lógica de validación: 'IDENTIFICADO' = válido - ok = resultado == "IDENTIFICADO" - return ok, data, None + return resultado == "IDENTIFICADO" except requests.RequestException as e: logging.error("Error de conexión con la API Verifacti: %s", e) diff --git a/app/utils/tax_catalog_helper.py b/app/utils/tax_catalog_helper.py new file mode 100644 index 0000000..f097ca8 --- /dev/null +++ b/app/utils/tax_catalog_helper.py @@ -0,0 +1,207 @@ +import json +from decimal import Decimal +from typing import Any, Dict, Iterable, Optional, Tuple +from functools import lru_cache +from pathlib import Path + + +def reduce_scale_pair(value: int, scale: int, *, min_scale: int = 0) -> Tuple[int, int]: + """ + Reduce la escala todo lo posible eliminando ceros finales de 'value', + sin bajar de 'min_scale'. Mantiene exactitud (no redondea). + Ejemplos: + (2100, 2) -> (21, 0) + (750, 2) -> (75, 1) + (400, 2) -> (4, 0) + (275, 2) -> (275, 2) # no se puede reducir (no termina en 0) + """ + v, s = int(value), int(scale) + while s > min_scale and v % 10 == 0: + v //= 10 + s -= 1 + return v, s + + +class TaxCatalog: + """ + Carga un catálogo de tipos (IVA, RE, IGIC, etc.) y permite consultar por 'code'. + + - get_value_scale(code) -> (value:int, scale:int) | None + - require_value_scale(code) -> (value:int, scale:int) o KeyError si no existe + - get_percent(code) -> Decimal('21.00') # ejemplo para iva_21 + - get_fraction(code) -> Decimal('0.21') # 21% como 0.21 + """ + + DEFAULT_FILENAME = "spain_tax_catalog.json" + + def __init__(self, index: Dict[str, Dict[str, Any]]): + self._index = index # code -> info normalizada + + # -------- Factorías / constructores -------- + @classmethod + def from_file(cls, path: str) -> "TaxCatalog": + with open(path, "r", encoding="utf-8") as f: + items = json.load(f) + return cls(cls._build_index(items)) + + @classmethod + def from_json(cls, json_str: str) -> "TaxCatalog": + items = json.loads(json_str) + return cls(cls._build_index(items)) + + @classmethod + def from_iterable(cls, items: Iterable[Dict[str, Any]]) -> "TaxCatalog": + return cls(cls._build_index(items)) + + @classmethod + def create(cls, filename: Optional[str] = None) -> "TaxCatalog": + """ + Carga automáticamente el JSON del mismo directorio que este archivo. + Uso: catalog = TaxCatalog.create() + """ + fname = filename or cls.DEFAULT_FILENAME + base_dir = Path(__file__).resolve( + ).parent if "__file__" in globals() else Path.cwd() + path = base_dir / fname + if not path.exists(): + raise FileNotFoundError(f"No se encontró el catálogo en: {path}") + return cls.from_file(str(path)) + + def reduce_value_scale( + self, code: str, default: Optional[Tuple[int, int]] = None, *, min_scale: int = 0 + ) -> Optional[Tuple[int, int]]: + """ + Usa get_value_scale(code) y reduce la escala con exactitud. + """ + vs = self.get_value_scale(code) + if vs is None: + return default + value, scale = vs + return reduce_scale_pair(value, scale, min_scale=min_scale) + + # -------- API pública -------- + def get_percent_reduced( + self, code: str, default: Optional[Decimal] = None + ) -> Optional[Decimal]: + """ + Devuelve el porcentaje como Decimal usando la pareja (value, scale) + ya reducida: p. ej., iva_21 -> Decimal('21'), iva_7_5 -> Decimal('7.5'). + """ + rs = self.reduce_value_scale(code) + if rs is None: + return default + value_r, scale_r = rs + return Decimal(value_r).scaleb(-scale_r) + + def get_value_scale( + self, code: str, default: Optional[Tuple[int, int]] = None + ) -> Optional[Tuple[int, int]]: + info = self._index.get(code.lower()) + if info is None: + return default + return info["value"], info["scale"] + + def require_value_scale(self, code: str) -> Tuple[int, int]: + res = self.get_value_scale(code) + if res is None: + raise KeyError(f"Código no encontrado en catálogo: {code}") + return res + + def get_percent(self, code: str, default: Optional[Decimal] = None) -> Optional[Decimal]: + vs = self.get_value_scale(code) + if vs is None: + return default + value, scale = vs + return Decimal(value).scaleb(-scale) # 2100, scale 2 -> 21.00 + + def get_fraction(self, code: str, default: Optional[Decimal] = None) -> Optional[Decimal]: + pct = self.get_percent(code) + if pct is None: + return default + # 21.00 -> 0.21 + return (pct / Decimal(100)).quantize(Decimal("0.0000001")).normalize() + + def is_non_exempt( + self, + code: str, + *, + include_zero_rate_as_exempt: bool = False, + default: Optional[bool] = None, + ) -> Optional[bool]: + """ + Devuelve True si 'code' NO es exento/asimilado (es decir, se repercute impuesto). + Devuelve False si es exento/asimilado. + Si include_zero_rate_as_exempt=True, los tipos 0% también se tratan como exentos. + Si el code no existe en el catálogo, devuelve 'default'. + """ + key = (code or "").lower() + info = self._index.get(key) + if info is None: + return default + + value = info.get("value", 0) # p.ej. 2100 + aeat = (info.get("aeat_code") or "").upper() # p.ej. "01" + name = (info.get("name") or "").lower() + code_l = key + + # Códigos AEAT típicos de exento/no sujeto/intracomunitario/exportación + # (según tu catálogo de ejemplo) + EXEMPT_AEAT = {"04", "06", "E5", "E6", "E2", "12"} + + # Heurística de "exento o asimilado" + exempt_like = ( + "exent" in code_l # exenta/exento + or "exent" in name + or "no_sujeto" in code_l + or aeat in EXEMPT_AEAT + ) + + # Opcional: tratar 0% como exento (iva_0, igic_0...) + if include_zero_rate_as_exempt and int(value) == 0: + exempt_like = True + + return not exempt_like + + # -------- Internos -------- + @staticmethod + def _build_index(items: Iterable[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: + index: Dict[str, Dict[str, Any]] = {} + for it in items: + code = (it.get("code") or "").lower().strip() + if not code: + continue + + raw_value = str(it.get("value", "0")).strip() + raw_scale = str(it.get("scale", "0")).strip() + + try: + value_int = int(Decimal(raw_value)) + except Exception as e: + raise ValueError( + f"value inválido para code={code}: {raw_value}") from e + + try: + scale_int = int(raw_scale) + except Exception as e: + raise ValueError( + f"scale inválido para code={code}: {raw_scale}") from e + + index[code] = { + "name": it.get("name"), + "group": it.get("group"), + "description": it.get("description"), + "aeat_code": it.get("aeat_code"), + "value": value_int, # p.ej. 2100 + "scale": scale_int, # p.ej. 2 + } + return index + + +# -------- Singleton con caché (opcional, recomendable) -------- +@lru_cache(maxsize=1) +def get_default_tax_catalog() -> TaxCatalog: + """ + Devuelve una instancia singleton cargada de 'spain_tax_catalog.json' + en el mismo directorio. Se cachea para evitar relecturas. + """ + return TaxCatalog.create() diff --git a/app/utils/text_converter.py b/app/utils/text_converter.py index 8144200..22c6296 100644 --- a/app/utils/text_converter.py +++ b/app/utils/text_converter.py @@ -1,4 +1,5 @@ import logging +import re def text_converter(texto, charset_destino='ISO8859_1', longitud_maxima=None): @@ -36,3 +37,16 @@ def text_converter(texto, charset_destino='ISO8859_1', longitud_maxima=None): except Exception as e: logging.error(f"Error inesperado al convertir texto: {str(e)}") return "" + + +def limpiar_cadena(texto: str) -> str: + """ + Elimina espacios, guiones y cualquier carácter no alfanumérico. + Ejemplos: + 'B 83999441' -> 'B83999441' + 'B-83999441' -> 'B83999441' + 'B_83 99-94.41' -> 'B83999441' + """ + if not isinstance(texto, str): + texto = str(texto) + return re.sub(r'[^A-Za-z0-9]', '', texto)