version sinc con verifactu test
This commit is contained in:
parent
931f9a6825
commit
405a6317ed
@ -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=
|
||||
|
||||
|
||||
@ -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'),
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
23
app/main.py
23
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)
|
||||
|
||||
@ -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
|
||||
|
||||
37
app/utils/importes_helper.py
Normal file
37
app/utils/importes_helper.py
Normal file
@ -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
|
||||
@ -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)
|
||||
|
||||
207
app/utils/tax_catalog_helper.py
Normal file
207
app/utils/tax_catalog_helper.py
Normal file
@ -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()
|
||||
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user