ASE/src/send_orchestrator.py

#!.venv/bin/python
"""
Orchestratore dei worker che inviano i dati ai clienti
"""

# Import necessary libraries
import asyncio
import logging

# Import custom modules for configuration and database connection
from utils.config import loader_send_data as setting
from utils.connect.send_data import process_workflow_record
from utils.csv.loaders import get_next_csv_atomic
from utils.database import WorkflowFlags
from utils.general import alterna_valori
from utils.orchestrator_utils import run_orchestrator, shutdown_event, worker_context

# from utils.ftp.send_data import   ftp_send_elab_csv_to_customer, api_send_elab_csv_to_customer, \
#                                   ftp_send_raw_csv_to_customer, api_send_raw_csv_to_customer


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

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


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

    Il worker preleva un record dal database che indica dati pronti per
    l'invio (sia raw che elaborati), li processa 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}")

    debug_mode = logging.getLogger().getEffectiveLevel() == logging.DEBUG
    logger.info("Avviato")

    alternatore = alterna_valori(
        [WorkflowFlags.CSV_RECEIVED, WorkflowFlags.SENT_RAW_DATA],
        [WorkflowFlags.DATA_ELABORATED, WorkflowFlags.SENT_ELAB_DATA],
    )

    try:
        while not shutdown_event.is_set():
            try:
                logger.info("Inizio elaborazione")

                status, fase = next(alternatore)
                record = await get_next_csv_atomic(pool, cfg.dbrectable, status, fase)

                if record:
                    await process_workflow_record(record, fase, cfg, pool)
                    await asyncio.sleep(ELAB_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=debug_mode)
                await asyncio.sleep(1)

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


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


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