# Instalar Python y pip si no están disponibles
sudo apt update && sudo apt install -y python3 python3-pip python3-venv  # Ubuntu/Debian
sudo dnf install -y python3 python3-pip                                    # CentOS/Rocky

# Crear entorno virtual dedicado para dbt
python3 -m venv ~/.venv/dbt
source ~/.venv/dbt/bin/activate

# Instalar dbt con el adaptador de tu base de datos
# PostgreSQL / Redshift
pip install dbt-postgres

# Snowflake
pip install dbt-snowflake

# BigQuery
pip install dbt-bigquery

# DuckDB (ideal para desarrollo local)
pip install dbt-duckdb

# Verificar instalación
dbt --version

echo 'source ~/.venv/dbt/bin/activate' >> ~/.bashrc
source ~/.bashrc

# Inicializar un nuevo proyecto dbt
dbt init mi_proyecto_dbt

# El asistente preguntará:
# - Nombre del proyecto (ya introducido)
# - Adaptador de base de datos a usar
# - Credenciales de conexión

cd mi_proyecto_dbt
ls -la

mi_proyecto_dbt/
├── dbt_project.yml        # Configuración del proyecto
├── profiles.yml           # Conexiones a bases de datos (NO en el repo)
├── models/                # Modelos SQL
│   ├── staging/          # Capa de staging (datos crudos limpiados)
│   ├── intermediate/     # Transformaciones intermedias
│   └── marts/            # Tablas finales para BI/analytics
├── tests/                 # Tests personalizados
├── macros/                # Funciones SQL reutilizables (Jinja)
├── seeds/                 # Datos CSV estáticos
├── snapshots/             # Capturas de datos que cambian con el tiempo
└── analyses/              # Análisis SQL ad-hoc

name: mi_proyecto_dbt
version: '1.0.0'
config-version: 2

# Variables del proyecto
profile: mi_perfil_postgres

model-paths: ["models"]
test-paths: ["tests"]
seed-paths: ["seeds"]
macro-paths: ["macros"]
snapshot-paths: ["snapshots"]

# Configuración global de modelos
models:
  mi_proyecto_dbt:
    # Capa staging: vistas (no materializadas)
    staging:
      +materialized: view
      +schema: staging
    # Capa marts: tablas físicas
    marts:
      +materialized: table
      +schema: analytics

# ~/.dbt/profiles.yml

mi_perfil_postgres:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: dbt_user
      password: "{{ env_var('DBT_POSTGRES_PASSWORD') }}"
      dbname: mi_datawarehouse
      schema: dbt_dev
      threads: 4
      keepalives_idle: 0

    prod:
      type: postgres
      host: postgres-prod.empresa.com
      port: 5432
      user: dbt_prod
      password: "{{ env_var('DBT_POSTGRES_PASSWORD_PROD') }}"
      dbname: datawarehouse
      schema: analytics
      threads: 8

# Probar la conexión
dbt debug

# La salida debería mostrar "All checks passed!"

-- models/staging/stg_ventas.sql
-- Limpieza básica de la tabla cruda de ventas

SELECT
    id_venta::INTEGER                           AS id_venta,
    TRIM(LOWER(cliente_email))                  AS email_cliente,
    importe::NUMERIC(10, 2)                     AS importe,
    fecha_venta::DATE                           AS fecha_venta,
    COALESCE(categoria, 'Sin categoría')        AS categoria,
    CASE
        WHEN estado IN ('completada', 'pagada') THEN 'completada'
        WHEN estado = 'cancelada'               THEN 'cancelada'
        ELSE 'pendiente'
    END                                         AS estado_normalizado,
    _cargado_at                                 AS cargado_en_staging_at
FROM {{ source('raw', 'ventas_raw') }}
WHERE id_venta IS NOT NULL
  AND fecha_venta >= '2020-01-01'

-- models/marts/mart_ventas_mensual.sql
-- Agrupación mensual de ventas para dashboards BI

WITH ventas_staged AS (
    SELECT * FROM {{ ref('stg_ventas') }}
),

clientes AS (
    SELECT * FROM {{ ref('stg_clientes') }}
)

SELECT
    DATE_TRUNC('month', v.fecha_venta)   AS mes,
    v.categoria,
    c.segmento_cliente,
    COUNT(DISTINCT v.id_venta)           AS num_ventas,
    SUM(v.importe)                       AS total_ventas,
    AVG(v.importe)                       AS ticket_medio,
    COUNT(DISTINCT v.email_cliente)      AS clientes_unicos
FROM ventas_staged v
LEFT JOIN clientes c ON v.email_cliente = c.email
WHERE v.estado_normalizado = 'completada'
GROUP BY 1, 2, 3

# Ejecutar todos los modelos
dbt run

# Ejecutar un modelo específico
dbt run --select stg_ventas

# Ejecutar un modelo y todos sus descendientes
dbt run --select stg_ventas+

# Ejecutar modelos de una carpeta
dbt run --select marts.*

# models/staging/schema.yml

version: 2

models:
  - name: stg_ventas
    description: "Tabla de ventas limpiada desde la fuente raw"
    columns:
      - name: id_venta
        description: "Identificador único de la venta"
        tests:
          - unique              # No hay duplicados
          - not_null           # No hay nulos

      - name: estado_normalizado
        tests:
          - accepted_values:   # Solo valores permitidos
              values: ['completada', 'cancelada', 'pendiente']

      - name: importe
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"  # El importe no puede ser negativo

      - name: email_cliente
        tests:
          - not_null
          - relationships:     # Integridad referencial
              to: ref('stg_clientes')
              field: email

-- Test: no debe haber ventas con fecha futura
SELECT *
FROM {{ ref('stg_ventas') }}
WHERE fecha_venta > CURRENT_DATE

# Ejecutar todos los tests
dbt test

# Tests de un modelo específico
dbt test --select stg_ventas

# Ejecutar modelos y tests juntos
dbt build

# Generar la documentación del proyecto
dbt docs generate

# Servir la documentación en el navegador (localhost:8080)
dbt docs serve --port 8080

models:
  - name: mart_ventas_mensual
    description: >
      Agregación mensual de ventas completadas, enriquecida con
      el segmento de cliente. Usada por los dashboards de Metabase
      y los informes mensuales de dirección.

    config:
      tags: ['marts', 'ventas', 'mensual']

-- models/marts/mart_eventos_incremental.sql
{{
  config(
    materialized='incremental',
    unique_key='id_evento',
    incremental_strategy='delete+insert'
  )
}}

SELECT
    id_evento,
    usuario_id,
    tipo_evento,
    propiedades,
    ocurrido_en
FROM {{ source('raw', 'eventos') }}

{% if is_incremental() %}
  -- Solo procesar registros nuevos desde la última ejecución
  WHERE ocurrido_en > (SELECT MAX(ocurrido_en) FROM {{ this }})
{% endif %}

# .github/workflows/dbt-ci.yml
name: dbt CI

on:
  pull_request:
    branches: [main]

jobs:
  dbt-ci:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Configurar Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Instalar dbt
        run: pip install dbt-postgres

      - name: Crear profiles.yml desde secrets
        run: |
          mkdir -p ~/.dbt
          cat > ~/.dbt/profiles.yml <<EOF
          mi_perfil_postgres:
            target: ci
            outputs:
              ci:
                type: postgres
                host: ${{ secrets.DB_HOST }}
                port: 5432
                user: ${{ secrets.DB_USER }}
                password: ${{ secrets.DB_PASSWORD }}
                dbname: datawarehouse_ci
                schema: dbt_ci_${{ github.event.pull_request.number }}
                threads: 4
          EOF

      - name: Verificar conexión
        run: dbt debug

      - name: Ejecutar modelos y tests
        run: dbt build --target ci
        env:
          DBT_POSTGRES_PASSWORD: ${{ secrets.DB_PASSWORD }}

# Verificar que el perfil en dbt_project.yml coincide con profiles.yml
grep "profile:" dbt_project.yml
cat ~/.dbt/profiles.yml | grep -A1 "outputs:"

# Especificar perfil explícitamente
dbt run --profile mi_perfil_postgres

# Ver el SQL compilado para depurar
dbt compile --select nombre_modelo
cat target/compiled/mi_proyecto/models/nombre_modelo.sql

# Ver los registros que fallan en un test
dbt test --select nombre_modelo --store-failures
# Los registros que fallan se guardan en el schema de fallos configurado

# Ver tiempos de ejecución
dbt run --select nombre_modelo
# Revisar el output para ver el tiempo de cada modelo

# Considerar:
# 1. Añadir índices en las columnas de JOIN
# 2. Cambiar materialización a incremental
# 3. Aumentar el número de threads en profiles.yml