Antigravity Migration Agent
415 lines
17 KiB
Python
415 lines
17 KiB
Python
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() |