#!.venv/bin/python # Import necessary libraries import logging import importlib import asyncio import os import aiomysql import contextvars # Import custom modules for configuration and database connection from utils.config import loader_load_data as setting from utils.database import CSV_RECEIVED from utils.csv.loaders import get_next_csv_atomic # Crea una context variable per identificare il worker worker_context = contextvars.ContextVar("worker_id", default="00") # Formatter personalizzato che include il worker_id class WorkerFormatter(logging.Formatter): """Formatter personalizzato per i log che include l'ID del worker.""" def format(self, record): """Formatta il record di log includendo l'ID del worker. Args: record: Il record di log da formattare. Returns: La stringa formattata del record di log. """ record.worker_id = worker_context.get() return super().format(record) # 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 async def worker(worker_id: int, cfg: object, pool) -> 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. Args: worker_id (int): L'ID univoco del worker. cfg (object): L'oggetto di configurazione. pool: Il pool di connessioni al database. """ # Imposta il context per questo worker worker_context.set(f"W{worker_id}") debug_mode = logging.getLogger().getEffectiveLevel() == logging.DEBUG logger.info("Avviato") while True: try: logger.info("Inizio elaborazione") record = await get_next_csv_atomic(pool, cfg.dbrectable, CSV_RECEIVED) 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 Exception as e: logger.error(f"Errore durante l'esecuzione: {e}", exc_info=debug_mode) await asyncio.sleep(1) async def load_csv(record: tuple, cfg: object, pool) -> 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 (id, unit_type, tool_type, unit_name, tool_name). cfg: L'oggetto di configurazione contenente i parametri del sistema. pool: 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") id, unit_type, tool_type, unit_name, tool_name = [ x.lower().replace(" ", "_") if isinstance(x, str) else x for x in record ] logger.info( f"Trovato CSV da elaborare: ID={id}, Tipo={unit_type}_{tool_type}, Nome={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}", ] modulo = None for module_name in module_names: try: logger.debug(f"Caricamento dinamico del modulo: {module_name}") modulo = importlib.import_module(module_name) logger.info(f"Funzione 'main_loader' caricata dal modulo {module_name}") break except (ImportError, AttributeError) as e: logger.debug( f"Modulo {module_name} non presente o non valido. {e}", exc_info=debug_mode, ) if not modulo: logger.error(f"Nessun modulo trovato {module_names}") return False # Ottiene la funzione 'main_loader' dal modulo funzione = getattr(modulo, "main_loader") # Esegui la funzione logger.info(f"Elaborazione con modulo {modulo} per ID={id}") await funzione(cfg, id, pool) logger.info(f"Elaborazione completata per ID={id}") return True async def main(): """Funzione principale che inizializza e avvia il sistema. Questa funzione si occupa di: - Caricare la configurazione. - Impostare il logging. - Creare un pool di connessioni al database. - Avviare i worker concorrenti per l'elaborazione dei CSV. - Gestire l'arresto controllato del sistema. """ logger.info("Avvio del sistema...") cfg = setting.Config() logger.info("Configurazione caricata correttamente") try: # Configura il logging globale log_level = os.getenv("LOG_LEVEL", "INFO").upper() debug_mode = logging.getLogger().getEffectiveLevel() == logging.DEBUG # Configura il logging con il formatter personalizzato handler = logging.FileHandler(cfg.logfilename) formatter = WorkerFormatter( "%(asctime)s - PID: %(process)d.Worker-%(worker_id)s.%(name)s.%(funcName)s.%(levelname)s: %(message)s" ) handler.setFormatter(formatter) # Rimuovi eventuali handler esistenti e aggiungi il nostro logger.handlers.clear() logger.addHandler(handler) logger.setLevel(getattr(logging, log_level)) logger.info("Logging configurato correttamente") # Numero massimo di worker concorrenti logger.info(f"Avvio di {cfg.max_threads} worker concorrenti") pool = await aiomysql.create_pool( host=cfg.dbhost, user=cfg.dbuser, password=cfg.dbpass, db=cfg.dbname, minsize=cfg.max_threads, maxsize=cfg.max_threads * 4, pool_recycle=3600, ) # Avvia i worker workers = [ asyncio.create_task(worker(i, cfg, pool)) for i in range(cfg.max_threads) ] logger.info("Sistema avviato correttamente. In attesa di nuovi task...") try: await asyncio.gather(*workers, return_exceptions=debug_mode) finally: pool.close() await pool.wait_closed() except KeyboardInterrupt: logger.info("Info: Shutdown richiesto... chiusura in corso") except Exception as e: logger.error(f"Errore principale: {e}", exc_info=debug_mode) if __name__ == "__main__": asyncio.run(main())