256 lines
11 KiB
Python
256 lines
11 KiB
Python
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):
|
|
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= '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, 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"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
|
|
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]
|
|
# 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"Ha ido bien enviar_datos")
|
|
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 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
|
|
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:
|
|
|
|
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) "
|
|
)
|
|
|
|
update_verifactu_records = ("UPDATE verifactu_records "
|
|
"set estado = %s,"
|
|
"operacion = %s, "
|
|
"updated_at = CURRENT_TIMESTAMP "
|
|
"WHERE uuid = %s "
|
|
)
|
|
|
|
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">>> 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(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
|
|
|
|
|
|
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
|