Uecko/Uecko_ERP_FactuGES_sync
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cambios

Browse Source
This commit is contained in:
David Arranz 2025-10-03 20:22:15 +02:00
parent b25df0ce9b
commit 931f9a6825
6 changed files with 447 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
app/db/__init__.py
View File

@ -4,3 +4,4 @@ from .sync_catalog import sync_catalog
from .sync_dealers import sync_dealers
from .sync_orders import sync_orders
from .sync_invoices import sync_invoices
from .sync_invoices_verifactu import sync_invoices_verifactu

31
app/db/sync_invoices.py
View File

@ -1,7 +1,7 @@
import logging
from uuid import uuid4
from config import load_config
from decimal import Decimal
from decimal import Decimal, ROUND_HALF_UP
def sync_invoices(conn_factuges, conn_mysql, last_execution_date):
@ -173,7 +173,7 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
)
insert_customer_invoices_query = (
"INSERT INTO customer_invoices (id, company_id, status, series, invoice_number, invoice_date, operation_date, "
"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, "
@ -291,8 +291,10 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
# 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 = None if factura_detalle['IMPORTE_TOTAL_DET'] is None else (
factura_detalle['IMPORTE_TOTAL_DET'] or 0)*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'])
@ -378,23 +380,24 @@ def insertar_datos(conn_mysql, filas, conn_factuges, config):
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':
if tax_code == 'iva_21':
tax_amount_value = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.21))*100
elif tax_code == 'IVA18':
(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 = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.18))*100
elif tax_code == 'IVA16':
(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 = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.16))*100
elif tax_code == 'IVA10':
(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 = (
(factura_detalle['IMPORTE_TOTAL'] or 0)*Decimal(0.10))*100
(Decimal(str(factura_detalle['IMPORTE_TOTAL_DET'] or 0))*Decimal('0.10')).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))*100
else:
tax_amount_value = item_total_amount
tax_amount_value = (
factura_detalle['IMPORTE_TOTAL_DET'] or 0)*100
logging.info(
f"Inserting customer_invoice_item_taxes {item_id} {item_position} {tax_code} {tax_amount_value}")
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))

367
app/db/sync_invoices_verifactu.py Normal file
View File

@ -0,0 +1,367 @@
import logging
from uuid import uuid4
from config import load_config
from decimal import Decimal
def sync_invoices_verifactu(conn_mysql, last_execution_date):
config = load_config()
# OPCION A SACAMOS EL RESUMEN DE LA TAXES DE LA CABECERA
# SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value,
# cit.tax_code, sum(cit.taxable_amount_value), sum(cit.taxes_amount_value)
# FROM customer_invoices as ci
# LEFT JOIN customer_invoice_taxes cit on (ci.id = cit.invoice_id)
# WHERE (ci.is_proforma = 0) AND (ci.status= 'draft')
# group by 1,2,3,4,5,6,7,8,9
# OPCION B SACAMOS LOS IVAS DE LOS DETALLES DE LOS ITEM
# SELECT ci.id, ci.series, ci.invoice_number, ci.invoice_date, ci.description, ci.customer_tin, ci.customer_name, ci.total_amount_value,
# ciit.tax_code, sum(ciit.taxable_amount_value), sum(ciit.taxes_amount_value)
# FROM customer_invoices as ci
# LEFT JOIN customer_invoice_items cii on (ci.id = cii.invoice_id)
# LEFT JOIN customer_invoice_item_taxes ciit on (cii.item_id = ciit.item_id)
# WHERE (ci.is_proforma = 0) AND (ci.status= 'draft')
# 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"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"
)
# Crear un cursor para ejecutar consultas SQL
cursor_mysql = None
try:
cursor_mysql = conn_mysql.cursor()
# Ejecutar la consulta de FACTURAS_CLIENTE
cursor_mysql.execute(consulta_sql_customer_invoices_issue)
filas = cursor_mysql.fetchall()
# Obtener los nombres de las columnas
columnas = [desc[0] for desc in cursor_mysql.description]
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)}")
# Verificar si hay filas en el resultado
if tuplas_seleccionadas:
enviar_datos(tuplas_seleccionadas, config)
else:
logging.info(f"There are no rows to send")
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
def enviar_datos(invoices_to_verifactu, config):
# Recorrer todas las facturas para crear json de envio
try:
logging.info(f"Send to Verifactu")
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}")
except Exception as e:
# Escribir el error en el archivo de errores
logging.error(str(e))
raise e # Re-lanzar la excepción para detener el procesamiento
def 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) "
)
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_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)"
)
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())
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
else:
tax_amount_value = (factura_detalle['IMPORTE_TOTAL'] or 0)*100
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
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()

7
app/main.py
View File

@ -5,7 +5,7 @@ import __version__
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
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
@ -41,8 +41,13 @@ def main():
conn_mysql = get_mysql_connection(config)
# Sync invoices
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

1
app/utils/__init__.py
View File

@ -3,3 +3,4 @@ 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

55
app/utils/send_rest_api.py Normal file
View File

@ -0,0 +1,55 @@
import requests
import logging
from typing import Optional, Dict, Any, Tuple
def validar_nif(
nif: str,
nombre: str,
*,
timeout: int = 10,
api_key: Optional[str] = None,
bearer_token: Optional[str] = None,
) -> Tuple[bool, Optional[Dict[str, Any]], Optional[str]]:
"""
Llama al endpoint de Verifacti para validar un NIF.
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 = "https://api.verifacti.com/nifs/validar"
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,
}
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}"
data = resp.json()
resultado = data.get("resultado")
# La lógica de validación: 'IDENTIFICADO' = válido
ok = resultado == "IDENTIFICADO"
return ok, data, None
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"