import requests import os import pyodbc import logging import json #from dotenv import load_dotenv from airflow.models import Variable from ftplib import FTP from datetime import datetime from kafka import KafkaProducer logging.basicConfig(level=logging.INFO) airflow_logger = logging.getLogger(__name__) # ========================================== # CONFIGURACIÓN # ========================================== APP_NAME = 'PYTHON_CREATE_URL-FTP_APP' VERSION = '2.0.0' PROCEDURE_ID_SUFFIX = '000301900001' FTP_BASE_URL = 'https://000oqta.rcomhost.com' FTP_BASE_PATH = '/www/imglm/2026/' #WEBHOOK_URL = os.getenv('WEBHOOK_URL_NOTIFICATION') # Reemplaza con tu URL de webhook WEBHOOK_URL = Variable.get("WEBHOOK_URL_NOTIFICATION") INFORMATIONS = [ ['glmpmi', '_02354', 'glm_pmi', '00180000'], ['glmccf', '_02357', 'glm_ccf', '00190000'], ['glmea', '_02084', 'glm_ea', '00110000'], ['glmcp', '_02059', 'glm_cp', '00070000'], ['glmcl', '_02086', 'glm_cl', '00060000'], ['is001', '_02290', 'IS001', '00010000'], ['is003', '_02301', 'IS003', '00050000'], ['is004', '_02310', 'IS004', '00040000'], ['is005', '_02323', 'IS005', '00090000'], ] # ========================================== # KAFKA LOGGER # ========================================== class KafkaLogger: def __init__(self, source,bootstrap_servers, use_ssl): """Inicializa el KafkaLogger con la configuración necesaria para conectarse a Kafka. Args: source (str): Identificador del origen de los logs, útil para filtrar en Kafka. """ self.source = source self.topic = 'executions_logs' use_ssl = use_ssl config = { 'bootstrap_servers': bootstrap_servers, 'value_serializer': lambda v: json.dumps(v, default=str).encode('utf-8'), 'acks': 1, 'retries': 3, } if use_ssl: config.update({ 'security_protocol': os.getenv('SECURITY_PROTOCOL'), 'sasl_mechanism': os.getenv('SASL_MECHANISM'), 'sasl_plain_username': os.getenv('SASL_PLAIN_USERNAME'), 'sasl_plain_password': os.getenv('SASL_PLAIN_PASSWORD'), 'ssl_cafile': os.getenv('SSL_CAFILE'), 'ssl_check_hostname': os.getenv('SSL_CHECK_HOSTNAME') == 'True', }) self.producer = KafkaProducer(**config) def log(self, database, db_id, procedure_name, status, message, executed_by, target_date, date, extra_info=None): """Envía un mensaje de log a Kafka con la información proporcionada. Args: database (str): Nombre de la base de datos relacionada con el log. db_id (str): Identificador único relacionado con la base de datos. procedure_name (str): Nombre del procedimiento o tarea que se está registrando. status (str): Estado del procedimiento (e.g., 'STARTING', 'SUCCESS', 'FAILED'). message (str): Mensaje descriptivo del evento que se está registrando. executed_by (str): Usuario o entidad que ejecutó el procedimiento. target_date (str): Fecha objetivo para la ejecución del procedimiento. date (datetime): Fecha y hora en que se registra el evento. extra_info (any, optional): Información adicional relevante para el log. Puede ser cualquier tipo de dato serializable a JSON. """ data = { "database": database, "procedure_id": str(db_id) + PROCEDURE_ID_SUFFIX, "procedure_name": procedure_name, "procedure_type": "PYTHON_APP", "status": status, "message": message, "executed_by": executed_by, "source": self.source, "extra_info": str(extra_info).replace("'", " ") if extra_info else None, "target_date": str(target_date), "date": str(date), } try: self.producer.send(self.topic, value=data).get(timeout=5) except Exception as e: print(f"[KafkaLogger] Error: {e}") def close(self): self.producer.close() # ========================================== # SQL # ========================================== def read_table_where(db_database, db_password, db_user, db_server, db_table, folder): """ Lee los registros de una tabla filtrando por folder. Args: db_database (str): Nombre de la base de datos. db_password (str): Contraseña del usuario. db_user (str): Usuario de la base de datos. db_server (str): Servidor de la base de datos. db_table (str): Nombre de la tabla. folder (str): Carpeta a filtrar. Returns: list: Lista de diccionarios con los registros. """ connection = pyodbc.connect( f'DRIVER={{ODBC Driver 18 for SQL Server}};' f'SERVER={db_server};' f'DATABASE={db_database};' f'UID={db_user};' f'PWD={{{db_password}}};' f'TrustServerCertificate=yes;' f'Encrypt=no;' ) cursor = connection.cursor() cursor.execute(f"SELECT * FROM [{db_table}] WHERE Folder LIKE ?", (f'%{folder}%',)) columns = [col[0] for col in cursor.description] rows = cursor.fetchall() cursor.close() connection.close() return [dict(zip(columns, row)) for row in rows] def insert_records(db_database, db_password, db_user, db_server, db_table, data): """ Inserta registros en una tabla de SQL Server en lotes de 1000. Args: db_database (str): Nombre de la base de datos. db_password (str): Contraseña del usuario. db_user (str): Usuario de la base de datos. db_server (str): Servidor de la base de datos. db_table (str): Nombre de la tabla. data (list): Lista de diccionarios con los datos a insertar. Returns: int: Número de registros insertados. """ if not data: return 0 connection = pyodbc.connect( f'DRIVER={{ODBC Driver 18 for SQL Server}};' f'SERVER={db_server};' f'DATABASE={db_database};' f'UID={db_user};' f'PWD={{{db_password}}};' f'TrustServerCertificate=yes;' f'Encrypt=no;' ) cursor = connection.cursor() columns = [k for k in data[0].keys() if k not in ['ID', 'ExportDate']] columns_sql = ", ".join([f'[{c}]' for c in columns]) placeholders = ", ".join(["?"] * len(columns)) query = f"INSERT INTO [{db_table}] ({columns_sql}) VALUES ({placeholders})" values = [tuple(row[c] for c in columns) for row in data] batch_size = 1000 inserted = 0 for i in range(0, len(values), batch_size): batch = values[i:i + batch_size] try: cursor.executemany(query, batch) connection.commit() inserted += len(batch) airflow_logger.info(f"Batch {i // batch_size + 1} inserted: {len(batch)} records") #print(f"Lote {i // batch_size + 1} insertado: {len(batch)} registros") except Exception as e: connection.rollback() log(f'FAILED', f'Error insertando lote {i // batch_size + 1}', extra_info=e) airflow_logger.error(f"Error insertando lote {i // batch_size + 1}: {e}") #print(f"Error en lote {i // batch_size + 1}: {e}") cursor.close() connection.close() return inserted # ========================================== # FTP # ========================================== def read_ftp(ftp_server, ftp_user, ftp_password, code_folder, identity_id, year, month): """ Lee los archivos de un servidor FTP y devuelve una lista de URLs. Args: ftp_server (str): Dirección del servidor FTP. ftp_user (str): Nombre de usuario para la conexión FTP. ftp_password (str): Contraseña para la conexión FTP. code_folder (str): Carpeta que contiene los archivos a leer. identity_id (str): ID de identidad para filtrar los archivos. year (str): Año para filtrar los archivos. month (str): Mes para filtrar los archivos. Returns: list: Lista de URLs de los archivos encontrados. """ try: ftp = FTP(ftp_server) ftp.encoding = 'utf-8' ftp.sendcmd('OPTS UTF8 ON') ftp.login(ftp_user, ftp_password) except Exception as e: #print(f"Conexión FTP fallida: {e}") raise path = FTP_BASE_PATH + code_folder file_names = ftp.nlst(path) urls = [FTP_BASE_URL + f.replace('/www', '') for f in file_names] results = [] for url in urls: pos_folder = url.find(code_folder) pos_date = url.find(f'_{year}-{month}') pos_id = url.find(identity_id) results.append({ 'Folder': url[pos_folder:pos_folder + len(code_folder)], 'Date': url[pos_date + 1:pos_date + 11], 'IDcodigo': url[pos_id:pos_id + 24], 'ObjectCode': url[pos_date + 12:pos_id], 'TaskName': url[pos_folder + len(code_folder) + 1:pos_date], 'URL': url, }) ftp.quit() return results # ========================================== # MAIN # ========================================== def run_save_url(**context): airflow_logger.info(f"{'='*50}") airflow_logger.info(f"▶ INICIANDO {APP_NAME} {VERSION}") airflow_logger.info(f"{'='*50}") #print(f"{'='*50}") #print(f"▶ INICIANDO {APP_NAME} {VERSION}") #print(f"{'='*50}") #kafka_logger = KafkaLogger( # source=f'{APP_NAME}_{VERSION}', # bootstrap_servers=os.getenv('BOOTSTRAP_SERVERS'), # use_ssl = os.getenv('KAFKA_USE_SSL', 'False') == 'True' # ) kafka_logger = KafkaLogger( source=f'{APP_NAME}_{VERSION}', bootstrap_servers=Variable.get("BOOTSTRAP_SERVERS_DOCKER"), use_ssl=Variable.get("KAFKA_USE_SSL", default_var="False") == "True" ) airflow_logger.info(f"✅ KafkaLogger inicializado | SSL: {os.getenv('KAFKA_USE_SSL')}") #print #INICIANDO DE VARIABLES DE ENTORNO #ftp_server = os.getenv('FTP_000oqta_SERVER') ftp_server = Variable.get("FTP_000oqta_SERVER") #ftp_user = os.getenv('FTP_000oqta_USER') ftp_user = Variable.get("FTP_000oqta_USER") #ftp_password = os.getenv('FTP_000oqta_PASSWORD') ftp_password = Variable.get("FTP_000oqta_PASSWORD") #db_user = os.getenv('DB_MSSQ_USER') db_user = Variable.get("DB_MSSQ_USER") #db_password = os.getenv('DB_MSSQ_PASSWORD') db_password = Variable.get("DB_MSSQ_PASSWORD") #db_server = os.getenv('DB_MSSQ_SERVER') db_server = Variable.get("DB_MSSQ_SERVER") db_table = 'URL_FTP' now = datetime.now() today = now.strftime('%Y-%m-%d') year_folder = now.strftime('%Y') month_folder = now.strftime('%m') #INICIANDO LOGICA DEL PROGRAMA for name_folder, identity_id, db_database, database_id in INFORMATIONS: code_folder = name_folder + month_folder procedure_name = f'Generacion de URL {db_database}' def log(status, message, extra_info=None): kafka_logger.log( database=db_database, db_id=database_id, procedure_name=procedure_name, status=status, message=message, executed_by=db_user, target_date=today, date=now, extra_info=extra_info ) def send_whatsapp(step, error): message = ( f"🚨 *ALERTA PIPELINE*\n" f"━━━━━━━━━━━━━━━\n" f"📦 *BD:* {db_database}\n" f"⚙️ *Paso:* {step}\n" f"❌ *Error:* {str(error)[:200]}\n" f"🕐 *Hora:* {datetime.now()}\n" f"━━━━━━━━━━━━━━━\n" f"📌 Revisar logs en Airflow" ) requests.post(WEBHOOK_URL, json={"message": message}) # INICIO log('STARTING', f'Inicializando Generacion de URL de {db_database}') airflow_logger.info(f"{'─'*50}") airflow_logger.info(f"▶ START | {db_database} | folder: {code_folder}") #print(f"{'─'*50}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ▶ START | {db_database} | folder: {code_folder}") # LEER FTP try: ftp_data = read_ftp(ftp_server, ftp_user, ftp_password, code_folder, identity_id, year_folder, month_folder) log('SUCCESS', f'Lectura FTP folder {code_folder} exitosa') airflow_logger.info(f"✅ FTP | {db_database} | {len(ftp_data)} archivos en {code_folder}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ FTP | {db_database} | {len(ftp_data)} archivos encontrados en {code_folder}") except Exception as e: log('FAILED', f'Error leyendo FTP folder {code_folder}', extra_info=e) send_whatsapp(f'Error leyendo FTP folder {code_folder} para {db_database}', e) airflow_logger.error(f"❌ FTP | {db_database} | {e}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ FTP | {db_database} | {e}") continue # LEER SQL try: sql_data = read_table_where(db_database, db_password, db_user, db_server, db_table, code_folder) saved_ids = {r['IDcodigo'] for r in sql_data} log('SUCCESS', f'Lectura URLs de {db_database} exitosa') airflow_logger.info(f"✅ SQL | {db_database} | {len(saved_ids)} registros existentes") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ SQL | {db_database} | {len(saved_ids)} registros existentes") except Exception as e: log('FAILED', f'Error leyendo URLs de {db_database}', extra_info=e) send_whatsapp(f'Error leyendo URLs de {db_database} para comparar con FTP.', e) airflow_logger.error(f"❌ SQL | {db_database} | {e}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ SQL | {db_database} | {e}") continue # INSERTAR try: to_insert = [r for r in ftp_data if r['IDcodigo'] not in saved_ids] info_insert = f'{db_database} → FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | A insertar: {len(to_insert)}' airflow_logger.info(f"ℹ️ DIFF | {db_database} | FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | A insertar: {len(to_insert)}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ℹ️ DIFF | {db_database} | FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | A insertar: {len(to_insert)}") if to_insert: inserted = insert_records(db_database, db_password, db_user, db_server, db_table, to_insert) log('SUCCESS', f'Insercion en {db_database} exitosa', extra_info=f'| FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | Insertados: {len(to_insert)}') airflow_logger.info(f"✅ INSERT | {db_database} | FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | Insertados: {len(to_insert)}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ✅ INSERT | {db_database} | FTP: {len(ftp_data)} | SQL: {len(saved_ids)} | Insertados: {len(to_insert)}") else: log('SUCCESS', f'No hay registros nuevos para insertar en {db_database}') airflow_logger.warning(f"⚠️ INSERT | {db_database} | Sin registros nuevos para insertar") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ⚠️ INSERT | {db_database} | Sin registros nuevos para insertar") except Exception as e: log('FAILED', f'Error insertando en {db_database}', extra_info=e) send_whatsapp(f'Error insertando URLs en {db_database}.', e) airflow_logger.error(f"❌ INSERT | {db_database} | {e}") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ❌ INSERT | {db_database} | {e}") continue # FIN log('FINISHED', f'Finalizando Generacion de URL de {db_database}') airflow_logger.info(f"⏹ END | {db_database} | proceso completado") #print(f"[{datetime.now().strftime('%H:%M:%S')}] ⏹ END | {db_database} | proceso completado") #FINALIZACION DEL PROGRAMA - CIERRE DE LOGGER Y MENSAJES FINALES kafka_logger.close() airflow_logger.info(f"{'='*50}") airflow_logger.info(f"⏹ FINALIZADO {APP_NAME} {VERSION}") airflow_logger.info(f"{'='*50}") #print(f"{'='*50}") #print(f"⏹ FINALIZADO {APP_NAME} {VERSION}") #print(f"{'='*50}") #run_save_url()