# from django.core.management.base import BaseCommand # from django.db import connections, transaction # from incompany.models import * # from django.conf import settings # from django.utils import timezone # import pytz # from django.apps import apps # import pymysql # import logging # from logging.handlers import RotatingFileHandler # logger = logging.getLogger(__name__) # logger.setLevel(logging.INFO) # # Define the file path for the log file # log_file = f'{settings.BASE_DIR}/logs/sync_data_incompany.log' # # Create a file handler and set its properties # file_handler = RotatingFileHandler(log_file, maxBytes=1024*1024, backupCount=5) # file_handler.setLevel(logging.INFO) # # Define the log format # formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') # file_handler.setFormatter(formatter) # # Add the file handler to the logger # logger.addHandler(file_handler) # class Command(BaseCommand): # help = 'Synchronizes opportunities data from the source database' # missing_value = "Não Informado" # def handle(self, *args, **options): # # Assuming 'default' is your target database and 'source_db' is your source database alias in settings.py # self.stdout.write(self.style.SUCCESS('Starting synchronization process...')) # try: # import time # start_time = time.time() # # Step 1: Create Tables if they do not exist # self.create_tables() # # Step 2: Truncate the tables: # # - bi_acessos # # - bi_acoes_funil # # - bi_atividades # # - bi_oportunidades # # - bi_propostas # # - bi_vendas # self.truncate_table() # # Step 2: Fetch data from the source_db # self.sync_data() # # Step 3: Create Star Schema # self.create_star_schema() # # Step 4: Refresh Tables # self.refresh_tables() # end_time = time.time() # elapsed_time = end_time - start_time # self.stdout.write(self.style.SUCCESS(f'Synchronization process completed successfully in {elapsed_time} seconds.')) # logger.info(f'Synchronization process completed successfully in {elapsed_time} seconds.') # except Exception as e: # self.stdout.write(self.style.ERROR(f'Synchronization process failed: {e}')) # logger.error(f'Synchronization process failed: {e}') # def create_tables(self): # self.stdout.write(self.style.NOTICE('Creating tables by calling stored procedure...')) # logger.info('Creating tables by calling stored procedure...') # try: # with connections['db_incompany'].cursor() as cursor: # cursor.execute("CALL CreateTablesToSync()") # self.stdout.write(self.style.SUCCESS('Tables created successfully via stored procedure.')) # logger.info('Tables created successfully via stored procedure.') # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error creating tables: {e}')) # logger.error(f'Error creating tables: {e}') # raise e # def truncate_table(self): # with transaction.atomic(): # with connections['db_incompany'].cursor() as cursor: # tables_to_truncate = [ # 'bi_acessos', # 'bi_acoes_funil', # 'bi_atividades', # 'bi_oportunidades', # 'bi_propostas', # 'bi_vendas', # 'bi_utms', # ] # self.stdout.write(self.style.NOTICE('Truncating tables...')) # logger.info('Truncating tables...') # try: # with connections['db_incompany'].cursor() as cursor: # for table in tables_to_truncate: # try: # self.stdout.write(self.style.NOTICE(f'Truncating table {table}...')) # logger.info(f'Truncating table {table}...') # cursor.execute(f"TRUNCATE TABLE {table}") # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error truncating table {table}: {e}')) # logger.error(f'Error truncating table {table}: {e}') # raise e # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error truncating tables: {e}')) # logger.error(f'Error truncating tables: {e}') # raise e # self.stdout.write(self.style.SUCCESS('Truncation completed.')) # logger.info('Truncation completed.') # def sync_data(self): # self.stdout.write(self.style.NOTICE('Fetching data from source database...')) # logger.info('Fetching data from source database...') # # Choose the correct connection based on DEBUG setting # connection_name = 'source_db_development' if settings.DEBUG else 'source_db_production' # source_connection = connections[connection_name] # # Create a dictionary to map source tables to target tables # table_mapping = { # 'vw_pixcrm_incompany_acessos': 'bi_acessos', # 'vw_pixcrm_incompany_acoes_funil': 'bi_acoes_funil', # 'vw_pixcrm_incompany_atividades': 'bi_atividades', # 'vw_pixcrm_incompany_oportunidades': 'bi_oportunidades', # 'vw_pixcrm_incompany_propostas': 'bi_propostas', # 'vw_pixcrm_incompany_vendas': 'bi_vendas', # 'vw_pixcrm_incompany_utms': 'bi_utms' # } # try: # with source_connection.cursor() as source_cursor: # for source_table, target_table in table_mapping.items(): # source_cursor.execute(f"SELECT * FROM {source_table}") # columns = [col[0] for col in source_cursor.description] # rows_to_create = [] # for row in source_cursor.fetchall(): # processed_row = { # col: ("Não Informado" if (value is None or value == "") and isinstance(value, str) else # 0 if (value is None or value == "") and isinstance(value, (int, float)) else # value) # for col, value in zip(columns, row) # } # for field, value in processed_row.items(): # if (value is None or value == "" or value == "Não informada") and isinstance(value, str): # processed_row[field] = self.missing_value # rows_to_create.append(processed_row) # with connections['db_incompany'].cursor() as target_cursor: # for row in rows_to_create: # placeholders = ', '.join(['%s'] * len(row)) # columns = ', '.join(row.keys()) # values = tuple(row.values()) # target_cursor.execute(f"INSERT INTO {target_table} ({columns}) VALUES ({placeholders})", values) # self.stdout.write(self.style.SUCCESS('Data fetched and processed. Records inserted into target tables.')) # logger.info('Data fetched and processed. Records inserted into target tables.') # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error fetching and processing data: {e}')) # logger.error(f'Error fetching and processing data: {e}') # raise e # def create_star_schema(self): # """ # Creates the star schema for the BI SimmGroup database, using a procedure to create the views. # """ # self.stdout.write(self.style.NOTICE('Creating star schema...')) # logger.info('Creating star schema...') # try: # with connections['db_incompany'].cursor() as cursor: # cursor.execute("CALL CreateStarSchema()") # self.stdout.write(self.style.SUCCESS('Star schema created successfully.')) # logger.info('Star schema created successfully.') # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error creating star schema: {e}')) # logger.error(f'Error creating star schema: {e}') # raise e # def refresh_tables(self): # self.stdout.write(self.style.NOTICE('Refreshing tables (batch mode)...')) # logger.info('Refreshing tables (batch mode)...') # # List of (table, view) pairs # table_view_pairs = [ # ('dCanais', 'vw_dCanais'), # ('dCategorias', 'vw_dCategorias'), # ('dConsultores', 'vw_dConsultores'), # ('dEmpreendimentos', 'vw_dEmpreendimentos'), # ('dEstadosCivis', 'vw_dEstadosCivis'), # ('dEtapasFunil', 'vw_dEtapasFunil'), # ('dFaixasIdade', 'vw_dFaixasIdade'), # ('dFechamentoDias', 'vw_dFechamentoDias'), # ('dInteresses', 'vw_dInteresses'), # ('dMidias', 'vw_dMidias'), # ('dMotivosPerdaPausa', 'vw_dMotivosPerdaPausa'), # ('dProfissoes', 'vw_dProfissoes'), # ('dQuartos', 'vw_dQuartos'), # ('dRegioes', 'vw_dRegioes'), # ('dRendas', 'vw_dRendas'), # ('dSexos', 'vw_dSexos'), # ('dStatus', 'vw_dStatus'), # ('dStatusAtividade', 'vw_dStatusAtividade'), # ('dStatusProposta', 'vw_dStatusProposta'), # ('dSuites', 'vw_dSuites'), # ('dTemperaturas', 'vw_dTemperaturas'), # ('dTipos', 'vw_dTipos'), # ('dTiposAtividade', 'vw_dTiposAtividade'), # ('dUTMSource', 'vw_dUTMSource'), # ('dUTMMedium', 'vw_dUTMMedium'), # ('dUTMCampaign', 'vw_dUTMCampaign'), # ('dUTMContent', 'vw_dUTMContent'), # ('dUTMTerm', 'vw_dUTMTerm'), # #('dTiposProposta', 'vw_dTiposProposta'), # ('dUnidades', 'vw_dUnidades'), # ('dUnidadesEmpreendimento', 'vw_dUnidadesEmpreendimento'), # ('dVagas', 'vw_dVagas'), # ('fAcessos', 'vw_fAcessos'), # ('fOportunidades', 'vw_fOportunidades'), # ('fAcoesFunil', 'vw_fAcoesFunil'), # ('fAtividades', 'vw_fAtividades'), # ('fPropostas', 'vw_fPropostas'), # ('fVendas', 'vw_fVendas'), # ] # # Batch size (number of tables per transaction) # batch_size = 100 # try: # for i in range(0, len(table_view_pairs), batch_size): # batch = table_view_pairs[i:i+batch_size] # with transaction.atomic(): # with connections['db_incompany'].cursor() as cursor: # for table, view in batch: # try: # self.stdout.write(self.style.NOTICE(f'Dropping table {table}...')) # logger.info(f'Dropping table {table}...') # cursor.execute(f"DROP TABLE IF EXISTS `{table}`;") # self.stdout.write(self.style.NOTICE(f'Creating table {table} from {view}...')) # logger.info(f'Creating table {table} from {view}...') # cursor.execute(f"CREATE TABLE IF NOT EXISTS `{table}` SELECT * FROM `{view}`;") # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error with table {table}: {e}')) # logger.error(f'Error with table {table}: {e}') # raise e # self.stdout.write(self.style.SUCCESS(f'Batch {i//batch_size+1} completed.')) # logger.info(f'Batch {i//batch_size+1} completed.') # self.stdout.write(self.style.SUCCESS('All tables refreshed.')) # logger.info('All tables refreshed.') # except Exception as e: # self.stdout.write(self.style.ERROR(f'Error refreshing tables: {e}')) # logger.error(f'Error refreshing tables: {e}') # raise e #incompany/management/commands/sync_data_incompany.py from django.core.management.base import BaseCommand from django.db import connections, transaction, close_old_connections from django.conf import settings import time import logging from logging.handlers import RotatingFileHandler # --- Configuração do Logger (sem alterações) --- logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) log_file = f'{settings.BASE_DIR}/logs/sync_data_incompany.log' file_handler = RotatingFileHandler(log_file, maxBytes=1024*1024, backupCount=5) file_handler.setLevel(logging.INFO) formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') file_handler.setFormatter(formatter) logger.addHandler(file_handler) class Command(BaseCommand): help = 'Sincroniza dados do banco de dados de origem para o de destino de forma otimizada.' missing_value = "Não Informado" def handle(self, *args, **options): """ Ponto de entrada principal para o comando de gerenciamento. Executa o processo de ETL passo a passo. """ close_old_connections() self.stdout.write(self.style.SUCCESS('Iniciando processo de sincronização...')) start_time = time.time() try: self.create_tables() self.truncate_tables() self.sync_data_in_batches() self.create_star_schema() self.refresh_tables_in_batches() # Renomeado para refletir a nova lógica elapsed_time = time.time() - start_time self.stdout.write(self.style.SUCCESS(f'Sincronização concluída com sucesso em {elapsed_time:.2f} segundos.')) logger.info(f'Sincronização concluída com sucesso em {elapsed_time:.2f} segundos.') except Exception as e: self.stdout.write(self.style.ERROR(f'O processo de sincronização falhou: {e}')) logger.error(f'O processo de sincronização falhou: {e}', exc_info=True) def create_tables(self): """ Executa a stored procedure para criar as tabelas de BI, se não existirem. """ self.stdout.write(self.style.NOTICE('Verificando/Criando tabelas de BI via Stored Procedure...')) logger.info('Verificando/Criando tabelas de BI via Stored Procedure...') try: with connections['db_incompany'].cursor() as cursor: cursor.execute("CALL CreateTablesToSync()") self.stdout.write(self.style.SUCCESS('Tabelas de BI criadas/verificadas com sucesso.')) logger.info('Tabelas de BI criadas/verificadas com sucesso.') except Exception as e: self.stdout.write(self.style.ERROR(f'Erro ao criar tabelas de BI: {e}')) logger.error(f'Erro ao criar tabelas de BI: {e}') raise e def truncate_tables(self): """ Limpa (TRUNCATE) todas as tabelas de BI para receber os novos dados. """ tables_to_truncate = [ 'bi_acessos', 'bi_acoes_funil', 'bi_atividades', 'bi_oportunidades', 'bi_propostas', 'bi_vendas', 'bi_utms', ] self.stdout.write(self.style.NOTICE('Limpando tabelas de BI...')) logger.info('Limpando tabelas de BI...') try: with transaction.atomic(using='db_incompany'): with connections['db_incompany'].cursor() as cursor: for table in tables_to_truncate: self.stdout.write(f'Limpando tabela {table}...') logger.info(f'Limpando tabela {table}...') cursor.execute(f"TRUNCATE TABLE `{table}`") self.stdout.write(self.style.SUCCESS('Tabelas de BI limpas com sucesso.')) logger.info('Tabelas de BI limpas com sucesso.') except Exception as e: self.stdout.write(self.style.ERROR(f'Erro ao limpar tabelas de BI: {e}')) logger.error(f'Erro ao limpar tabelas de BI: {e}') raise e def sync_data_in_batches(self): """ Busca dados da origem e insere no destino em lotes. Verifica o tipo de dado de cada coluna para aplicar o valor padrão correto. """ self.stdout.write(self.style.NOTICE('Iniciando sincronização de dados em lotes...')) logger.info('Iniciando sincronização de dados em lotes...') connection_name = 'source_db_development' if settings.DEBUG else 'source_db_production' source_connection = connections[connection_name] table_mapping = { 'vw_pixcrm_incompany_acessos': 'bi_acessos', 'vw_pixcrm_incompany_acoes_funil': 'bi_acoes_funil', 'vw_pixcrm_incompany_atividades': 'bi_atividades', 'vw_pixcrm_incompany_oportunidades': 'bi_oportunidades', 'vw_pixcrm_incompany_propostas': 'bi_propostas', 'vw_pixcrm_incompany_vendas': 'bi_vendas', 'vw_pixcrm_incompany_utms': 'bi_utms' } batch_size = 1000 try: column_types = {} with connections['db_incompany'].cursor() as target_cursor: db_name = connections['db_incompany'].settings_dict['NAME'] for target_table in table_mapping.values(): target_cursor.execute(""" SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s """, [db_name, target_table]) column_types[target_table] = {row[0]: row[1] for row in target_cursor.fetchall()} with source_connection.cursor() as source_cursor: for source_view, target_table in table_mapping.items(): self.stdout.write(self.style.NOTICE(f'Processando {source_view} -> {target_table}')) logger.info(f'Processando {source_view} -> {target_table}') source_cursor.execute(f"SELECT * FROM {source_view}") columns = [col[0] for col in source_cursor.description] target_table_types = column_types.get(target_table, {}) total_rows_inserted = 0 while True: rows = source_cursor.fetchmany(batch_size) if not rows: break rows_to_insert = [] for row in rows: processed_row = [] for col_name, value in zip(columns, row): if value is None or value == "" or str(value).strip() == "Não informada": data_type = target_table_types.get(col_name, 'varchar').lower() if any(date_type in data_type for date_type in ['date', 'datetime', 'timestamp']): processed_row.append(None) elif any(numeric_type in data_type for numeric_type in ['int', 'decimal', 'double', 'float', 'numeric', 'bit']): processed_row.append(0) else: processed_row.append(self.missing_value) else: processed_row.append(value) rows_to_insert.append(tuple(processed_row)) with transaction.atomic(using='db_incompany'): with connections['db_incompany'].cursor() as target_cursor: placeholders = ', '.join(['%s'] * len(columns)) insert_query = f"INSERT INTO `{target_table}` ({', '.join(f'`{c}`' for c in columns)}) VALUES ({placeholders})" target_cursor.executemany(insert_query, rows_to_insert) total_rows_inserted += len(rows) self.stdout.write(f' -> {total_rows_inserted} registros inseridos em {target_table}...') self.stdout.write(self.style.SUCCESS('Sincronização de dados em lote concluída.')) logger.info('Sincronização de dados em lote concluída.') except Exception as e: self.stdout.write(self.style.ERROR(f'Erro durante a sincronização de dados: {e}')) logger.error(f'Erro durante a sincronização de dados: {e}') raise e def create_star_schema(self): """ Executa a stored procedure para criar as views do Star Schema. """ self.stdout.write(self.style.NOTICE('Criando/Atualizando views do Star Schema...')) logger.info('Criando/Atualizando views do Star Schema...') try: with connections['db_incompany'].cursor() as cursor: cursor.execute("CALL CreateStarSchema()") self.stdout.write(self.style.SUCCESS('Views do Star Schema criadas/atualizadas com sucesso.')) logger.info('Views do Star Schema criadas/atualizadas com sucesso.') except Exception as e: self.stdout.write(self.style.ERROR(f'Erro ao criar Star Schema: {e}')) logger.error(f'Erro ao criar Star Schema: {e}') raise e def refresh_tables_in_batches(self): """ Materializa as views em tabelas físicas em lotes para evitar timeouts. """ self.stdout.write(self.style.NOTICE('Materializando views em tabelas (em lotes)...')) logger.info('Materializando views em tabelas (em lotes)...') table_view_pairs = [ ('dCanais', 'vw_dCanais'), ('dCategorias', 'vw_dCategorias'), ('dConsultores', 'vw_dConsultores'), ('dEmpreendimentos', 'vw_dEmpreendimentos'), ('dEstadosCivis', 'vw_dEstadosCivis'), ('dEtapasFunil', 'vw_dEtapasFunil'), ('dFaixasIdade', 'vw_dFaixasIdade'), ('dFechamentoDias', 'vw_dFechamentoDias'), ('dInteresses', 'vw_dInteresses'), ('dMidias', 'vw_dMidias'), ('dMotivosPerdaPausa', 'vw_dMotivosPerdaPausa'), ('dProfissoes', 'vw_dProfissoes'), ('dQuartos', 'vw_dQuartos'), ('dRegioes', 'vw_dRegioes'), ('dRendas', 'vw_dRendas'), ('dSexos', 'vw_dSexos'), ('dStatus', 'vw_dStatus'), ('dStatusAtividade', 'vw_dStatusAtividade'), ('dStatusProposta', 'vw_dStatusProposta'), ('dSuites', 'vw_dSuites'), ('dTemperaturas', 'vw_dTemperaturas'), ('dTipos', 'vw_dTipos'), ('dTiposAtividade', 'vw_dTiposAtividade'), ('dUTMSource', 'vw_dUTMSource'), ('dUTMMedium', 'vw_dUTMMedium'), ('dUTMCampaign', 'vw_dUTMCampaign'), ('dUTMContent', 'vw_dUTMContent'), ('dUTMTerm', 'vw_dUTMTerm'), ('dUnidades', 'vw_dUnidades'), ('dUnidadesEmpreendimento', 'vw_dUnidadesEmpreendimento'), ('dVagas', 'vw_dVagas'), ('fAcessos', 'vw_fAcessos'), ('fOportunidades', 'vw_fOportunidades'), ('fAcoesFunil', 'vw_fAcoesFunil'), ('fAtividades', 'vw_fAtividades'), ('fPropostas', 'vw_fPropostas'), ('fVendas', 'vw_fVendas'), ] db_connection = connections['db_incompany'] batch_size = 5000 # Lotes maiores pois é uma operação interna no mesmo DB try: for table, view in table_view_pairs: self.stdout.write(f'Atualizando tabela `{table}` a partir da view `{view}`...') logger.info(f'Atualizando tabela `{table}` a partir da view `{view}`...') temp_table = f"{table}_new" # 1. Cria uma tabela temporária vazia com a estrutura da view with db_connection.cursor() as cursor: cursor.execute(f"DROP TABLE IF EXISTS `{temp_table}`;") cursor.execute(f"CREATE TABLE `{temp_table}` SELECT * FROM `{view}` LIMIT 0;") # 2. Copia os dados da view para a tabela temporária em lotes with db_connection.cursor() as read_cursor: read_cursor.execute(f"SELECT * FROM `{view}`") columns = [col[0] for col in read_cursor.description] total_rows_inserted = 0 while True: rows = read_cursor.fetchmany(batch_size) if not rows: break # Insere o lote na tabela temporária with transaction.atomic(using='db_incompany'): with db_connection.cursor() as write_cursor: placeholders = ', '.join(['%s'] * len(columns)) insert_query = f"INSERT INTO `{temp_table}` ({', '.join(f'`{c}`' for c in columns)}) VALUES ({placeholders})" write_cursor.executemany(insert_query, rows) total_rows_inserted += len(rows) self.stdout.write(f' -> {total_rows_inserted} registros materializados para {temp_table}...') # 3. Substitui a tabela antiga pela nova de forma atômica with db_connection.cursor() as cursor: cursor.execute(f"DROP TABLE IF EXISTS `{table}`;") cursor.execute(f"RENAME TABLE `{temp_table}` TO `{table}`;") self.stdout.write(self.style.SUCCESS(f'Tabela `{table}` atualizada com sucesso.')) logger.info(f'Tabela `{table}` atualizada com sucesso.') self.stdout.write(self.style.SUCCESS('Todas as tabelas foram atualizadas com sucesso.')) logger.info('Todas as tabelas foram atualizadas com sucesso.') except Exception as e: self.stdout.write(self.style.ERROR(f'Erro ao atualizar tabelas materializadas: {e}')) logger.error(f'Erro ao atualizar tabelas materializadas: {e}') # Tenta limpar a tabela temporária em caso de erro try: with db_connection.cursor() as cursor: cursor.execute(f"DROP TABLE IF EXISTS `{temp_table}`;") except Exception as cleanup_e: logger.error(f"Falha ao limpar tabela temporária `{temp_table}`: {cleanup_e}") raise e