Antigravity Migration Agent
148 lines
3.7 KiB
Markdown
148 lines
3.7 KiB
Markdown
# 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).
|
|
|