Antigravity Migration Agent
100 lines
3.3 KiB
Markdown
100 lines
3.3 KiB
Markdown
# Flujo y arquitectura
|
|
|
|
## Flujo end-to-end
|
|
|
|
El flujo, por cada entrada en `INFORMATIONS`, es:
|
|
|
|
1. Calcular `code_folder = <name_folder><MM>` (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 `_<YYYY>-<MM>` 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 '%<code_folder>%'`
|
|
|
|
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:
|
|
|
|
- `<database_id>` + `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.
|
|
|