Flujo y arquitectura

Flujo end-to-end

El flujo, por cada entrada en INFORMATIONS, es:

Calcular code_folder = <name_folder><MM> (mes actual).
Leer lista de archivos del FTP en FTP_BASE_PATH + code_folder.
Convertir paths FTP a URLs públicas y parsear metadatos desde la URL.
Leer desde SQL Server la tabla URL_FTP filtrando por Folder LIKE %code_folder%.
Calcular diferencia por IDcodigo:
- insertar solo los registros nuevos (los no existentes en SQL).
Emitir logs a Kafka y escribir logs en Airflow.
En caso de error, registrar FAILED y notificar por webhook; continuar con el siguiente item.

Diagrama (alto nivel)

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.

3.3 KiB Raw Permalink Blame History