from django.core.management.base import BaseCommand from django.db import connections, transaction from latina.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_latina.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_latina'].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_latina'].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_latina'].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_latina_acessos': 'bi_acessos', 'vw_pixcrm_latina_acoes_funil': 'bi_acoes_funil', 'vw_pixcrm_latina_atividades': 'bi_atividades', 'vw_pixcrm_latina_oportunidades': 'bi_oportunidades', 'vw_pixcrm_latina_propostas': 'bi_propostas', 'vw_pixcrm_latina_vendas': 'bi_vendas', 'vw_pixcrm_latina_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_latina'].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_latina'].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_latina'].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