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

2.2 KiB Raw Permalink Blame History