ASE/src/utils/general.py

import glob
import logging
import os
from itertools import chain, cycle

logger = logging.getLogger()


def alterna_valori(*valori: any, ping_pong: bool = False) -> any:
    """
    Genera una sequenza ciclica di valori, con opzione per una sequenza "ping-pong".

    Args:
        *valori (any): Uno o più valori da ciclare.
        ping_pong (bool, optional): Se True, la sequenza sarà valori -> valori al contrario.
                                    Ad esempio, per (1, 2, 3) diventa 1, 2, 3, 2, 1, 2, 3, ...
                                    Se False, la sequenza è semplicemente ciclica.
                                    Defaults to False.

    Yields:
        any: Il prossimo valore nella sequenza ciclica.

    """
    if not valori:
        return

    if ping_pong:
        # Crea la sequenza ping-pong: valori + valori al contrario (senza ripetere primo e ultimo)
        forward = valori
        backward = valori[-2:0:-1]  # Esclude ultimo e primo elemento
        ping_pong_sequence = chain(forward, backward)
        yield from cycle(ping_pong_sequence)
    else:
        yield from cycle(valori)


async def read_error_lines_from_logs(base_path: str, pattern: str) -> tuple[list[str], list[str]]:
    """
    Reads error and warning lines from log files matching a given pattern within a base path.

    This asynchronous function searches for log files, reads their content, and categorizes
    lines starting with 'Error' as errors and all other non-empty lines as warnings.

    Args:
        base_path (str): The base directory where log files are located.
        pattern (str): The glob-style pattern to match log filenames (e.g., "*.txt", "prefix_*_output_error.txt").

    Returns:
        tuple[list[str], list[str]]: A tuple containing two lists:
                                     - The first list contains all extracted error messages.
                                     - The second list contains all extracted warning messages."""
    import aiofiles

    # Costruisce il path completo con il pattern
    search_pattern = os.path.join(base_path, pattern)

    # Trova tutti i file che corrispondono al pattern
    matching_files = glob.glob(search_pattern)

    if not matching_files:
        logger.warning(f"Nessun file trovato per il pattern: {search_pattern}")
        return [], []

    all_errors = []
    all_warnings = []

    for file_path in matching_files:
        try:
            # Use async file I/O to prevent blocking the event loop
            async with aiofiles.open(file_path, encoding="utf-8") as file:
                content = await file.read()
                lines = content.splitlines()
                # Usando dict.fromkeys() per mantenere l'ordine e togliere le righe duplicate per i warnings
                non_empty_lines = [line.strip() for line in lines if line.strip()]

                # Fix: Accumulate errors and warnings from all files instead of overwriting
                file_errors = [line for line in non_empty_lines if line.startswith("Error")]
                file_warnings = [line for line in non_empty_lines if not line.startswith("Error")]

                all_errors.extend(file_errors)
                all_warnings.extend(file_warnings)

        except Exception as e:
            logger.error(f"Errore durante la lettura del file {file_path}: {e}")

    # Remove duplicates from warnings while preserving order
    unique_warnings = list(dict.fromkeys(all_warnings))

    return all_errors, unique_warnings