# 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 <db>`
- `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 (`<name_folder><MM>`)
- `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 '%<folder>%'`

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 <WEBHOOK_URL_NOTIFICATION>` con JSON: `{"message": "<texto>"}`

### 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).