From 9f799c9398e8d025bf0846458db1c85ce5b24bdf Mon Sep 17 00:00:00 2001 From: Antigravity Migration Agent Date: Sat, 9 May 2026 14:20:17 -0400 Subject: [PATCH] Initial commit: SaveUrl project migration from Drive --- .env.example | 33 ++ .gitignore | 73 +++ README.md | 57 +++ SaveUrl.py | 415 ++++++++++++++++++ dag_save_url_ftp.py | 41 ++ documentacion/00_INDICE.md | 23 + documentacion/01_VISION_GENERAL.md | 68 +++ documentacion/02_INSTALACION_LOCAL.md | 73 +++ documentacion/03_CONFIGURACION_VARIABLES.md | 72 +++ documentacion/04_DESPLIEGUE_AIRFLOW.md | 75 ++++ documentacion/05_FLUJO_Y_ARQUITECTURA.md | 99 +++++ .../06_REFERENCIA_TECNICA_SAVEURL.md | 147 +++++++ documentacion/07_RUNBOOK_OPERACION.md | 66 +++ documentacion/08_TROUBLESHOOTING.md | 135 ++++++ documentacion/09_SEGURIDAD_Y_SECRETS.md | 58 +++ requirements.txt | 4 + 16 files changed, 1439 insertions(+) create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 README.md create mode 100644 SaveUrl.py create mode 100644 dag_save_url_ftp.py create mode 100644 documentacion/00_INDICE.md create mode 100644 documentacion/01_VISION_GENERAL.md create mode 100644 documentacion/02_INSTALACION_LOCAL.md create mode 100644 documentacion/03_CONFIGURACION_VARIABLES.md create mode 100644 documentacion/04_DESPLIEGUE_AIRFLOW.md create mode 100644 documentacion/05_FLUJO_Y_ARQUITECTURA.md create mode 100644 documentacion/06_REFERENCIA_TECNICA_SAVEURL.md create mode 100644 documentacion/07_RUNBOOK_OPERACION.md create mode 100644 documentacion/08_TROUBLESHOOTING.md create mode 100644 documentacion/09_SEGURIDAD_Y_SECRETS.md create mode 100644 requirements.txt diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..cf43a6e --- /dev/null +++ b/.env.example @@ -0,0 +1,33 @@ +# ========================================== +# CONFIGURACIÓN DE BASE DE DATOS (MS SQL Server) +# ========================================== +DB_MSSQ_SERVER=sql-server-hostname +DB_MSSQ_USER=sql-user +DB_MSSQ_PASSWORD=sql-password +DB_MSSQ_PORT=1433 + +# ========================================== +# CONFIGURACIÓN GENERAL +# ========================================== +WEBHOOK_URL_NOTIFICATION=https://webhook-url-placeholder + +# ========================================== +# CONFIGURACIÓN DE FTP +# ========================================== +FTP_BASE_URL=https://000oqta.rcomhost.com +FTP_BASE_PATH=/www/imglm/2026/ +FTP_000oqta_SERVER=000oqta.rcomhost.com +FTP_000oqta_USER=ftp-user +FTP_000oqta_PASSWORD=ftp-password + +# ========================================== +# CONFIGURACIÓN DE KAFKA +# ========================================== +KAFKA_USE_SSL=False +BOOTSTRAP_SERVERS_DOCKER=kafka-kraft:9092 +SECURITY_PROTOCOL=SASL_SSL +SASL_MECHANISM=SCRAM-SHA-256 +SASL_PLAIN_USERNAME=pipelines +SASL_PLAIN_PASSWORD=kafka-password +SSL_CAFILE=/path/to/ca-cert.pem +SSL_CHECK_HOSTNAME=False diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0513d55 --- /dev/null +++ b/.gitignore @@ -0,0 +1,73 @@ +# ========================= +# Secrets / environment files +# ========================= +.env +.env.* +!.env.example + +# Common secrets files (if they ever appear) +*.pem +*.key +*.pfx +*.p12 +*.crt +*.cer +*.der + +# ========================= +# Python +# ========================= +__pycache__/ +*.py[cod] +*$py.class + +# Virtual environments +env/ +.env/ +venv/ +.venv/ +ENV/ +env.bak/ +venv.bak/ + +# Packaging / build artifacts +build/ +dist/ +*.egg-info/ +.eggs/ +pip-wheel-metadata/ + +# Unit test / coverage +.pytest_cache/ +.coverage +coverage.xml +htmlcov/ +.mypy_cache/ +.ruff_cache/ +.tox/ +.nox/ + +# Jupyter +.ipynb_checkpoints/ + +# ========================= +# Logs / runtime files +# ========================= +*.log +logs/ +*.pid + +# ========================= +# IDEs / OS +# ========================= +.vscode/ +.idea/ +.DS_Store +Thumbs.db +Desktop.ini + +# ========================= +# Airflow (common local artifacts) +# ========================= +airflow.cfg +webserver_config.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..b5f9734 --- /dev/null +++ b/README.md @@ -0,0 +1,57 @@ +# SaveUrl (FTP → URLs → SQL Server) + +Este proyecto **lee listados de archivos desde un FTP**, **construye URLs** por archivo y **guarda nuevas URLs en SQL Server** (tabla `URL_FTP`). La ejecución genera **logs “estructurados” en Kafka** y envía **notificación por webhook** ante errores. + +La documentación completa vive en la carpeta [`documentacion/`](documentacion/). Si solo necesitas “por dónde empezar”, usa los links por rol de abajo. + +## Inicio rápido (por rol) + +- **Operación / soporte**: [`documentacion/07_RUNBOOK_OPERACION.md`](documentacion/07_RUNBOOK_OPERACION.md) y [`documentacion/08_TROUBLESHOOTING.md`](documentacion/08_TROUBLESHOOTING.md) +- **Despliegue en Airflow**: [`documentacion/04_DESPLIEGUE_AIRFLOW.md`](documentacion/04_DESPLIEGUE_AIRFLOW.md) y [`documentacion/03_CONFIGURACION_VARIABLES.md`](documentacion/03_CONFIGURACION_VARIABLES.md) +- **Desarrollo / mantenimiento**: [`documentacion/02_INSTALACION_LOCAL.md`](documentacion/02_INSTALACION_LOCAL.md), [`documentacion/05_FLUJO_Y_ARQUITECTURA.md`](documentacion/05_FLUJO_Y_ARQUITECTURA.md) y [`documentacion/06_REFERENCIA_TECNICA_SAVEURL.md`](documentacion/06_REFERENCIA_TECNICA_SAVEURL.md) +- **Seguridad**: [`documentacion/09_SEGURIDAD_Y_SECRETS.md`](documentacion/09_SEGURIDAD_Y_SECRETS.md) +- **Índice completo**: [`documentacion/00_INDICE.md`](documentacion/00_INDICE.md) + +## Cómo se ejecuta (dos caminos) + +### 1) Desarrollo local (usa `.env`) + +Para desarrollo/pruebas locales se usa `.env` para que cada desarrollador pueda **cambiar credenciales** y reproducir casos rápidamente. + +- **Plantilla**: copiar `.env.example` → `.env` y completar valores (nunca subir `.env` al repo). +- **Guía**: ver [`documentacion/02_INSTALACION_LOCAL.md`](documentacion/02_INSTALACION_LOCAL.md) y [`documentacion/03_CONFIGURACION_VARIABLES.md`](documentacion/03_CONFIGURACION_VARIABLES.md) + +Comandos típicos en Windows: + +```powershell +python -m venv env +.\env\Scripts\Activate.ps1 +python SaveUrl.py +``` + +### 2) En Airflow (usa Airflow Variables) + +En el entorno Airflow, la configuración se lee desde **Airflow Variables** (no desde `.env`). + +- **Despliegue**: ver [`documentacion/04_DESPLIEGUE_AIRFLOW.md`](documentacion/04_DESPLIEGUE_AIRFLOW.md) +- **Variables requeridas**: ver [`documentacion/03_CONFIGURACION_VARIABLES.md`](documentacion/03_CONFIGURACION_VARIABLES.md) + +> Nota: el DAG importa `SaveUrl.py` desde una ruta esperada dentro del contenedor (detalles en la guía de despliegue). + +## Componentes principales + +- `SaveUrl.py`: lógica de negocio (entrypoint `run_save_url`). +- `dag_save_url_ftp.py`: DAG que ejecuta `run_save_url` en Airflow. +- `documentacion/`: documentación de operación, despliegue y referencia técnica. + +## Configuración + +- **Lista completa de variables (local y Airflow)**: [`documentacion/03_CONFIGURACION_VARIABLES.md`](documentacion/03_CONFIGURACION_VARIABLES.md) +- **Seguridad de secretos**: [`documentacion/09_SEGURIDAD_Y_SECRETS.md`](documentacion/09_SEGURIDAD_Y_SECRETS.md) + +## Troubleshooting + +Si ves errores de conexión a FTP/SQL Server, problemas de SSL/SASL de Kafka o fallos de webhook, empieza por: + +- [`documentacion/08_TROUBLESHOOTING.md`](documentacion/08_TROUBLESHOOTING.md) +- [`documentacion/07_RUNBOOK_OPERACION.md`](documentacion/07_RUNBOOK_OPERACION.md) diff --git a/SaveUrl.py b/SaveUrl.py new file mode 100644 index 0000000..6892db7 --- /dev/null +++ b/SaveUrl.py @@ -0,0 +1,415 @@ +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() \ No newline at end of file diff --git a/dag_save_url_ftp.py b/dag_save_url_ftp.py new file mode 100644 index 0000000..6a8e8b6 --- /dev/null +++ b/dag_save_url_ftp.py @@ -0,0 +1,41 @@ +from datetime import datetime, timedelta +from airflow import DAG +from airflow.operators.python import PythonOperator +from airflow.operators.empty import EmptyOperator +import sys +sys.path.insert(0, '/opt/airflow/dags/save_url_ftp') +from SaveUrl import run_save_url + +# ========================================== +# CONFIGURACIÓN DEL DAG +# ========================================== +DAG_ID = 'save_url_ftp' +DAG_DESCRIPTION = 'Genera y guarda URLs del FTP en SQL Server para todas las BDs' +DAG_SCHEDULE = '0 */2 * * *' # Cada 2 horas El formato cron es minuto hora día mes día_semana +TAGS = ['ftp', 'url', 'sql', 'kafka', 'notification', 'glmpmi', 'glmccf', 'glmea', 'glmcp', 'glmcl', 'is001', 'is003', 'is004', 'is005'] + +default_args = { + 'start_date': datetime(2026, 4, 1), + 'retries': 2, + 'retry_delay': timedelta(minutes=5), +} + +with DAG( + dag_id=DAG_ID, + description=DAG_DESCRIPTION, + schedule_interval=DAG_SCHEDULE, + default_args=default_args, + catchup=False, + max_active_runs=1, + tags=TAGS, +) as dag: + + start = EmptyOperator(task_id='inicio') + end = EmptyOperator(task_id='fin') + + save_urls = PythonOperator( + task_id='save_url_ftp', + python_callable=run_save_url, + ) + + start >> save_urls >> end \ No newline at end of file diff --git a/documentacion/00_INDICE.md b/documentacion/00_INDICE.md new file mode 100644 index 0000000..8103a56 --- /dev/null +++ b/documentacion/00_INDICE.md @@ -0,0 +1,23 @@ +# Documentación del proyecto `SaveUrl` + +Esta carpeta contiene la documentación completa del proyecto (sin depender del `README.md`). + +## Lectura rápida (por rol) + +- **Operación / soporte**: empieza por `07_RUNBOOK_OPERACION.md` y `08_TROUBLESHOOTING.md`. +- **Despliegue en Airflow**: `04_DESPLIEGUE_AIRFLOW.md` + `03_CONFIGURACION_VARIABLES.md`. +- **Desarrollo / mantenimiento**: `05_FLUJO_Y_ARQUITECTURA.md` + `06_REFERENCIA_TECNICA_SAVEURL.md`. +- **Seguridad**: `09_SEGURIDAD_Y_SECRETS.md`. + +## Documentos + +1. `01_VISION_GENERAL.md` +2. `02_INSTALACION_LOCAL.md` +3. `03_CONFIGURACION_VARIABLES.md` +4. `04_DESPLIEGUE_AIRFLOW.md` +5. `05_FLUJO_Y_ARQUITECTURA.md` +6. `06_REFERENCIA_TECNICA_SAVEURL.md` +7. `07_RUNBOOK_OPERACION.md` +8. `08_TROUBLESHOOTING.md` +9. `09_SEGURIDAD_Y_SECRETS.md` + diff --git a/documentacion/01_VISION_GENERAL.md b/documentacion/01_VISION_GENERAL.md new file mode 100644 index 0000000..2c38a3c --- /dev/null +++ b/documentacion/01_VISION_GENERAL.md @@ -0,0 +1,68 @@ +# Visión general + +## ¿Qué hace este proyecto? + +El proyecto **lee listados de archivos desde un servidor FTP**, **construye URLs públicas** para esos archivos y **guarda las URLs en SQL Server** dentro de una tabla llamada `URL_FTP`. + +Durante la ejecución también: + +- publica eventos de ejecución en **Kafka** (topic `executions_logs`) +- envía una **notificación por webhook** ante errores (mensaje tipo “alerta pipeline”) + +El entrypoint operativo está pensado para ejecutarse desde **Apache Airflow** a través de un DAG. + +## Componentes principales + +- `SaveUrl.py` + - contiene la lógica de negocio (`run_save_url`) + - implementa: + - lectura FTP (`read_ftp`) + - lectura SQL filtrada por carpeta (`read_table_where`) + - inserción en SQL en lotes (`insert_records`) + - logger a Kafka (`KafkaLogger`) +- `dag_save_url_ftp.py` + - define el DAG `save_url_ftp` y ejecuta `run_save_url` + +## Entornos y configuración + +Este proyecto se configura con **variables de Airflow** (Airflow Variables). Aunque el `README.md` menciona `.env`, el código actual usa `airflow.models.Variable.get(...)` para obtener: + +- credenciales SQL Server +- credenciales FTP +- configuración Kafka (incluyendo SSL/SASL) +- URL de webhook de notificación + +## Frecuencia de ejecución + +El DAG está programado con: + +- cron: `0 */2 * * *` (**cada 2 horas**) + +## Fuentes/“BDs” procesadas + +En cada ejecución se itera sobre una lista fija (en `SaveUrl.py`) con entradas del tipo: + +- nombre de carpeta base (ej. `glmpmi`) +- `identity_id` usado para parsear el nombre de archivo/URL +- nombre de “database” lógico usado para logs (ej. `glm_pmi`) +- `database_id` (para construir `procedure_id` del log) + +Para cada entrada se calcula un `code_folder` con el mes actual (formato `MM`), por ejemplo: `glmpmi04`. + +## Salida principal + +- **Tabla destino**: `URL_FTP` (SQL Server) +- **Campos calculados por registro** (a partir de la URL): + - `Folder` + - `Date` + - `IDcodigo` + - `ObjectCode` + - `TaskName` + - `URL` + +## Señales de observabilidad + +- Logs en Airflow (logger estándar de Python) +- Logs “estructurados” en Kafka en `executions_logs` +- Notificación por webhook en fallos de FTP/SQL/INSERT + diff --git a/documentacion/02_INSTALACION_LOCAL.md b/documentacion/02_INSTALACION_LOCAL.md new file mode 100644 index 0000000..6a43ea3 --- /dev/null +++ b/documentacion/02_INSTALACION_LOCAL.md @@ -0,0 +1,73 @@ +# Instalación local (desarrollo / diagnóstico) + +## Objetivo + +Aunque el flujo está pensado para ejecutarse en Airflow, esta guía sirve para: + +- reproducir fallos fuera del scheduler +- validar conectividad a FTP/SQL/Kafka +- ejecutar pruebas manuales de parsing de URLs + +## Requisitos + +- Windows (PowerShell) o Linux/macOS con Python 3.x +- Driver ODBC para SQL Server (recomendado: **ODBC Driver 18 for SQL Server**) +- Acceso de red a: + - servidor FTP (`FTP_000oqta_SERVER`) + - SQL Server (`DB_MSSQ_SERVER`) + - Kafka (`BOOTSTRAP_SERVERS` o `BOOTSTRAP_SERVERS_DOCKER`, según corresponda) + +## Crear entorno virtual + +En PowerShell: + +```powershell +cd d:\ProgramGLM\SaveUrl +python -m venv env +.\env\Scripts\Activate.ps1 +python -m pip install --upgrade pip +``` + +## Instalar dependencias + +El código importa al menos: + +- `requests` +- `pyodbc` +- `kafka` (librería `kafka-python` / `kafka-python-ng`) +- `apache-airflow` (porque `SaveUrl.py` usa `airflow.models.Variable`) + +Instalación típica (ajusta según tu entorno): + +```powershell +pip install requests pyodbc kafka-python apache-airflow +``` + +> Nota: instalar Airflow localmente en Windows puede ser complejo. Si tu objetivo es **solo** ejecutar localmente, considera ejecutar en un contenedor Linux o adaptar temporalmente el código para leer variables desde `os.getenv`/`.env`. Esta documentación **no modifica** el código. + +## Variables para ejecución local + +El `SaveUrl.py` actual lee configuración desde **Airflow Variables** (`Variable.get`). Para ejecutar fuera de Airflow tienes 2 opciones: + +1. **Ejecutar dentro de un entorno Airflow** (recomendado) y definir las Variables. +2. **Crear un “shim” local**: arrancar un script wrapper que inicialice Airflow y cargue Variables. (Esto depende de cómo esté configurado tu Airflow local). + +Por eso, para validación local rápida sin Airflow, lo más útil suele ser: + +- probar funciones aisladas (FTP/SQL) en un REPL, o +- ejecutar dentro de un contenedor Airflow/Composer donde ya existan las Variables. + +## Ejecución manual (si ya estás dentro de Airflow) + +Si Airflow ya está instalado/configurado y el archivo está en el path correcto, puedes invocar: + +```powershell +python -c "from SaveUrl import run_save_url; run_save_url()" +``` + +## Verificación mínima + +- **FTP**: confirma que existe el path base esperado: `FTP_BASE_PATH` + `` +- **SQL**: valida conexión ODBC al servidor y permisos de INSERT/SELECT sobre `URL_FTP` +- **Kafka**: valida que el broker esté accesible y el topic `executions_logs` exista (o que el broker permita auto-create) + diff --git a/documentacion/03_CONFIGURACION_VARIABLES.md b/documentacion/03_CONFIGURACION_VARIABLES.md new file mode 100644 index 0000000..ee75181 --- /dev/null +++ b/documentacion/03_CONFIGURACION_VARIABLES.md @@ -0,0 +1,72 @@ +# Configuración (Airflow Variables) + +## Fuente de configuración real + +El código en `SaveUrl.py` obtiene configuración principalmente desde **Airflow Variables** usando `airflow.models.Variable.get(...)`. + +Esto implica: + +- Debes definir estas Variables en tu instancia de Airflow (UI o CLI). +- El archivo `.env.example` sirve como **referencia de nombres**, pero **no es la fuente** en ejecución si corres bajo Airflow. + +## Variables requeridas + +### SQL Server + +- `DB_MSSQ_SERVER`: hostname o IP del SQL Server +- `DB_MSSQ_USER`: usuario +- `DB_MSSQ_PASSWORD`: password + +Notas: + +- La conexión ODBC usa `ODBC Driver 18 for SQL Server`. +- La tabla destino es fija: `URL_FTP`. +- El nombre de “database” (catálogo) se toma de la lista `INFORMATIONS` dentro del código (ej. `glm_pmi`, `IS001`, etc.). + +### FTP + +- `FTP_000oqta_SERVER` +- `FTP_000oqta_USER` +- `FTP_000oqta_PASSWORD` + +Notas: + +- El código construye el path como `FTP_BASE_PATH + code_folder`. +- `code_folder` se calcula como `` (mes actual), por ejemplo `glmpmi04`. + +### Kafka + +- `BOOTSTRAP_SERVERS_DOCKER`: brokers accesibles desde el entorno Airflow (por ejemplo `kafka-kraft:9092`) +- `KAFKA_USE_SSL`: `"True"` o `"False"` + +Si `KAFKA_USE_SSL == "True"`, además se leen (desde variables de entorno del proceso, vía `os.getenv`, no Airflow Variables): + +- `SECURITY_PROTOCOL` (ej. `SASL_SSL`) +- `SASL_MECHANISM` (ej. `SCRAM-SHA-256`) +- `SASL_PLAIN_USERNAME` +- `SASL_PLAIN_PASSWORD` +- `SSL_CAFILE` (path al CA dentro del contenedor/host) +- `SSL_CHECK_HOSTNAME` (`True`/`False`) + +> Importante: en el código actual, `BOOTSTRAP_SERVERS_DOCKER` y `KAFKA_USE_SSL` se obtienen como Airflow Variables, pero los demás parámetros de seguridad se obtienen desde `os.getenv`. En despliegues en contenedores, asegúrate de definir esas variables en el **entorno del scheduler/worker**. + +### Webhook de notificación + +- `WEBHOOK_URL_NOTIFICATION`: endpoint que acepta `POST` con JSON `{"message": "..."}` + +## Plantilla `.env.example` + +El archivo `.env.example` contiene los nombres esperados (útil para referencia), pero: + +- el proyecto **no debería** versionar un `.env` real con secretos +- en Airflow, se recomienda gestionar secretos con backend (Vault/Secrets Manager/etc.) o Variables protegidas + +## Validación recomendada (pre-flight) + +Antes de activar el DAG en producción, valida: + +- Variables existen (sin typos) +- SQL user tiene `SELECT` e `INSERT` sobre `URL_FTP` +- FTP credenciales permiten `NLST` sobre el path base +- Kafka accesible desde el contenedor/worker que ejecuta el DAG + diff --git a/documentacion/04_DESPLIEGUE_AIRFLOW.md b/documentacion/04_DESPLIEGUE_AIRFLOW.md new file mode 100644 index 0000000..3cbde44 --- /dev/null +++ b/documentacion/04_DESPLIEGUE_AIRFLOW.md @@ -0,0 +1,75 @@ +# Despliegue en Apache Airflow + +## Qué se despliega + +- DAG: `dag_save_url_ftp.py` +- Módulo Python: `SaveUrl.py` (debe ser importable por el DAG) + +## Ruta esperada por el DAG + +En `dag_save_url_ftp.py` se define un `sys.path.insert`: + +- `sys.path.insert(0, '/opt/airflow/dags/save_url_ftp')` + +Por tanto, dentro del contenedor/host donde corre Airflow, se espera: + +- `/opt/airflow/dags/save_url_ftp/SaveUrl.py` +- `/opt/airflow/dags/save_url_ftp/dag_save_url_ftp.py` + +> Si tu instalación usa otra ruta de `dags_folder`, ajusta el montaje/volumen para que esa estructura exista dentro del runtime. + +## Parámetros del DAG + +En `dag_save_url_ftp.py`: + +- `dag_id`: `save_url_ftp` +- schedule: `0 */2 * * *` (cada 2 horas) +- `retries`: 2 +- `retry_delay`: 5 minutos +- `catchup`: `False` +- `max_active_runs`: 1 + +Estructura de tareas: + +- `inicio` (EmptyOperator) → `save_url_ftp` (PythonOperator) → `fin` (EmptyOperator) + +## Variables de Airflow necesarias + +Define las Variables descritas en `03_CONFIGURACION_VARIABLES.md`. + +En particular, `SaveUrl.py` requiere: + +- `WEBHOOK_URL_NOTIFICATION` +- `BOOTSTRAP_SERVERS_DOCKER` +- `KAFKA_USE_SSL` +- `FTP_000oqta_SERVER`, `FTP_000oqta_USER`, `FTP_000oqta_PASSWORD` +- `DB_MSSQ_USER`, `DB_MSSQ_PASSWORD`, `DB_MSSQ_SERVER` + +## Permisos y conectividad + +- **Worker/Scheduler** que ejecute el PythonOperator debe tener: + - salida a red hacia FTP, SQL Server y Kafka + - driver ODBC instalado (si se usa `pyodbc` en el worker) + +## Operación esperada en Airflow + +Durante la ejecución, el task `save_url_ftp` escribe logs tipo: + +- inicio/final +- estado por “BD” lógica (cada item de `INFORMATIONS`) +- conteos: archivos FTP, registros existentes, registros a insertar + +Si ocurre un error en FTP/SQL/INSERT: + +- registra `FAILED` en Kafka +- envía webhook con el mensaje de alerta +- continúa con la siguiente “BD” (no falla toda la corrida por un único item) + +## Checklist de despliegue + +- DAG visible y sin errores de importación +- `SaveUrl.py` importable desde la ruta insertada en `sys.path` +- Variables creadas (y protegidas si aplica) +- Driver ODBC disponible en el runtime +- Conectividad validada (FTP/SQL/Kafka/webhook) + diff --git a/documentacion/05_FLUJO_Y_ARQUITECTURA.md b/documentacion/05_FLUJO_Y_ARQUITECTURA.md new file mode 100644 index 0000000..8fc2232 --- /dev/null +++ b/documentacion/05_FLUJO_Y_ARQUITECTURA.md @@ -0,0 +1,99 @@ +# Flujo y arquitectura + +## Flujo end-to-end + +El flujo, por cada entrada en `INFORMATIONS`, es: + +1. Calcular `code_folder = ` (mes actual). +2. Leer lista de archivos del FTP en `FTP_BASE_PATH + code_folder`. +3. Convertir paths FTP a URLs públicas y **parsear metadatos** desde la URL. +4. Leer desde SQL Server la tabla `URL_FTP` filtrando por `Folder LIKE %code_folder%`. +5. Calcular diferencia por `IDcodigo`: + - insertar solo los registros nuevos (los no existentes en SQL). +6. Emitir logs a Kafka y escribir logs en Airflow. +7. En caso de error, registrar `FAILED` y notificar por webhook; continuar con el siguiente item. + +## Diagrama (alto nivel) + +```mermaid +flowchart LR + A[Airflow DAG save_url_ftp] --> B[PythonOperator: run_save_url] + B --> C{Por cada item en INFORMATIONS} + C --> D[FTP: nlst(path)] + D --> E[Construir URL pública + parse] + E --> F[SQL: SELECT por Folder] + F --> G{Comparar IDcodigo} + G -->|Nuevos| H[SQL: INSERT batches] + G -->|Sin nuevos| I[Sin INSERT] + D --> K[Kafka: executions_logs] + F --> K + H --> K + B --> L[Webhook: alerta en errores] +``` + +## Detalles del parsing (URL → campos) + +`read_ftp()`: + +- obtiene `file_names` via `ftp.nlst(path)` +- construye URL pública como: + - `FTP_BASE_URL + f.replace('/www', '')` +- por cada URL calcula: + - `Folder`: segmento `code_folder` dentro de la URL + - `Date`: substring encontrado con patrón `_{YYYY}-{MM}` + - `IDcodigo`: substring que empieza donde aparece `identity_id` y toma 24 caracteres + - `ObjectCode`: substring entre `Date` y `IDcodigo` + - `TaskName`: substring entre el fin de `code_folder + "/"` y el `_YYYY-MM...` + - `URL`: la URL pública final + +Esto asume que los nombres de archivos/paths en el FTP siguen un patrón consistente, por ejemplo: + +- contiene `_-` como parte del nombre +- contiene el `identity_id` en el nombre + +## Interacción con SQL Server + +### Lectura + +`read_table_where()` ejecuta: + +- `SELECT * FROM [URL_FTP] WHERE Folder LIKE '%%'` + +Luego construye una lista de dicts con todas las columnas. + +### Inserción + +`insert_records()`: + +- inserta en lotes de 1000 (`executemany`) +- arma la lista de columnas desde las llaves del primer dict, excluyendo: + - `ID` + - `ExportDate` + +Implicaciones: + +- La tabla `URL_FTP` debe tener (como mínimo) columnas que coincidan con las llaves calculadas por `read_ftp()` (o defaults/allow null). +- Si existen columnas extra en la tabla, el código **no** las insertará a menos que estén presentes en el dict. + +## Observabilidad (Kafka) + +El proyecto envía logs “estructurados” a Kafka con campos como: + +- `database`, `procedure_id`, `procedure_name` +- `procedure_type` (fijo: `PYTHON_APP`) +- `status` (`STARTING`, `SUCCESS`, `FAILED`, `FINISHED`) +- `message`, `executed_by`, `source` +- `target_date`, `date` +- `extra_info` (serializado a string) + +El `procedure_id` se construye como: + +- `` + `PROCEDURE_ID_SUFFIX` + +## Decisiones y limitaciones actuales + +- **Configuración mezclada**: parte se lee como Airflow Variables y parte como variables de entorno (`os.getenv`) para Kafka SSL/SASL. +- **Fechas**: usa la fecha/hora local del worker (`datetime.now()`), no el `execution_date` de Airflow. +- **No hay deduplicación por URL**: la deduplicación se hace por `IDcodigo` únicamente. +- **Errores por item**: si falla un item de `INFORMATIONS`, el script continúa con el siguiente. + diff --git a/documentacion/06_REFERENCIA_TECNICA_SAVEURL.md b/documentacion/06_REFERENCIA_TECNICA_SAVEURL.md new file mode 100644 index 0000000..7094dcf --- /dev/null +++ b/documentacion/06_REFERENCIA_TECNICA_SAVEURL.md @@ -0,0 +1,147 @@ +# Referencia técnica (`SaveUrl.py`) + +## Constantes relevantes + +- `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`: se obtiene desde Airflow Variable `WEBHOOK_URL_NOTIFICATION` +- `INFORMATIONS`: lista de fuentes a procesar (carpeta, identity_id, database lógico, database_id) + +## Clase `KafkaLogger` + +### Responsabilidad + +Publica logs de ejecución a Kafka en el topic: + +- `executions_logs` + +### Inicialización + +Parámetros: + +- `source`: string (se usa `f"{APP_NAME}_{VERSION}"`) +- `bootstrap_servers`: lista/string de brokers +- `use_ssl`: boolean + +Comportamiento: + +- serializa `value` como JSON (`json.dumps(...).encode('utf-8')`) +- si `use_ssl` es `True`, agrega configuración SASL/SSL desde `os.getenv` + +### Método `log(...)` + +Firma (conceptual): + +- `log(database, db_id, procedure_name, status, message, executed_by, target_date, date, extra_info=None)` + +Payload enviado a Kafka (campos clave): + +- `database`: nombre lógico (ej. `glm_pmi`) +- `procedure_id`: `str(db_id) + PROCEDURE_ID_SUFFIX` +- `procedure_name`: texto como `Generacion de URL ` +- `procedure_type`: `PYTHON_APP` +- `status`: `STARTING` | `SUCCESS` | `FAILED` | `FINISHED` +- `message`: descripción del evento +- `executed_by`: usuario SQL (`DB_MSSQ_USER`) +- `source`: `source` del logger +- `extra_info`: string (sanitiza comillas simples) +- `target_date`: fecha objetivo (`YYYY-MM-DD`) +- `date`: timestamp (string) + +## Función `read_ftp(...)` + +### Responsabilidad + +Conecta al FTP y lista archivos en: + +- `FTP_BASE_PATH + code_folder` + +Luego construye URLs públicas y parsea metadatos en un dict. + +Parámetros: + +- `ftp_server`, `ftp_user`, `ftp_password` +- `code_folder`: carpeta calculada (``) +- `identity_id`: string para ubicar `IDcodigo` dentro del nombre +- `year`, `month`: se usan para encontrar `_{YYYY}-{MM}` + +Salida: + +- `list[dict]` con llaves: `Folder`, `Date`, `IDcodigo`, `ObjectCode`, `TaskName`, `URL` + +Errores esperados: + +- fallo de login o conectividad FTP +- path inexistente (depende de cómo responda `nlst`) +- parsing incorrecto si el nombre no contiene los patrones esperados + +## Función `read_table_where(...)` + +### Responsabilidad + +Lee registros existentes en SQL Server desde una tabla, filtrando por `Folder`. + +Query: + +- `SELECT * FROM [URL_FTP] WHERE Folder LIKE '%%'` + +Salida: + +- lista de dicts (columna → valor) + +Errores esperados: + +- driver ODBC no instalado en runtime +- credenciales inválidas / permisos insuficientes +- tabla inexistente + +## Función `insert_records(...)` + +### Responsabilidad + +Inserta registros en SQL Server en lotes de 1000. + +Detalles: + +- toma columnas desde el primer elemento del `data` +- excluye `ID` y `ExportDate` si existen en el dict +- arma un `INSERT INTO [URL_FTP] ([col1],...) VALUES (?,...)` + +Salida: + +- cantidad de registros insertados (int) + +Errores esperados: + +- tipos incompatibles con schema SQL +- restricciones (NOT NULL, UNIQUE) no satisfechas +- problemas de red/conexión + +## Función `run_save_url(**context)` + +### Responsabilidad + +Orquesta toda la ejecución: + +- inicializa `KafkaLogger` +- obtiene Variables de Airflow (FTP/SQL/Kafka/webhook) +- recorre `INFORMATIONS` +- para cada item: + - `read_ftp` → `read_table_where` → diff por `IDcodigo` → `insert_records` + - logging a Kafka por pasos + - notificación por webhook si hay errores + +### Notificación (webhook) + +En errores, construye un texto (formato “ALERTA PIPELINE”) y hace: + +- `POST ` con JSON: `{"message": ""}` + +### Consideraciones Airflow + +- El parámetro `**context` permite que el callable sea usado por `PythonOperator`. +- La función usa `datetime.now()` (no usa `execution_date` de Airflow). + diff --git a/documentacion/07_RUNBOOK_OPERACION.md b/documentacion/07_RUNBOOK_OPERACION.md new file mode 100644 index 0000000..cb76016 --- /dev/null +++ b/documentacion/07_RUNBOOK_OPERACION.md @@ -0,0 +1,66 @@ +# Runbook de operación + +## Objetivo operacional + +Mantener la sincronización: + +- FTP (archivos del mes/carpeta) → SQL Server (`URL_FTP`) + +con observabilidad en Kafka y alerta por webhook. + +## Qué esperar en una ejecución “sana” + +Por cada item en `INFORMATIONS` se esperan, en logs de Airflow: + +- `✅ FTP ... archivos en ` +- `✅ SQL ... registros existentes` +- `ℹ️ DIFF ... A insertar: ` +- `✅ INSERT ... Insertados: ` (o warning de “Sin registros nuevos”) + +Al final: + +- `⏹ FINALIZADO ...` + +## Inputs críticos (pre-flight) + +- **Airflow Variables** existen y tienen valores válidos: + - FTP: `FTP_000oqta_SERVER`, `FTP_000oqta_USER`, `FTP_000oqta_PASSWORD` + - SQL: `DB_MSSQ_SERVER`, `DB_MSSQ_USER`, `DB_MSSQ_PASSWORD` + - Kafka: `BOOTSTRAP_SERVERS_DOCKER`, `KAFKA_USE_SSL` + - Webhook: `WEBHOOK_URL_NOTIFICATION` +- El worker donde corre el DAG tiene: + - driver ODBC 18 instalado + - conectividad a FTP/SQL/Kafka/webhook + +## Procedimiento ante incidentes (rápido) + +1. Revisar logs del task `save_url_ftp` en Airflow. +2. Identificar el “BD lógico” afectado (ej. `glm_pmi`, `IS001`, etc.). +3. Clasificar el fallo: + - FTP (login, path, listing) + - SQL (conexión, permisos, tabla) + - INSERT (schema/constraint) + - Kafka (broker/SSL/SASL) + - Webhook (endpoint caído / 4xx/5xx) +4. Ejecutar acciones según tipo (ver `08_TROUBLESHOOTING.md`). + +## Operación programada del DAG + +El DAG corre cada 2 horas (`0 */2 * * *`). + +Si necesitas ejecución manual: + +- Ejecuta el task `save_url_ftp` desde la UI de Airflow (Trigger DAG / Run Task). + +## Indicadores recomendados (SLOs simples) + +- **Frescura**: últimas URLs del FTP aparecen en SQL < 2–4 horas. +- **Cobertura**: por cada item, `A insertar` tiende a 0 tras una corrida exitosa. +- **Errores**: tasa de `FAILED` por item debe ser baja; si sube, revisar conectividad o cambios de naming en FTP. + +## Mantenimiento preventivo + +- Revisar que `FTP_BASE_PATH` (año) sea correcto cuando cambie el año. +- Verificar que las rutas/certificados SSL (si aplica) sean válidos dentro del contenedor. +- Confirmar que el naming en FTP siga el patrón esperado (contiene `_{YYYY}-{MM}` e `identity_id`). + diff --git a/documentacion/08_TROUBLESHOOTING.md b/documentacion/08_TROUBLESHOOTING.md new file mode 100644 index 0000000..f76e1ae --- /dev/null +++ b/documentacion/08_TROUBLESHOOTING.md @@ -0,0 +1,135 @@ +# Troubleshooting + +## 1) El DAG no aparece o falla al importarse + +**Síntomas** + +- Airflow muestra “ImportError” o el DAG no se lista. + +**Causas comunes** + +- La ruta `'/opt/airflow/dags/save_url_ftp'` no existe dentro del contenedor. +- `SaveUrl.py` no está en esa carpeta (o tiene otro nombre). +- Dependencias no instaladas en el entorno del worker (ej. `pyodbc`, `kafka`). + +**Acciones** + +- Verifica que existan los archivos: + - `/opt/airflow/dags/save_url_ftp/dag_save_url_ftp.py` + - `/opt/airflow/dags/save_url_ftp/SaveUrl.py` +- Revisa el scheduler logs por import errors. + +## 2) Error conectando al FTP + +**Síntomas** + +- `❌ FTP | ...` en logs +- notificación por webhook con “Error leyendo FTP folder ...” + +**Causas comunes** + +- Credenciales incorrectas (`FTP_000oqta_USER` / `FTP_000oqta_PASSWORD`) +- Host incorrecto (`FTP_000oqta_SERVER`) +- Firewall / DNS / ruteo +- La carpeta del mes no existe (``) + +**Acciones** + +- Probar conectividad desde el mismo runtime (worker) hacia el host FTP. +- Confirmar que `FTP_BASE_PATH` + `code_folder` existe en el servidor. +- Verificar si el servidor cambió estructura para 2027/otro año (porque `FTP_BASE_PATH` incluye `2026`). + +## 3) Error leyendo SQL Server + +**Síntomas** + +- `❌ SQL | ...` en logs +- webhook “Error leyendo URLs ...” + +**Causas comunes** + +- Driver ODBC faltante en el worker (`ODBC Driver 18 for SQL Server`) +- Credenciales inválidas o permisos insuficientes +- SQL Server inaccesible desde la red del worker +- Tabla `URL_FTP` inexistente o en otro schema/db esperado + +**Acciones** + +- Verificar instalación del driver ODBC en el contenedor/host. +- Validar que el usuario tenga `SELECT` sobre `URL_FTP`. +- Confirmar nombre de la base de datos: se toma de `INFORMATIONS` (ej. `glm_pmi`, `IS001`). + +## 4) Error insertando (schema/constraints) + +**Síntomas** + +- `❌ INSERT | ...` en logs +- webhook “Error insertando URLs ...” + +**Causas comunes** + +- La tabla tiene columnas NOT NULL sin default que el dict no provee. +- Tipos incompatibles (ej. `Date` espera datetime pero recibe string). +- Restricción UNIQUE (por `IDcodigo`, `URL`, u otra combinación) que el código no contempla. + +**Acciones** + +- Revisar definición de `URL_FTP` y confirmar que acepta las columnas: + - `Folder`, `Date`, `IDcodigo`, `ObjectCode`, `TaskName`, `URL` +- Confirmar tipos: el código inserta strings. +- Si existe una columna “ExportDate”, el código NO la inserta (la excluye si aparece en dict). + +## 5) No inserta nada, pero debería + +**Síntomas** + +- `A insertar: 0` repetidamente, aunque hay archivos nuevos en FTP. + +**Causas comunes** + +- `IDcodigo` se parsea mal (mismo valor para distintos archivos) por cambios de naming. +- La consulta SQL `Folder LIKE %code_folder%` trae más registros de los necesarios y “tapa” IDs. +- El FTP lista rutas que no cumplen el patrón y el parse genera IDs inesperados. + +**Acciones** + +- Tomar una URL real del FTP y revisar manualmente el parsing: + - ¿contiene `_{YYYY}-{MM}`? + - ¿contiene el `identity_id` correcto? + - ¿`IDcodigo` de 24 chars es correcto? +- Comparar un `IDcodigo` “nuevo” con los existentes en SQL. + +## 6) Kafka no recibe logs + +**Síntomas** + +- Logs de Airflow muestran fallos de Kafka o no hay eventos en `executions_logs`. + +**Causas comunes** + +- `BOOTSTRAP_SERVERS_DOCKER` incorrecto para el runtime del worker. +- `KAFKA_USE_SSL=True` pero faltan variables de entorno `SECURITY_PROTOCOL`, `SASL_*`, `SSL_CAFILE`, etc. +- Certificados/rutas inválidas dentro del contenedor. + +**Acciones** + +- Validar conectividad al broker desde el contenedor/worker. +- Si SSL/SASL: asegurar que las variables están definidas como **env vars** del proceso. + +## 7) Webhook no notifica + +**Síntomas** + +- Se ve el error en logs pero no llega alerta. + +**Causas comunes** + +- `WEBHOOK_URL_NOTIFICATION` no definido o es inválido. +- Endpoint requiere auth/cabeceras adicionales (el código no las envía). +- Restricciones de red del worker. + +**Acciones** + +- Probar un `POST` manual al webhook desde el mismo entorno. +- Revisar respuesta HTTP (4xx/5xx) en logs si se captura (actualmente no se valida el status code). + diff --git a/documentacion/09_SEGURIDAD_Y_SECRETS.md b/documentacion/09_SEGURIDAD_Y_SECRETS.md new file mode 100644 index 0000000..f139135 --- /dev/null +++ b/documentacion/09_SEGURIDAD_Y_SECRETS.md @@ -0,0 +1,58 @@ +# Seguridad y manejo de secretos + +## No versionar secretos + +El repositorio incluye `.gitignore` para ignorar: + +- `.env` +- `env/` (virtualenv) + +Recomendación: + +- Mantener **cualquier credencial** fuera del control de versiones. + +## Dónde viven los secretos en este proyecto + +El código en `SaveUrl.py` toma secretos principalmente desde: + +- **Airflow Variables**: + - SQL: `DB_MSSQ_USER`, `DB_MSSQ_PASSWORD`, `DB_MSSQ_SERVER` + - FTP: `FTP_000oqta_USER`, `FTP_000oqta_PASSWORD`, `FTP_000oqta_SERVER` + - Webhook: `WEBHOOK_URL_NOTIFICATION` + - Kafka básico: `BOOTSTRAP_SERVERS_DOCKER`, `KAFKA_USE_SSL` +- **Variables de entorno del proceso** (solo cuando `KAFKA_USE_SSL=True`): + - `SECURITY_PROTOCOL`, `SASL_MECHANISM` + - `SASL_PLAIN_USERNAME`, `SASL_PLAIN_PASSWORD` + - `SSL_CAFILE`, `SSL_CHECK_HOSTNAME` + +Implicación: + +- Para Kafka con SSL/SASL, no basta con Airflow Variables: hay que configurar también **env vars** en el worker/scheduler (o usar un backend de secrets que las exponga como env vars). + +## Recomendaciones de seguridad + +- **Airflow Secrets Backend**: usar un backend (Vault/AWS Secrets Manager/Azure Key Vault/GCP Secret Manager) en vez de Variables planas. +- **Rotación**: + - FTP y SQL: rotación periódica y seguimiento de expiración. + - Kafka: rotación de credenciales SASL y CA si cambia. +- **Principio de mínimo privilegio**: + - SQL user: permisos mínimos necesarios sobre `URL_FTP` (SELECT/INSERT). + - FTP user: permisos mínimos para listar/leer rutas necesarias (NLST). +- **TLS**: + - si se habilita SSL en Kafka, asegurar que `SSL_CAFILE` apunte a un CA dentro del contenedor. + +## Señales de filtración / exposición + +Revisar que no exista: + +- `.env` real en el repo +- credenciales hardcodeadas en archivos `.py` +- rutas de certificados locales de un desarrollador (ej. paths absolutos de Windows) en configuración de producción + +## Checklist antes de producción + +- Variables/secretos definidos en sistema seguro +- `.env` excluido y no presente en artefactos de build +- permisos SQL y FTP mínimos +- endpoints (Kafka/webhook) accesibles solo desde redes necesarias + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..22bd495 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +requests==2.31.0 +pyodbc==5.1.0 +kafka-python-ng==2.2.2 +apache-airflow==2.8.1