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