NutriFlow — Pipeline de Datos Clínicos

# dags/etl_pacientes.py
from airflow.decorators import dag, task
from datetime import datetime

@dag(
    dag_id='etl_pacientes',
    schedule='0 6 * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'pacientes', 'nutriflow'],
)
def etl_pacientes():
    @task()
    def extract(**context) -> dict:
        """Extrae de PostgreSQL hospitales"""
        ds = context['ds']  # idempotente
        conn = PostgresHook('hospital_db')
        df = conn.get_pandas_df(
            f"""SELECT * FROM pacientes
            WHERE fecha_actualizacion = '{ds}'"""
        )
        return {'rows': len(df), 'data': df.to_dict()}

    @task()
    def transform(extracted: dict) -> dict:
        """Normaliza CIE-10 y unidades clínicas"""
        df = pd.DataFrame(extracted['data'])
        df['cie10_normalizado'] = df['diagnostico'].map(
            normalize_cie10
        )
        df['bmi'] = df['peso_kg'] / (df['altura_m'] ** 2)
        return {'rows': len(df), 'data': df.to_dict()}

    @task()
    def load_bq(transformed: dict, **context):
        """UPSERT en BigQuery (idempotente)"""
        ds = context['ds']
        df = pd.DataFrame(transformed['data'])
        client = bigquery.Client()
        client.load_table_from_dataframe(
            df, f'nutriflow.pacientes${ds.replace("-","")}',
            job_config=bigquery.LoadJobConfig(
                write_disposition='WRITE_TRUNCATE'
            )
        ).result()
        return transformed['rows']

    extracted = extract()
    transformed = transform(extracted)
    load_bq(transformed)

etl_pacientes()

# dags/nutriflow_factory.py
from datetime import datetime, timedelta
from airflow import DAG

PIPELINE_CONFIGS = [
    {
        'name': 'pacientes',
        'schedule': '0 6 * * *',
        'source': 'postgresql://hospital_db',
        'target_table': 'nutriflow.pacientes',
    },
    {
        'name': 'metricas_adherencia',
        'schedule': '@hourly',
        'source': 'nutriflow.pacientes',
        'target_table': 'nutriflow.kpis_adherencia',
    },
    {
        'name': 'informes_clinicos',
        'schedule': '0 8 * * 1',
        'source': 'nutriflow.kpis_adherencia',
        'target_table': 's3://nutriflow-reports/',
    },
]

def create_pipeline_dag(config: dict) -> DAG:
    """Factory: genera DAG desde configuración"""
    dag = DAG(
        dag_id=f"nutriflow_{config['name']}",
        default_args={
            'owner': 'data-team',
            'retries': 3,
            'retry_delay': timedelta(minutes=5),
            'retry_exponential_backoff': True,
            'on_failure_callback': slack_failure_alert,
        },
        schedule=config['schedule'],
        start_date=datetime(2024, 1, 1),
        catchup=False,
        tags=['nutriflow', config['name']],
    )
    return dag

# Registra los 3 DAGs dinámicamente
for cfg in PIPELINE_CONFIGS:
    globals()[f"dag_{cfg['name']}"] = create_pipeline_dag(cfg)

# dags/metricas_adherencia.py
from airflow.operators.python import BranchPythonOperator
from airflow.utils.trigger_rule import TriggerRule

@task()
def calcular_kpis(**context) -> dict:
    ds = context['ds']
    bq = bigquery.Client()
    result = bq.query(f"""
        SELECT AVG(adherencia_pct) AS adherencia_media,
               COUNT(*) AS total_pacientes
        FROM nutriflow.kpis_adherencia
        WHERE fecha_calculo = '{ds}'
    """).to_dataframe()
    return result.iloc[0].to_dict()

def elegir_rama(**context) -> str:
    kpis = context['ti'].xcom_pull('calcular_kpis')
    adh = kpis['adherencia_media']
    if adh >= 0.85:
        return 'rama_alta_adherencia'
    elif adh >= 0.70:
        return 'rama_media_adherencia'
    else:
        return 'rama_alerta_critica'

branch = BranchPythonOperator(
    task_id='branch_adherencia',
    python_callable=elegir_rama,
)

alerta_critica = SlackWebhookOperator(
    task_id='rama_alerta_critica',
    slack_webhook_conn_id='slack_data',
    message="⚠ Adherencia CRITICA: {{ ti.xcom_pull('calcular_kpis')['adherencia_media']:.1%}}",
)

join = EmptyOperator(
    task_id='join',
    trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS,
)

kpis >> branch >> [alta, media, alerta_critica] >> join

# dags/informes_clinicos.py
from airflow.sensors.external_task import ExternalTaskSensor

def slack_failure_alert(context):
    """Callback de fallo con link a logs"""
    ti = context['task_instance']
    msg = (f"*NutriFlow Pipeline Fallido*\n"
           f"DAG: {ti.dag_id}\n"
           f"Task: {ti.task_id}\n"
           f"Fecha: {context['ds']}\n"
           f"Error: {context.get('exception')}\n"
           f"Logs: {ti.log_url}")
    SlackWebhookOperator(
        task_id='_alert',
        slack_webhook_conn_id='slack_data',
        message=msg,
    ).execute(context)

# Sensor: espera a que el ETL diario complete
wait_etl = ExternalTaskSensor(
    task_id='wait_for_etl_pacientes',
    external_dag_id='etl_pacientes',
    external_task_id='load_bq',
    execution_date_fn=lambda dt: dt.replace(hour=6),
    timeout=60 * 60 * 3,   # 3h máximo
    poke_interval=60 * 5,  # verifica cada 5 min
    mode='reschedule',      # libera workers
    on_failure_callback=slack_failure_alert,
)

# Cleanup siempre ejecuta (ALL_DONE)
cleanup = PythonOperator(
    task_id='cleanup_tmp',
    python_callable=limpiar_tmp,
    trigger_rule=TriggerRule.ALL_DONE,
)

wait_etl >> gen_pdfs >> upload_s3 >> notify
gen_pdfs >> cleanup