""" Nó para processamento de arquivos CSV """ import os import shutil import logging import time import pandas as pd import numpy as np from typing import Dict, Any, TypedDict, List, Optional from sqlalchemy.types import DateTime, Integer, Float, String, Boolean from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import multiprocessing as mp from utils.config import UPLOADED_CSV_PATH from utils.object_manager import get_object_manager import numpy as np def analyze_numeric_column(sample_values: pd.Series) -> Dict[str, Any]: """ Análise otimizada para detectar se coluna é numérica Args: sample_values: Amostra dos valores da coluna Returns: Dicionário com análise numérica """ analysis = { "is_numeric": False, "is_integer": False, "numeric_ratio": 0.0, "has_decimals": False } if len(sample_values) == 0: return analysis # Converte para string e limpa valores str_values = sample_values.astype(str).str.strip() # Remove valores vazios e nulos clean_values = str_values[ ~str_values.isin(['', 'nan', 'null', 'none', '-', 'NaN', 'NULL']) ] if len(clean_values) == 0: return analysis # Tenta conversão numérica vetorizada try: # Substitui vírgulas por pontos para formato brasileiro numeric_values = clean_values.str.replace(',', '.', regex=False) # Tenta conversão para float converted = pd.to_numeric(numeric_values, errors='coerce') # Conta valores válidos valid_count = converted.notna().sum() total_count = len(clean_values) analysis["numeric_ratio"] = valid_count / total_count if total_count > 0 else 0 # Se mais de 80% são números válidos, considera numérico if analysis["numeric_ratio"] > 0.8: analysis["is_numeric"] = True # Verifica se são inteiros valid_numbers = converted.dropna() if len(valid_numbers) > 0: # Verifica se todos os números válidos são inteiros analysis["is_integer"] = all( float(x).is_integer() for x in valid_numbers if not pd.isna(x) and abs(x) < 1e15 # Evita overflow ) analysis["has_decimals"] = not analysis["is_integer"] except Exception as e: logging.debug(f"Erro na análise numérica: {e}") analysis["is_numeric"] = False return analysis def detect_date_format(date_string: str) -> str: """ Detecta o formato mais provável de uma string de data Args: date_string: String para analisar Returns: 'iso', 'american', 'brazilian' ou 'auto' """ date_str = str(date_string).strip() # Formato ISO (YYYY-MM-DD ou YYYY/MM/DD) if len(date_str) >= 10 and date_str[4] in ['-', '/', '.'] and date_str[7] in ['-', '/', '.']: if date_str[:4].isdigit() and int(date_str[:4]) > 1900: return 'iso' # Verifica se pode ser formato americano (MM/DD/YYYY) if '/' in date_str: parts = date_str.split('/') if len(parts) == 3: try: month, day, year = int(parts[0]), int(parts[1]), int(parts[2]) # Se o primeiro número é > 12, provavelmente é DD/MM/YYYY if month > 12: return 'brazilian' # Se o segundo número é > 12, provavelmente é MM/DD/YYYY elif day > 12: return 'american' # Se ambos <= 12, é ambíguo, assume brasileiro por padrão else: return 'brazilian' except: pass # Formato brasileiro por padrão (DD/MM/YYYY, DD-MM-YYYY, DD.MM.YYYY) return 'brazilian' def smart_date_conversion(date_string: str): """ Converte string para data usando detecção inteligente de formato Args: date_string: String da data Returns: Timestamp do pandas ou levanta exceção """ format_type = detect_date_format(date_string) if format_type == 'iso': return pd.to_datetime(date_string, errors='raise') elif format_type == 'american': return pd.to_datetime(date_string, format='%m/%d/%Y', errors='raise') elif format_type == 'brazilian': return pd.to_datetime(date_string, dayfirst=True, errors='raise') else: # Fallback para detecção automática return pd.to_datetime(date_string, errors='raise') async def process_dates_advanced(series: pd.Series) -> pd.Series: """ Processa datas com múltiplos formatos de forma robusta Args: series: Série pandas com datas em formato texto Returns: Série com datas convertidas para datetime """ # Formatos de data para tentar em ordem de prioridade date_formats = [ '%d/%m/%Y', # 01/12/2024 '%d-%m-%Y', # 01-12-2024 '%Y-%m-%d', # 2024-12-01 '%d/%m/%y', # 01/12/24 '%d-%m-%y', # 01-12-24 '%Y/%m/%d', # 2024/12/01 '%d.%m.%Y', # 01.12.2024 '%Y.%m.%d', # 2024.12.01 '%d/%m/%Y %H:%M:%S', # 01/12/2024 14:30:00 '%Y-%m-%d %H:%M:%S', # 2024-12-01 14:30:00 ] result_series = pd.Series(index=series.index, dtype='datetime64[ns]') for idx, value in series.items(): if pd.isna(value) or str(value).strip() in ['', 'nan', 'null', 'none', '-']: result_series[idx] = pd.NaT continue value_str = str(value).strip() converted = False # Tenta conversão automática com detecção inteligente de formato try: result_series[idx] = smart_date_conversion(value_str) converted = True except: pass # Se não funcionou, tenta formatos específicos if not converted: for fmt in date_formats: try: result_series[idx] = pd.to_datetime(value_str, format=fmt, errors='raise') converted = True break except: continue # Se ainda não converteu, marca como NaT if not converted: result_series[idx] = pd.NaT logging.warning(f"Não foi possível converter '{value_str}' para data") return result_series class CSVProcessingState(TypedDict): """Estado para processamento de CSV""" file_path: str success: bool message: str csv_data_sample: dict column_info: dict processing_stats: dict async def detect_column_types(df: pd.DataFrame, sample_size: int = 1000) -> Dict[str, Any]: """ Detecta automaticamente os tipos de colunas de forma genérica e otimizada Args: df: DataFrame do pandas sample_size: Número de linhas para amostragem (otimização) Returns: Dicionário com informações dos tipos detectados """ column_info = { "detected_types": {}, "sql_types": {}, "date_columns": [], "numeric_columns": [], "text_columns": [], "processing_rules": {} } # Usa amostra para otimizar performance em datasets grandes sample_df = df.sample(n=min(sample_size, len(df)), random_state=42) if len(df) > sample_size else df logging.info(f"[OPTIMIZATION] Usando amostra de {len(sample_df)} linhas para detecção de tipos") for col in df.columns: # Detecta tipo original original_type = str(df[col].dtype) column_info["detected_types"][col] = original_type # Usa amostra para análise sample_col = sample_df[col] if col in sample_df.columns else df[col] # Detecta números já convertidos pelo pandas if sample_col.dtype in ['int64', 'Int64', 'float64', 'Float64']: if 'int' in str(sample_col.dtype).lower(): column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Integer() column_info["processing_rules"][col] = "keep_as_int" else: column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Float() column_info["processing_rules"][col] = "keep_as_float" continue # Tenta detectar datas de forma mais robusta if sample_col.dtype == 'object': # Tenta detectar datas com múltiplos formatos sample_values = sample_col.dropna().head(20) date_success_count = 0 # Formatos de data comuns para testar date_formats = [ '%d/%m/%Y', # 01/12/2024 '%d-%m-%Y', # 01-12-2024 '%Y-%m-%d', # 2024-12-01 '%d/%m/%y', # 01/12/24 '%d-%m-%y', # 01-12-24 '%Y/%m/%d', # 2024/12/01 '%d.%m.%Y', # 01.12.2024 '%Y.%m.%d', # 2024.12.01 ] for val in sample_values: val_str = str(val).strip() if not val_str or val_str.lower() in ['nan', 'null', 'none', '-']: continue # Tenta conversão automática com detecção inteligente try: smart_date_conversion(val_str) date_success_count += 1 continue except: pass # Tenta formatos específicos for fmt in date_formats: try: pd.to_datetime(val_str, format=fmt, errors='raise') date_success_count += 1 break except: continue # Se mais de 70% dos valores são datas válidas, considera como coluna de data if len(sample_values) > 0 and date_success_count / len(sample_values) > 0.7: column_info["date_columns"].append(col) column_info["sql_types"][col] = DateTime() column_info["processing_rules"][col] = "parse_dates_advanced" continue # Tenta detectar números em colunas de texto (otimizado) elif sample_col.dtype == 'object': # Análise otimizada de números em texto sample_values = sample_col.dropna().head(50) # Aumenta amostra para melhor precisão if len(sample_values) == 0: column_info["text_columns"].append(col) column_info["sql_types"][col] = String() column_info["processing_rules"][col] = "keep_as_text" continue # Análise vetorizada para performance numeric_analysis = analyze_numeric_column(sample_values) if numeric_analysis["is_numeric"]: if numeric_analysis["is_integer"]: column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Integer() column_info["processing_rules"][col] = "convert_text_to_int_safe" else: column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Float() column_info["processing_rules"][col] = "convert_text_to_float_safe" else: # Mantém como texto column_info["text_columns"].append(col) column_info["sql_types"][col] = String() column_info["processing_rules"][col] = "keep_as_text" # Tenta detectar números em colunas de texto elif df[col].dtype == 'object': # Verifica se pode ser convertido para número sample_values = df[col].dropna().head(20) numeric_count = 0 for val in sample_values: try: # Remove caracteres comuns e tenta converter clean_val = str(val).replace(',', '.').replace('-', '').strip() if clean_val: float(clean_val) numeric_count += 1 except: pass # Se mais de 70% são números, trata como numérico if len(sample_values) > 0 and numeric_count / len(sample_values) > 0.7: # Verifica se são inteiros ou floats has_decimal = any('.' in str(val) or ',' in str(val) for val in sample_values) if has_decimal: column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Float() column_info["processing_rules"][col] = "convert_text_to_float" else: column_info["numeric_columns"].append(col) column_info["sql_types"][col] = Integer() column_info["processing_rules"][col] = "convert_text_to_int" else: # Mantém como texto column_info["text_columns"].append(col) column_info["sql_types"][col] = String() column_info["processing_rules"][col] = "keep_as_text" else: # Outros tipos mantém como texto column_info["text_columns"].append(col) column_info["sql_types"][col] = String() column_info["processing_rules"][col] = "keep_as_text" return column_info async def process_dataframe_generic(df: pd.DataFrame, column_info: Dict[str, Any]) -> pd.DataFrame: """ Processa DataFrame com OTIMIZAÇÕES EXTREMAS para performance máxima Args: df: DataFrame original column_info: Informações dos tipos detectados Returns: DataFrame processado """ logging.info(f"[ULTRA_OPTIMIZATION] Iniciando processamento ULTRA-OTIMIZADO de {len(df)} linhas") start_time = time.time() # OTIMIZAÇÃO 1: Evita cópia desnecessária - modifica in-place quando possível processed_df = df # OTIMIZAÇÃO 2: Agrupa colunas por tipo de processamento processing_groups = { 'dates': [], 'keep_numeric': [], 'convert_numeric': [], 'text': [] } for col, rule in column_info["processing_rules"].items(): if col not in processed_df.columns: continue if 'date' in rule: processing_groups['dates'].append((col, rule)) elif 'keep_as' in rule: processing_groups['keep_numeric'].append((col, rule)) elif 'convert' in rule: processing_groups['convert_numeric'].append((col, rule)) else: processing_groups['text'].append((col, rule)) # OTIMIZAÇÃO 3: Processamento paralelo por grupos await process_groups_parallel(processed_df, processing_groups) total_time = time.time() - start_time logging.info(f"[ULTRA_OPTIMIZATION] Processamento ULTRA-OTIMIZADO concluído em {total_time:.2f}s") return processed_df async def process_groups_parallel(df: pd.DataFrame, groups: Dict[str, List]): """ Processa grupos de colunas em paralelo para máxima performance """ tasks = [] # Processa cada grupo for group_name, columns in groups.items(): if not columns: continue if group_name == 'dates': tasks.append(process_date_columns_batch(df, columns)) elif group_name == 'keep_numeric': tasks.append(process_keep_numeric_batch(df, columns)) elif group_name == 'convert_numeric': tasks.append(process_convert_numeric_batch(df, columns)) # text não precisa processamento # Executa todos os grupos em paralelo if tasks: import asyncio await asyncio.gather(*tasks) async def process_date_columns_batch(df: pd.DataFrame, date_columns: List[tuple]): """Processa colunas de data em lote""" for col, rule in date_columns: try: if rule == "parse_dates_advanced": # OTIMIZAÇÃO: Processamento vetorizado de datas df[col] = process_dates_vectorized(df[col]) else: df[col] = pd.to_datetime(df[col], dayfirst=True, errors='coerce') except Exception as e: logging.warning(f"Erro ao processar data {col}: {e}") async def process_keep_numeric_batch(df: pd.DataFrame, numeric_columns: List[tuple]): """Processa colunas numéricas que já estão no tipo correto""" for col, rule in numeric_columns: try: if rule == "keep_as_int" and df[col].dtype != 'Int64': df[col] = df[col].astype("Int64") elif rule == "keep_as_float" and df[col].dtype != 'float64': df[col] = df[col].astype("float64") except Exception as e: logging.warning(f"Erro ao manter tipo {col}: {e}") async def process_convert_numeric_batch(df: pd.DataFrame, convert_columns: List[tuple]): """Processa conversões numéricas em lote com máxima otimização""" for col, rule in convert_columns: try: if rule == "convert_text_to_int_safe": df[col] = convert_to_int_ultra_optimized(df[col]) elif rule == "convert_text_to_float_safe": df[col] = convert_to_float_ultra_optimized(df[col]) except Exception as e: logging.warning(f"Erro ao converter {col}: {e}") if rule == "parse_dates": processed_df[col] = pd.to_datetime( processed_df[col], dayfirst=True, errors='coerce' ) elif rule == "parse_dates_advanced": # Processamento avançado de datas com múltiplos formatos processed_df[col] = await process_dates_advanced(processed_df[col]) elif rule == "keep_as_int": # Já é inteiro, apenas garante tipo correto if processed_df[col].dtype != 'Int64': processed_df[col] = processed_df[col].astype("Int64") elif rule == "keep_as_float": # Já é float, apenas garante tipo correto if processed_df[col].dtype != 'float64': processed_df[col] = processed_df[col].astype("float64") elif rule == "convert_text_to_int_safe": # Conversão otimizada e segura para inteiros processed_df[col] = convert_to_int_optimized(processed_df[col]) elif rule == "convert_text_to_float_safe": # Conversão otimizada e segura para floats processed_df[col] = convert_to_float_optimized(processed_df[col]) elif rule == "keep_as_text": # Mantém como texto, apenas garante que é string processed_df[col] = processed_df[col].astype(str) except Exception as e: logging.warning(f"Erro ao processar coluna {col} com regra {rule}: {e}") # Em caso de erro, mantém coluna original continue col_time = time.time() - col_start_time logging.debug(f"[OPTIMIZATION] Coluna {col} processada em {col_time:.2f}s") total_time = time.time() - start_time logging.info(f"[OPTIMIZATION] Processamento concluído em {total_time:.2f}s") return processed_df def convert_to_int_optimized(series: pd.Series) -> pd.Series: """ Conversão otimizada para inteiros Args: series: Série para converter Returns: Série convertida para Int64 """ try: # Operações vetorizadas para performance cleaned = series.astype(str).str.strip() # Remove valores inválidos cleaned = cleaned.replace(['', 'nan', 'null', 'none', '-', 'NaN', 'NULL'], np.nan) # Substitui vírgulas por pontos cleaned = cleaned.str.replace(',', '.', regex=False) # Converte para numérico numeric = pd.to_numeric(cleaned, errors='coerce') # Verifica se pode ser convertido para inteiro sem perda # Só converte se todos os valores válidos são inteiros valid_mask = numeric.notna() if valid_mask.any(): valid_numbers = numeric[valid_mask] # Verifica se são inteiros (sem parte decimal significativa) is_integer_mask = np.abs(valid_numbers - np.round(valid_numbers)) < 1e-10 if is_integer_mask.all(): # Todos são inteiros, pode converter result = numeric.round().astype("Int64") else: # Tem decimais, mantém como float mas avisa logging.warning(f"Coluna contém decimais, mantendo como float") result = numeric.astype("Float64") else: # Nenhum valor válido result = pd.Series([pd.NA] * len(series), dtype="Int64") return result except Exception as e: logging.error(f"Erro na conversão otimizada para int: {e}") return series def convert_to_float_optimized(series: pd.Series) -> pd.Series: """ Conversão otimizada para floats Args: series: Série para converter Returns: Série convertida para float64 """ try: # Operações vetorizadas para performance cleaned = series.astype(str).str.strip() # Remove valores inválidos cleaned = cleaned.replace(['', 'nan', 'null', 'none', '-', 'NaN', 'NULL'], np.nan) # Substitui vírgulas por pontos (formato brasileiro) cleaned = cleaned.str.replace(',', '.', regex=False) # Converte para numérico result = pd.to_numeric(cleaned, errors='coerce') return result except Exception as e: logging.error(f"Erro na conversão otimizada para float: {e}") return series def convert_to_int_ultra_optimized(series: pd.Series) -> pd.Series: """ Conversão ULTRA-OTIMIZADA para inteiros usando NumPy puro """ try: # OTIMIZAÇÃO EXTREMA: Usa NumPy diretamente values = series.values # Se já é numérico, converte diretamente if pd.api.types.is_numeric_dtype(series): return pd.Series(values, dtype="Int64") # Para strings, usa operações vetorizadas do NumPy str_values = np.asarray(series.astype(str)) # Máscara para valores válidos valid_mask = ~np.isin(str_values, ['', 'nan', 'null', 'none', '-', 'NaN', 'NULL']) # Inicializa resultado result = np.full(len(series), pd.NA, dtype=object) if valid_mask.any(): valid_values = str_values[valid_mask] # Remove vírgulas e converte cleaned = np.char.replace(valid_values, ',', '.') # Conversão vetorizada try: numeric_values = pd.to_numeric(cleaned, errors='coerce') # Só converte se são realmente inteiros int_mask = np.abs(numeric_values - np.round(numeric_values)) < 1e-10 int_values = np.round(numeric_values[int_mask]).astype('Int64') # Atribui valores convertidos valid_indices = np.where(valid_mask)[0] int_indices = valid_indices[int_mask] result[int_indices] = int_values except Exception: pass return pd.Series(result, dtype="Int64") except Exception as e: logging.error(f"Erro na conversão ultra-otimizada para int: {e}") return series def convert_to_float_ultra_optimized(series: pd.Series) -> pd.Series: """ Conversão ULTRA-OTIMIZADA para floats usando NumPy puro """ try: # OTIMIZAÇÃO EXTREMA: Usa NumPy diretamente values = series.values # Se já é numérico, retorna diretamente if pd.api.types.is_numeric_dtype(series): return series.astype('float64') # Para strings, usa operações vetorizadas do NumPy str_values = np.asarray(series.astype(str)) # Máscara para valores válidos valid_mask = ~np.isin(str_values, ['', 'nan', 'null', 'none', '-', 'NaN', 'NULL']) # Inicializa resultado result = np.full(len(series), np.nan, dtype='float64') if valid_mask.any(): valid_values = str_values[valid_mask] # Remove vírgulas (formato brasileiro) cleaned = np.char.replace(valid_values, ',', '.') # Conversão vetorizada ultra-rápida numeric_values = pd.to_numeric(cleaned, errors='coerce') result[valid_mask] = numeric_values return pd.Series(result, dtype='float64') except Exception as e: logging.error(f"Erro na conversão ultra-otimizada para float: {e}") return series def process_dates_vectorized(series: pd.Series) -> pd.Series: """ Processamento vetorizado ULTRA-OTIMIZADO de datas """ try: # OTIMIZAÇÃO: Tenta conversão direta primeiro try: return pd.to_datetime(series, dayfirst=True, errors='coerce') except: pass # Se falhou, usa abordagem mais robusta mas ainda otimizada str_values = series.astype(str) # Detecta formato mais comum na amostra sample = str_values.dropna().head(100) if len(sample) > 0: first_val = sample.iloc[0] # Detecta formato baseado no primeiro valor if len(first_val) >= 10 and first_val[4] in ['-', '/']: # Formato ISO return pd.to_datetime(series, errors='coerce') else: # Formato brasileiro return pd.to_datetime(series, dayfirst=True, errors='coerce') return pd.to_datetime(series, errors='coerce') except Exception as e: logging.error(f"Erro no processamento vetorizado de datas: {e}") return series async def csv_processing_node(state: CSVProcessingState) -> CSVProcessingState: """ Nó principal para processamento de CSV Args: state: Estado do processamento CSV Returns: Estado atualizado """ try: file_path = state["file_path"] # Copia arquivo para diretório de upload shutil.copy(file_path, UPLOADED_CSV_PATH) logging.info(f"[CSV_PROCESSING] Arquivo copiado para: {UPLOADED_CSV_PATH}") # OTIMIZAÇÃO EXTREMA: Leitura de CSV ultra-otimizada separators = [';', ',', '\t', '|'] df = None used_separator = None # Detecta separador com amostra mínima for sep in separators: try: test_df = pd.read_csv(file_path, sep=sep, nrows=3, engine='c') # Engine C é mais rápido if len(test_df.columns) > 1: # OTIMIZAÇÃO: Lê com configurações de performance máxima df = pd.read_csv( file_path, sep=sep, encoding='utf-8', on_bad_lines="skip", engine='c', # Engine C para máxima performance low_memory=False, # Evita warnings de tipos mistos dtype=str # Lê tudo como string primeiro (mais rápido) ) used_separator = sep break except: continue if df is None: raise ValueError("Não foi possível detectar o formato do CSV") logging.info(f"[CSV_PROCESSING] CSV lido com separador '{used_separator}', {len(df)} linhas, {len(df.columns)} colunas") # Detecta tipos de colunas automaticamente column_info = await detect_column_types(df) # Processa DataFrame processed_df = await process_dataframe_generic(df, column_info) # Estatísticas do processamento processing_stats = { "original_rows": len(df), "processed_rows": len(processed_df), "original_columns": len(df.columns), "processed_columns": len(processed_df.columns), "separator_used": used_separator, "date_columns_detected": len(column_info["date_columns"]), "numeric_columns_detected": len(column_info["numeric_columns"]), "text_columns_detected": len(column_info["text_columns"]) } # Amostra dos dados para o estado csv_data_sample = { "head": processed_df.head(5).to_dict(), "dtypes": processed_df.dtypes.astype(str).to_dict(), "columns": list(processed_df.columns) } # Armazena DataFrame processado no gerenciador de objetos obj_manager = get_object_manager() df_id = obj_manager.store_object(processed_df, "processed_dataframe") # Atualiza estado state.update({ "success": True, "message": f"✅ CSV processado com sucesso! {processing_stats['processed_rows']} linhas, {processing_stats['processed_columns']} colunas", "csv_data_sample": csv_data_sample, "column_info": column_info, "processing_stats": processing_stats, "dataframe_id": df_id }) logging.info(f"[CSV_PROCESSING] Processamento concluído: {processing_stats}") except Exception as e: error_msg = f"❌ Erro ao processar CSV: {e}" logging.error(f"[CSV_PROCESSING] {error_msg}") state.update({ "success": False, "message": error_msg, "csv_data_sample": {}, "column_info": {}, "processing_stats": {} }) return state