206 lines
8.0 KiB
Python
206 lines
8.0 KiB
Python
import logging
|
|
from typing import Dict, Any, Tuple, Optional, List, Iterable
|
|
from app.config import load_config
|
|
from decimal import Decimal
|
|
from app.utils import validar_nif, estado_factura, crear_factura, TaxCatalog, 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}
|
|
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, cursor_mysql, config)
|
|
logging.info(f"Ok send Verifactu")
|
|
else:
|
|
logging.info(f"There are no rows to send")
|
|
|
|
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(error)
|
|
raise error
|
|
|
|
|
|
def enviar_datos(invoices_to_verifactu, cursor_mysql, config):
|
|
|
|
# Recorrer todas las facturas para crear json de envio
|
|
try:
|
|
invoice_id = None
|
|
factura = None
|
|
for fila in invoices_to_verifactu:
|
|
# 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 != str(fila['id']):
|
|
|
|
procesar_factura_verifactu(factura, cursor_mysql, config)
|
|
|
|
# preparamos nueva factura
|
|
ok, respuesta = preparar_factura(fila)
|
|
if not ok:
|
|
logging.info(
|
|
f"ERROR >>>>>> Factura {fila['reference']} no cumple requisitos para ser mandada a Verifactu:")
|
|
logging.info(
|
|
f">>>>>> Faltan campos requeridos: {respuesta}")
|
|
factura = None
|
|
continue
|
|
factura = respuesta
|
|
|
|
ok, linea = preparar_linea(fila)
|
|
if not ok:
|
|
logging.info(
|
|
f"ERROR >>>>>> 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
|
|
logging.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 != None:
|
|
# Creamos registro de factura en verifactu
|
|
if factura.get('uuid') == '':
|
|
# 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")
|
|
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")))
|
|
logging.info(
|
|
f">>> Factura {factura.get("reference")} registrada en Verifactu")
|
|
return True
|
|
else:
|
|
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(SQL.update_verifactu_records_with_uuid, (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
|
|
|
|
|
|
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
|
|
|
|
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
|
|
|
|
|
|
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
|