import gspread from oauth2client.service_account import ServiceAccountCredentials import pymysql import subprocess import os from dotenv import load_dotenv import logging from datetime import datetime, timedelta # Configuración inicial de logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') # Cargar las variables de entorno load_dotenv() # Archivo de bloqueo para evitar ejecuciones simultáneas LOCK_FILE = 'process.lock' # Configuración de la base de datos y Google Sheets LOG_FILE = 'log-nucleos.log' CREDS_FILE = 'secretiespacios.json' SHEET_NAME = 'Leads Nucleos PlanOk' WORKSHEET_NAME = 'Sheet1' DB_CONFIG = { 'host': os.getenv('DB_HOST'), 'user': os.getenv('DB_USER'), 'password': os.getenv('DB_PASSWORD'), 'database': os.getenv('DB_NAME'), 'cursorclass': pymysql.cursors.DictCursor } def setup_logging(log_file): """Configura el logging para el proceso.""" logger = logging.getLogger() logger.setLevel(logging.DEBUG) handler = logging.FileHandler(log_file) handler.setLevel(logging.DEBUG) formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) # Si hay handlers previos, los eliminamos if logger.hasHandlers(): logger.handlers.clear() logger.addHandler(handler) def read_google_sheet(sheet_name, creds_file, worksheet_name=None): """Lee datos de una hoja de Google Sheets y devuelve una lista de registros.""" scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"] creds = ServiceAccountCredentials.from_json_keyfile_name(creds_file, scope) client = gspread.authorize(creds) if worksheet_name: sheet = client.open(sheet_name).worksheet(worksheet_name) else: sheet = client.open(sheet_name).sheet1 records = sheet.get_all_records() return records def get_existing_centralizador_ids(db_config): """Obtiene los IDs de centralizador existentes en la base de datos.""" connection = pymysql.connect(**db_config) with connection: with connection.cursor() as cursor: cursor.execute("SELECT ID_centralizador FROM Lead_entrantes") results = cursor.fetchall() existing_ids = {row['ID_centralizador'] for row in results} return existing_ids def convert_date_format(date_str): try: if not date_str: return None # Definir una lista de posibles formatos de fecha date_formats = [ '%Y-%m-%d %H:%M:%S', # Formato que estás recibiendo '%Y-%m-%d', # Formato de solo fecha '%d/%m/%Y', # Otros formatos posibles '%d-%m-%Y', '%d/%m/%Y %H:%M:%S', '%Y/%m/%d %H:%M:%S' ] for fmt in date_formats: try: date_obj = datetime.strptime(date_str, fmt) # Retornar solo la parte de la fecha en el formato deseado return date_obj.strftime('%Y-%m-%d') except ValueError: continue # Si ningún formato coincide, registrar el error logging.error(f"Error al convertir la fecha: {date_str}") return None except Exception as e: logging.error(f"Error inesperado al convertir la fecha: {date_str} - {e}") return None def convert_time_format(time_str): try: if not time_str: return None # Definir una lista de posibles formatos de hora time_formats = [ '%Y-%m-%d %H:%M:%S', # Si la cadena incluye fecha y hora '%H:%M:%S', '%H:%M' ] for fmt in time_formats: try: time_obj = datetime.strptime(time_str, fmt) # Retornar solo la parte de la hora en el formato deseado return time_obj.strftime('%H:%M:%S') except ValueError: continue logging.error(f"Error al convertir la hora: {time_str}") return None except Exception as e: logging.error(f"Error inesperado al convertir la hora: {time_str} - {e}") return None def process_records(records, filter_date): """ Procesa los registros obtenidos de Google Sheets filtrando por fecha. Solo se procesan aquellos registros cuya 'Fecha de ingreso' sea mayor o igual a filter_date. """ processed_records = [] for record in records: try: fecha_ingreso_str = record.get('Fecha de ingreso', '') fecha_ingreso = convert_date_format(fecha_ingreso_str) if not fecha_ingreso: logging.warning(f"Registro sin fecha válida: {record}") continue # Convertir la fecha a objeto date para comparar fecha_ingreso_obj = datetime.strptime(fecha_ingreso, '%Y-%m-%d').date() if fecha_ingreso_obj < filter_date: # Si el registro es anterior a la fecha de filtro, se omite continue hora_ingreso = convert_time_format(record.get('Hora de ingreso', '')) now = datetime.now().strftime('%Y-%m-%d %H:%M:%S') processed_record = { 'Fecha_ingreso': fecha_ingreso, 'Hora_ingreso': hora_ingreso, 'ID_centralizador': record.get('ID centralizador', ''), 'ID_cliente': record.get('ID cliente', ''), 'Rut_cliente': record.get('Rut cliente', ''), 'Nombre_cliente': record.get('Nombre cliente', 'Sin Nombre'), 'Apellido_cliente': record.get('Apellido cliente', ''), 'Numero_telefono_fijo_cliente': record.get('Número teléfono (fijo) cliente', ''), 'Numero_telefono_movil_cliente': record.get('Número teléfono (móvil) cliente', ''), 'Email_cliente': record.get('Email cliente', ''), 'Medio_llegada': record.get('Medio de llegada', ''), 'Via_ingreso': record.get('Vía de ingreso', ''), 'ID_proyecto': record.get('ID proyecto', ''), 'Nombre_vendedor': record.get('Nombre vendedor', ''), 'Telefono_vendedor': record.get('Teléfono vendedor', ''), 'Email_vendedor': record.get('Email vendedor', ''), 'Rut_vendedor': record.get('Rut vendedor', ''), 'Tipologia': record.get('Tipología', ''), 'Producto': record.get('Producto', ''), 'Etapa': record.get('Etapa', ''), 'Timestamp': record.get('Timestamp', ''), 'Now': now, 'Estado': record.get('Estado', 'Pendiente'), 'Proyecto': record.get('Proyecto', '') } processed_records.append(processed_record) except KeyError as e: logging.error(f"Falta la clave {e} en el registro: {record}") except Exception as e: logging.error(f"Error al procesar el registro {record}: {e}") return processed_records def insert_data_to_db(data, db_config): """Inserta los datos procesados en la base de datos.""" connection = pymysql.connect(**db_config) with connection: with connection.cursor() as cursor: sql = """ INSERT INTO Lead_entrantes ( Timestamp, Now, Fecha_ingreso, Hora_ingreso, ID_centralizador, Estado, ID_cliente, Rut_cliente, Nombre_cliente, Apellido_cliente, Numero_telefono_fijo_cliente, Numero_telefono_movil_cliente, Email_cliente, Medio_llegada, Via_ingreso, Proyecto, ID_proyecto, Etapa, Producto, Tipologia, Nombre_vendedor, Telefono_vendedor, Email_vendedor, Rut_vendedor ) VALUES ( %(Timestamp)s, %(Now)s, %(Fecha_ingreso)s, %(Hora_ingreso)s, %(ID_centralizador)s, %(Estado)s, %(ID_cliente)s, %(Rut_cliente)s, %(Nombre_cliente)s, %(Apellido_cliente)s, %(Numero_telefono_fijo_cliente)s, %(Numero_telefono_movil_cliente)s, %(Email_cliente)s, %(Medio_llegada)s, %(Via_ingreso)s, %(Proyecto)s, %(ID_proyecto)s, %(Etapa)s, %(Producto)s, %(Tipologia)s, %(Nombre_vendedor)s, %(Telefono_vendedor)s, %(Email_vendedor)s, %(Rut_vendedor)s ) """ logging.info(f"Insertando {len(data)} registros en la base de datos.") cursor.executemany(sql, data) connection.commit() def run_process(): """Ejecuta el proceso completo de lectura y actualización de la base de datos.""" setup_logging(LOG_FILE) try: # Calcular la fecha de filtro (hoy - 7 días) filter_date = datetime.today().date() - timedelta(days=7) logging.info("Leyendo desde la fecha: %s", filter_date.strftime('%Y-%m-%d')) logging.info("Iniciando la lectura de Google Sheets y la actualización de la base de datos.") # Leer datos desde Google Sheets sheet_data = read_google_sheet(SHEET_NAME, CREDS_FILE, WORKSHEET_NAME) if sheet_data: first_record = sheet_data[0] logging.info("Claves del primer registro: %s", first_record.keys()) # Procesar los registros filtrando por fecha processed_data = process_records(sheet_data, filter_date) # Obtener los IDs de centralizador existentes en la base de datos existing_ids = get_existing_centralizador_ids(DB_CONFIG) # Filtrar registros nuevos (los que no existan ya en la BD) new_records = [record for record in processed_data if record['ID_centralizador'] not in existing_ids] # Insertar nuevos registros en la base de datos if new_records: insert_data_to_db(new_records, DB_CONFIG) logging.info(f"Se han insertado {len(new_records)} registros nuevos en la base de datos.") else: logging.info("No hay registros nuevos para insertar.") logging.info("Proceso completado con éxito.") except KeyError as e: logging.exception(f"KeyError: Falta la clave {e} en el registro.") except Exception as e: logging.exception("Ha ocurrido una excepción durante el procesamiento.") # Ejecutar el script adicional después de la inserción de datos try: logging.info("Iniciando la ejecución de crear_contacto.py para procesar los datos insertados.") subprocess.run(['python', 'crear_contacto.py'], check=True) logging.info("Script crear_contacto.py ejecutado correctamente.") except subprocess.CalledProcessError as e: logging.error(f"Error al ejecutar crear_contacto.py: {e}") def main(): """Punto de entrada principal del script.""" # Comprobar y crear archivo de bloqueo if os.path.exists(LOCK_FILE): logging.info("El proceso ya se está ejecutando. Saliendo.") return # Crear el archivo de bloqueo with open(LOCK_FILE, 'w') as f: f.write('') try: run_process() finally: # Eliminar el archivo de bloqueo al finalizar if os.path.exists(LOCK_FILE): os.remove(LOCK_FILE) if __name__ == '__main__': main()