260 lines
8.9 KiB
Python
260 lines
8.9 KiB
Python
from typing import Any, Dict, Iterable, List, Optional, Tuple
|
|
|
|
from app.config import load_config, logger
|
|
from app.utils import TaxCatalog, crear_factura, estado_factura, unscale_to_str
|
|
|
|
from . import sql_sentences as SQL
|
|
|
|
|
|
def sync_invoices_verifactu(conn_mysql, last_execution_date):
|
|
config = load_config()
|
|
|
|
# Crear un cursor para ejecutar consultas SQL
|
|
cursor_mysql = None
|
|
try:
|
|
cursor_mysql = conn_mysql.cursor()
|
|
# Ejecutar la consulta de customer invoices a enviar
|
|
cursor_mysql.execute(SQL.consulta_sql_customer_invoices_issue)
|
|
filas = cursor_mysql.fetchall()
|
|
|
|
# Obtener los nombres de las columnas
|
|
columnas = [desc[0] for desc in cursor_mysql.description]
|
|
# 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)
|
|
|
|
# invoices_to_verifactu = {str(fila[0]) for fila in filas}
|
|
logger.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, cursor_mysql, config)
|
|
logger.info("Ok send Verifactu")
|
|
else:
|
|
logger.info("There are no rows to send")
|
|
|
|
except Exception as error:
|
|
if cursor_mysql is not None:
|
|
cursor_mysql.close()
|
|
logger.error(
|
|
f"(ERROR) Failed to fetch from database:{config['FWEB_MYSQL_DATABASE']} - using user:{config['FWEB_MYSQL_USER']}")
|
|
logger.error(error)
|
|
raise error
|
|
|
|
|
|
def enviar_datos(
|
|
invoices_to_verifactu,
|
|
cursor_mysql,
|
|
config: Dict[str, Any]
|
|
) -> None:
|
|
|
|
factura: Optional[Dict[str, Any]] = None
|
|
invoice_id: Optional[str] = None
|
|
|
|
# Recorrer todas las facturas para crear json de envio
|
|
try:
|
|
for fila in invoices_to_verifactu:
|
|
fila_id = str(fila["id"])
|
|
|
|
# Si los ids de factura anterior y actual no coinciden, 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 != fila_id:
|
|
|
|
# Cerrar la factura anterior (si existe)
|
|
if factura is not None:
|
|
procesar_factura_verifactu(factura, cursor_mysql, config)
|
|
|
|
# preparamos nueva factura
|
|
ok, factura_nueva, error = preparar_factura(fila)
|
|
if not ok:
|
|
logger.info(
|
|
f"ERROR >>>>>> Factura {fila['reference']} no cumple requisitos para ser mandada a Verifactu:")
|
|
logger.info(f"ERROR >>> Factura {fila['reference']} no válida: {error}")
|
|
continue
|
|
|
|
factura = factura_nueva
|
|
invoice_id = fila_id
|
|
|
|
# Añadir línea a factura actual
|
|
ok, linea = preparar_linea(fila)
|
|
if not ok:
|
|
ref = factura["reference"] if factura else "(desconocida)"
|
|
logger.info(
|
|
f"ERROR >>>>>> Factura {ref} no cumple requisitos para ser mandada a Verifactu:")
|
|
logger.info(f">>>>>> Faltan campos requeridos: {linea}")
|
|
factura = None
|
|
continue
|
|
|
|
# Línea válida
|
|
# Garantizamos que factura no es None
|
|
if factura is not None:
|
|
factura["lineas"].append(linea)
|
|
|
|
# Procesar la última factura
|
|
if factura is not None:
|
|
procesar_factura_verifactu(factura, cursor_mysql, config)
|
|
|
|
except Exception as e:
|
|
# Escribir el error en el archivo de errores
|
|
logger.error(str(e))
|
|
raise e # Re-lanzar la excepción para detener el procesamiento
|
|
|
|
|
|
def procesar_factura_verifactu(
|
|
factura: Optional[Dict[str, Any]],
|
|
cursor_mysql,
|
|
config: Dict[str, Any]
|
|
) -> bool:
|
|
|
|
if factura is None:
|
|
return False
|
|
|
|
# Creamos registro de factura en verifactu
|
|
if factura.get('uuid') == '':
|
|
# logger.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") or {}
|
|
qr_verifactu = f"data:image/png;base64,{data.get('qr', '')}"
|
|
|
|
cursor_mysql.execute(
|
|
SQL.update_verifactu_records_with_invoiceId,
|
|
(
|
|
data.get("estado"),
|
|
data.get("uuid"),
|
|
data.get("url"),
|
|
qr_verifactu,
|
|
factura.get("id"),
|
|
)
|
|
)
|
|
logger.info(
|
|
f">>> Factura {factura.get('reference')} registrada en Verifactu")
|
|
return True
|
|
else:
|
|
logger.info(
|
|
f">>> Factura {factura.get('reference')} enviada a Verifactu con error {respuesta}")
|
|
return False
|
|
|
|
# Actualizamos registro de factura en verifactu
|
|
|
|
# logger.info(f"Send to update Verifactu: {factura}")
|
|
uuid_val = factura.get("uuid")
|
|
if not uuid_val: # None, '', o falsy
|
|
logger.error(f"Factura {factura.get('reference')} sin UUID válido")
|
|
return False
|
|
|
|
respuesta = estado_factura(uuid_val, config)
|
|
|
|
if respuesta.get("status") == 200 and respuesta.get("ok"):
|
|
data = respuesta.get("data") or {}
|
|
cursor_mysql.execute(
|
|
SQL.update_verifactu_records_with_uuid,
|
|
(data.get('estado'),
|
|
data.get('operacion'),
|
|
uuid_val)
|
|
)
|
|
logger.info(
|
|
f">>> Factura {factura.get('reference')} actualizado registro de Verifactu")
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
def preparar_factura(
|
|
fila: Dict[str, Any]
|
|
) -> Tuple[bool, Dict[str, Any] | None, str | None]:
|
|
"""
|
|
Prepara el JSON de factura para Verifactu a partir de 'fila'.
|
|
|
|
Retorna:
|
|
(ok, factura, error)
|
|
"""
|
|
|
|
campos_requeridos = (
|
|
"series", "invoice_number", "invoice_date",
|
|
"description", "total_amount_value"
|
|
)
|
|
|
|
ok, missing = validar_requeridos(fila, campos_requeridos)
|
|
|
|
if not ok:
|
|
# Devolvemos string, NO lista
|
|
return False, None, ", ".join(missing)
|
|
|
|
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": [],
|
|
|
|
# CAMPOS PARA LOGICA NUESTRA
|
|
"id": str(fila['id']),
|
|
"reference": str(fila['reference']),
|
|
"uuid": fila['uuid'],
|
|
}
|
|
|
|
return True, factura, None
|
|
|
|
|
|
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
|