ASE/src/load_orchestrator.py

#!.venv/bin/python
"""
Orchestratore dei worker che caricano i dati su dataraw
"""

# Import necessary libraries
import asyncio
import importlib
import logging

# Import custom modules for configuration and database connection
from utils.config import loader_load_data as setting
from utils.csv.loaders import get_next_csv_atomic
from utils.database import WorkflowFlags
from utils.orchestrator_utils import run_orchestrator, shutdown_event, worker_context

# Initialize the logger for this module
logger = logging.getLogger()

# Delay tra un processamento CSV e il successivo (in secondi)
CSV_PROCESSING_DELAY = 0.2
# Tempo di attesa se non ci sono record da elaborare
NO_RECORD_SLEEP = 60

# Module import cache to avoid repeated imports (performance optimization)
_module_cache = {}


async def worker(worker_id: int, cfg: dict, pool: object) -> None:
    """Esegue il ciclo di lavoro per l'elaborazione dei file CSV.

    Il worker preleva un record CSV dal database, ne elabora il contenuto
    e attende prima di iniziare un nuovo ciclo.

    Supporta graceful shutdown controllando il shutdown_event tra le iterazioni.

    Args:
        worker_id (int): L'ID univoco del worker.
        cfg (dict): L'oggetto di configurazione.
        pool (object): Il pool di connessioni al database.
    """
    # Imposta il context per questo worker
    worker_context.set(f"W{worker_id:02d}")

    logger.info("Avviato")

    try:
        while not shutdown_event.is_set():
            try:
                logger.info("Inizio elaborazione")
                record = await get_next_csv_atomic(
                    pool,
                    cfg.dbrectable,
                    WorkflowFlags.CSV_RECEIVED,
                    WorkflowFlags.DATA_LOADED,
                )

                if record:
                    success = await load_csv(record, cfg, pool)
                    if not success:
                        logger.error("Errore durante l'elaborazione")
                    await asyncio.sleep(CSV_PROCESSING_DELAY)
                else:
                    logger.info("Nessun record disponibile")
                    await asyncio.sleep(NO_RECORD_SLEEP)

            except asyncio.CancelledError:
                logger.info("Worker cancellato. Uscita in corso...")
                raise

            except Exception as e:  # pylint: disable=broad-except
                logger.error("Errore durante l'esecuzione: %s", e, exc_info=1)
                await asyncio.sleep(1)

    except asyncio.CancelledError:
        logger.info("Worker terminato per shutdown graceful")
    finally:
        logger.info("Worker terminato")


async def load_csv(record: tuple, cfg: object, pool: object) -> bool:
    """Carica ed elabora un record CSV utilizzando il modulo di parsing appropriato.

    Args:
        record: Una tupla contenente i dettagli del record CSV da elaborare
                (rec_id, unit_type, tool_type, unit_name, tool_name).
        cfg: L'oggetto di configurazione contenente i parametri del sistema.
        pool (object): Il pool di connessioni al database.

    Returns:
        True se l'elaborazione del CSV è avvenuta con successo, False altrimenti.
    """

    debug_mode = logging.getLogger().getEffectiveLevel() == logging.DEBUG
    logger.debug("Inizio ricerca nuovo CSV da elaborare")

    rec_id, unit_type, tool_type, unit_name, tool_name = [x.lower().replace(" ", "_") if isinstance(x, str) else x for x in record]
    logger.info(
        "Trovato CSV da elaborare: ID=%s, Tipo=%s_%s, Nome=%s_%s",
        rec_id,
        unit_type,
        tool_type,
        unit_name,
        tool_name,
    )

    # Costruisce il nome del modulo da caricare dinamicamente
    module_names = [
        f"utils.parsers.by_name.{unit_name}_{tool_name}",
        f"utils.parsers.by_name.{unit_name}_{tool_type}",
        f"utils.parsers.by_name.{unit_name}_all",
        f"utils.parsers.by_type.{unit_type}_{tool_type}",
    ]

    # Try to get from cache first (performance optimization)
    modulo = None
    cache_key = None

    for module_name in module_names:
        if module_name in _module_cache:
            # Cache hit! Use cached module
            modulo = _module_cache[module_name]
            cache_key = module_name
            logger.info("Modulo caricato dalla cache: %s", module_name)
            break

    # If not in cache, import dynamically
    if not modulo:
        for module_name in module_names:
            try:
                logger.debug("Caricamento dinamico del modulo: %s", module_name)
                modulo = importlib.import_module(module_name)
                # Store in cache for future use
                _module_cache[module_name] = modulo
                cache_key = module_name
                logger.info("Modulo caricato per la prima volta: %s", module_name)
                break
            except (ImportError, AttributeError) as e:
                logger.debug(
                    "Modulo %s non presente o non valido. %s",
                    module_name,
                    e,
                    exc_info=debug_mode,
                )

    if not modulo:
        logger.error("Nessun modulo trovato %s", module_names)
        return False

    # Ottiene la funzione 'main_loader' dal modulo
    funzione = modulo.main_loader

    # Esegui la funzione
    logger.info("Elaborazione con modulo %s per ID=%s", modulo, rec_id)
    await funzione(cfg, rec_id, pool)
    logger.info("Elaborazione completata per ID=%s", rec_id)
    return True


async def main():
    """Funzione principale che avvia il load_orchestrator."""
    await run_orchestrator(setting.Config, worker)


if __name__ == "__main__":
    asyncio.run(main())