Sistema de Cobrança (Oracle) 
    ↓ (ETL noturno manual)
CSV exportado para servidor FTP
    ↓ (Script Python local)
Banco SQL Server (Data Mart)
    ↓ (Power BI Import Mode)
Dashboard atualizado 1x/dia

# Azure Resource Group e recursos core
az group create --name rg-lakehouse-prod --location brazilsouth

# Azure Data Lake Storage Gen2
az storage account create \
  --name adlsrecuperacao \
  --resource-group rg-lakehouse-prod \
  --location brazilsouth \
  --sku Standard_LRS \
  --kind StorageV2 \
  --hierarchical-namespace true

# Containers para camadas Medallion
az storage container create --name bronze --account-name adlsrecuperacao
az storage container create --name silver --account-name adlsrecuperacao
az storage container create --name gold --account-name adlsrecuperacao
az storage container create --name checkpoints --account-name adlsrecuperacao

# Configuração inicial do Unity Catalog via Databricks SDK
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import *

w = WorkspaceClient()

# Criar Metastore (uma vez por região)
metastore = w.metastores.create(
    name="metastore-brazil-south",
    storage_root="abfss://unity-catalog@adlsrecuperacao.dfs.core.windows.net/",
    region="brazilsouth"
)

# Criar Catalog para o projeto
catalog = w.catalogs.create(
    name="recuperacao_credito",
    comment="Dados de recuperação de crédito - Lakehouse Architecture"
)

# Criar Schemas (databases) para cada camada
w.schemas.create(
    name="bronze",
    catalog_name="recuperacao_credito",
    comment="Raw data landing zone"
)

w.schemas.create(
    name="silver",
    catalog_name="recuperacao_credito",
    comment="Cleaned and validated data"
)

w.schemas.create(
    name="gold",
    catalog_name="recuperacao_credito",
    comment="Business-ready aggregated data"
)

# Criar Service Principal para CI/CD
az ad sp create-for-rbac \
  --name "sp-databricks-cicd" \
  --role contributor \
  --scopes /subscriptions/{subscription-id}/resourceGroups/rg-lakehouse-prod

# Atribuir permissões no ADLS (Storage Blob Data Contributor)
az role assignment create \
  --assignee {service-principal-id} \
  --role "Storage Blob Data Contributor" \
  --scope /subscriptions/{subscription-id}/resourceGroups/rg-lakehouse-prod/providers/Microsoft.Storage/storageAccounts/adlsrecuperacao

# notebooks/bronze/ingest_sistema_cobranca.py
from pyspark.sql.functions import *
from pyspark.sql.types import *

# ===== FONTE 1: API REST do Sistema de Cobrança =====
# Problema: API retorna paginação, timeout frequente, rate limit de 100 req/min

def ingest_api_cobranca_incremental():
    """
    Ingere dados da API com retry exponencial e checkpoint
    """
    import requests
    from requests.adapters import HTTPAdapter
    from urllib3.util.retry import Retry
    
    # Configurar retry strategy
    retry_strategy = Retry(
        total=5,
        backoff_factor=2,  # 1s, 2s, 4s, 8s, 16s
        status_forcelist=[429, 500, 502, 503, 504]
    )
    
    session = requests.Session()
    session.mount("https://", HTTPAdapter(max_retries=retry_strategy))
    
    # Ler último checkpoint (última data ingerida)
    try:
        last_run = spark.sql("""
            SELECT MAX(data_extracao) as last_date 
            FROM recuperacao_credito.bronze.api_cobranca_raw
        """).collect()[0]['last_date']
    except:
        last_run = "2024-01-01"  # Data inicial se tabela não existe
    
    # API endpoint com filtro incremental
    base_url = "https://api-cobranca.empresa.com/v2/contratos"
    params = {
        "data_atualizacao_gte": last_run,
        "page_size": 1000
    }
    
    all_records = []
    page = 1
    
    while True:
        params['page'] = page
        response = session.get(base_url, params=params, timeout=30)
        response.raise_for_status()
        
        data = response.json()
        if not data['results']:
            break
            
        all_records.extend(data['results'])
        page += 1
        
        # Rate limiting: 100 req/min = 1.6 req/s
        time.sleep(0.6)
    
    # Converter para DataFrame
    schema = StructType([
        StructField("id_contrato", StringType(), False),
        StructField("cpf_cliente", StringType(), False),
        StructField("valor_original", DecimalType(15,2), False),
        StructField("valor_atualizado", DecimalType(15,2), False),
        StructField("data_vencimento", DateType(), False),
        StructField("status", StringType(), False),
        StructField("data_atualizacao", TimestampType(), False)
    ])
    
    df = spark.createDataFrame(all_records, schema=schema)
    
    # Adicionar metadados de auditoria
    df_bronze = df \
        .withColumn("data_extracao", current_timestamp()) \
        .withColumn("fonte", lit("api_sistema_cobranca")) \
        .withColumn("versao_api", lit("v2"))
    
    # Escrita idempotente com MERGE (evita duplicatas em reprocessamento)
    df_bronze.createOrReplaceTempView("bronze_temp")
    
    spark.sql("""
        MERGE INTO recuperacao_credito.bronze.api_cobranca_raw AS target
        USING bronze_temp AS source
        ON target.id_contrato = source.id_contrato 
           AND target.data_atualizacao = source.data_atualizacao
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

# ===== FONTE 2: Arquivos CSV diários via SFTP =====
# Problema: Schema muda sem aviso, arquivos corrompidos ocasionalmente

# Configuração do Autoloader para detecção automática
autoloader_config = {
    "cloudFiles.format": "csv",
    "cloudFiles.schemaLocation": "abfss://checkpoints@adlsrecuperacao.dfs.core.windows.net/schema/pagamentos",
    "cloudFiles.inferColumnTypes": "true",
    "cloudFiles.schemaEvolutionMode": "addNewColumns",  # Permite novas colunas automaticamente
    "cloudFiles.schemaHints": "valor_pago DECIMAL(15,2), data_pagamento DATE",
    "cloudFiles.maxFilesPerTrigger": 100,
    "cloudFiles.useNotifications": "true",  # Azure Event Grid para eficiência
    
    # Tratamento de arquivos corrompidos
    "badRecordsPath": "abfss://bronze@adlsrecuperacao.dfs.core.windows.net/bad_records/pagamentos",
    "mode": "PERMISSIVE"  # Linha com erro vira null, mas não quebra pipeline
}

df_pagamentos = (spark.readStream
    .format("cloudFiles")
    .options(**autoloader_config)
    .option("header", "true")
    .option("delimiter", ";")
    .option("encoding", "ISO-8859-1")  # Encoding comum no Brasil
    .load("abfss://landing@adlsrecuperacao.dfs.core.windows.net/pagamentos/")
    .withColumn("arquivo_origem", input_file_name())
    .withColumn("data_ingestao", current_timestamp())
)

# Escrita com checkpoint para garantir exactly-once semantics
query = (df_pagamentos.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "abfss://checkpoints@adlsrecuperacao.dfs.core.windows.net/bronze_pagamentos")
    .option("mergeSchema", "true")
    .trigger(availableNow=True)  # Processa todos disponíveis e para
    .table("recuperacao_credito.bronze.pagamentos_csv")
)

query.awaitTermination()

# ===== FONTE 3: Change Data Capture do Oracle =====
# Problema: Volume alto (1M+ rows/dia), necessidade de streaming

from pyspark.sql.functions import from_json, col

# Schema do CDC event do Oracle
cdc_schema = StructType([
    StructField("operation", StringType()),  # INSERT, UPDATE, DELETE
    StructField("timestamp", TimestampType()),
    StructField("table", StringType()),
    StructField("before", StringType()),  # JSON do registro anterior
    StructField("after", StringType())     # JSON do registro novo
])

# Ler do Azure Event Hub (conectado ao Oracle via Debezium)
df_cdc = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", dbutils.secrets.get("key-vault", "eventhub-connection"))
    .option("eventhubs.consumerGroup", "databricks-bronze")
    .option("maxEventsPerTrigger", 10000)
    .load()
    .select(from_json(col("body").cast("string"), cdc_schema).alias("data"))
    .select("data.*")
    .withColumn("data_ingestao_streaming", current_timestamp())
)

# Escrita em modo streaming (low latency)
(df_cdc.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "abfss://checkpoints@adlsrecuperacao.dfs.core.windows.net/bronze_cdc_oracle")
    .trigger(processingTime="30 seconds")  # Micro-batch a cada 30s
    .table("recuperacao_credito.bronze.cdc_oracle_clientes")
    .start()
)

# notebooks/silver/transform_cobranca_clean.py
import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import *

# ===== LIMPEZA E PADRONIZAÇÃO =====

@dlt.table(
    name="cobranca_cleaned",
    comment="Dados de cobrança limpos, deduplicados e validados",
    table_properties={
        "quality": "silver",
        "pipelines.autoOptimize.zOrderCols": "cpf_clean,data_vencimento"
    }
)
@dlt.expect_or_drop("cpf_valido", "length(cpf_clean) = 11 AND cpf_clean RLIKE '^[0-9]{11}$'")
@dlt.expect_or_drop("valor_positivo", "valor_divida_atualizado > 0")
@dlt.expect_or_drop("data_vencimento_valida", "data_vencimento >= '2020-01-01' AND data_vencimento <= current_date()")
@dlt.expect_or_fail("chave_unica", "id_contrato IS NOT NULL")  # Faz pipeline falhar se chave nula
def cobranca_cleaned():
    
    # UDF para validação de CPF (algoritmo oficial)
    @udf(returnType=BooleanType())
    def validar_cpf(cpf):
        if not cpf or len(cpf) != 11:
            return False
        
        # Verifica se não é sequência repetida (111.111.111-11)
        if cpf == cpf[0] * 11:
            return False
        
        # Validação dos dígitos verificadores
        def calcular_digito(cpf_parcial, peso_inicial):
            soma = sum(int(cpf_parcial[i]) * (peso_inicial - i) for i in range(len(cpf_parcial)))
            resto = soma % 11
            return 0 if resto < 2 else 11 - resto
        
        digito1 = calcular_digito(cpf[:9], 10)
        digito2 = calcular_digito(cpf[:10], 11)
        
        return cpf[-2:] == f"{digito1}{digito2}"
    
    return (
        dlt.read_stream("recuperacao_credito.bronze.api_cobranca_raw")
        
        # Limpeza de CPF: remove pontuação e valida
        .withColumn("cpf_clean", regexp_replace(col("cpf_cliente"), r'[.\-/]', ''))
        .withColumn("cpf_valido_flag", validar_cpf(col("cpf_clean")))
        
        # Padronização de valores monetários
        .withColumn("valor_divida_atualizado", 
                   when(col("valor_atualizado").isNull(), col("valor_original"))
                   .otherwise(col("valor_atualizado")))
        
        # Cálculo de aging (dias em atraso)
        .withColumn("dias_atraso", 
                   datediff(current_date(), col("data_vencimento")))
        
        # Classificação de risco baseada em aging
        .withColumn("faixa_risco",
                   when(col("dias_atraso") <= 30, "BAIXO")
                   .when(col("dias_atraso") <= 90, "MEDIO")
                   .when(col("dias_atraso") <= 180, "ALTO")
                   .otherwise("CRITICO"))
        
        # Deduplicação: em caso de múltiplos registros, pega o mais recente
        .withColumn("row_num", 
                   row_number().over(
                       Window.partitionBy("id_contrato")
                       .orderBy(col("data_atualizacao").desc())
                   ))
        .filter(col("row_num") == 1)
        .drop("row_num")
        
        # Metadados de processamento
        .withColumn("processado_em", current_timestamp())
        .withColumn("versao_pipeline", lit("v2.1"))
    )


# ===== ENRIQUECIMENTO COM DIMENSÕES =====

@dlt.table(
    name="cobranca_enriched",
    comment="Dados de cobrança enriquecidos com dados do cliente e localização"
)
@dlt.expect_or_drop("cliente_existe", "id_cliente IS NOT NULL")
def cobranca_enriched():
    
    df_cobranca = dlt.read("cobranca_cleaned")
    
    # Dimensão de clientes (SCD Type 2 - histórico de mudanças)
    df_clientes = dlt.read("recuperacao_credito.silver.dim_clientes")
    
    # Dimensão de localização (CEP -> Cidade/Estado/Região)
    df_localizacao = dlt.read("recuperacao_credito.silver.dim_localizacao")
    
    return (
        df_cobranca
        
        # Join com dimensão de clientes
        .join(
            df_clientes.filter(col("is_current") == True),  # Apenas registros atuais do SCD
            df_cobranca.cpf_clean == df_clientes.cpf,
            "left"
        )
        
        # Join com localização
        .join(
            df_localizacao,
            df_clientes.cep == df_localizacao.cep,
            "left"
        )
        
        # Selecionar colunas finais
        .select(
            # Chaves
            col("id_contrato"),
            col("cpf_clean").alias("cpf"),
            df_clientes.id_cliente,
            
            # Atributos de cobrança
            col("valor_divida_atualizado"),
            col("data_vencimento"),
            col("dias_atraso"),
            col("faixa_risco"),
            col("status"),
            
            # Atributos do cliente
            df_clientes.nome_cliente,
            df_clientes.data_nascimento,
            df_clientes.score_credito,
            df_clientes.renda_estimada,
            df_clientes.segmento,  # Varejo, Agro, Empresarial
            
            # Atributos de localização
            df_localizacao.cidade,
            df_localizacao.estado,
            df_localizacao.regiao,  # Norte, Nordeste, Sul, Sudeste, Centro-Oeste
            df_localizacao.tipo_cidade,  # Capital, Interior, Região Metropolitana
            
            # Metadados
            col("processado_em"),
            col("versao_pipeline")
        )
    )


# ===== QUALITY METRICS (Monitoramento de Qualidade) =====

@dlt.table(
    name="quality_metrics_cobranca",
    comment="Métricas de qualidade dos dados de cobrança"
)
def quality_metrics():
    
    df = dlt.read("cobranca_cleaned")
    
    return df.agg(
        count("*").alias("total_registros"),
        
        # Completude
        (sum(when(col("cpf_clean").isNull(), 1).otherwise(0)) / count("*") * 100).alias("pct_cpf_nulo"),
        (sum(when(col("data_vencimento").isNull(), 1).otherwise(0)) / count("*") * 100).alias("pct_data_nula"),
        
        # Validade
        (sum(when(col("cpf_valido_flag") == False, 1).otherwise(0)) / count("*") * 100).alias("pct_cpf_invalido"),
        (sum(when(col("valor_divida_atualizado") <= 0, 1).otherwise(0)) / count("*") * 100).alias("pct_valor_invalido"),
        
        # Unicidade
        (count("*") - countDistinct("id_contrato")).alias("total_duplicatas"),
        
        # Estatísticas descritivas
        avg("valor_divida_atualizado").alias("valor_medio"),
        stddev("valor_divida_atualizado").alias("valor_desvio_padrao"),
        percentile_approx("dias_atraso", 0.5).alias("dias_atraso_mediana"),
        
        # Timestamp de cálculo
        current_timestamp().alias("calculado_em")
    )

# notebooks/gold/kpis_recuperacao.py
import dlt
from pyspark.sql.functions import *
from pyspark.sql.window import Window

@dlt.table(
    name="kpi_recuperacao_mensal",
    comment="KPIs padronizados de recuperação de crédito - atualização D+1",
    table_properties={
        "quality": "gold",
        "pipelines.autoOptimize.zOrderCols": "ano_mes,regiao",
        "delta.enableChangeDataFeed": "true"  # Habilita CDC para downstream
    }
)
def kpi_recuperacao_mensal():
    
    df_cobranca = dlt.read("recuperacao_credito.silver.cobranca_enriched")
    df_pagamentos = dlt.read("recuperacao_credito.silver.pagamentos_enriched")
    
    # Definição padronizada de "Recuperação" (acordo entre áreas):
    # - Contrato com status PAGO ou ACORDO_ATIVO
    # - Pagamento recebido > 0 nos últimos 30 dias
    
    df_base = (
        df_cobranca
        .join(df_pagamentos, "id_contrato", "left")
        .withColumn("ano_mes", date_format(col("data_vencimento"), "yyyy-MM"))
        .withColumn("ano", year(col("data_vencimento")))
        .withColumn("mes", month(col("data_vencimento")))
    )
    
    return (
        df_base
        .groupBy("ano_mes", "ano", "mes", "regiao", "segmento", "faixa_risco")
        .agg(
            # === VOLUME ===
            count("id_contrato").alias("total_contratos"),
            countDistinct("cpf").alias("total_clientes_unicos"),
            
            # === VALORES ===
            sum("valor_divida_atualizado").alias("valor_total_carteira"),
            sum(when(col("status").isin("PAGO", "ACORDO_ATIVO"), 
                    col("valor_pago")).otherwise(0)).alias("valor_recuperado"),
            
            # === TAXAS (KPIs principais) ===
            # Taxa de Recuperação = (Contratos Recuperados / Total Contratos) * 100
            (sum(when(col("status").isin("PAGO", "ACORDO_ATIVO"), 1).otherwise(0)) 
             / count("*") * 100).alias("taxa_recuperacao_contratos_pct"),
            
            # Taxa de Recuperação Financeira = (Valor Recuperado / Valor Total) * 100
            (sum(when(col("status").isin("PAGO", "ACORDO_ATIVO"), col("valor_pago")).otherwise(0))
             / sum("valor_divida_atualizado") * 100).alias("taxa_recuperacao_financeira_pct"),
            
            # === AGING ===
            avg("dias_atraso").alias("dias_atraso_medio"),
            percentile_approx("dias_atraso", 0.5).alias("dias_atraso_mediana"),
            max("dias_atraso").alias("dias_atraso_maximo"),
            
            # === DISTRIBUIÇÃO POR FAIXA ===
            sum(when(col("faixa_risco") == "BAIXO", 1).otherwise(0)).alias("contratos_risco_baixo"),
            sum(when(col("faixa_risco") == "MEDIO", 1).otherwise(0)).alias("contratos_risco_medio"),
            sum(when(col("faixa_risco") == "ALTO", 1).otherwise(0)).alias("contratos_risco_alto"),
            sum(when(col("faixa_risco") == "CRITICO", 1).otherwise(0)).alias("contratos_risco_critico"),
            
            # === METADADOS ===
            current_timestamp().alias("atualizado_em"),
            lit("v1.0").alias("versao_metrica")
        )
    )


# ===== ANÁLISE DE TENDÊNCIA (MoM e YoY) =====

@dlt.table(
    name="kpi_recuperacao_tendencias",
    comment="Análise de tendências com comparação mês a mês e ano a ano"
)
def kpi_recuperacao_tendencias():
    
    df_kpi = dlt.read("kpi_recuperacao_mensal")
    
    # Window specs para cálculos de tendência
    window_mom = Window.partitionBy("regiao", "segmento").orderBy("ano_mes")
    window_yoy = Window.partitionBy("regiao", "segmento", "mes").orderBy("ano")
    
    return (
        df_kpi
        
        # Valores do mês anterior (MoM)
        .withColumn("taxa_recuperacao_mes_anterior", 
                   lag("taxa_recuperacao_contratos_pct", 1).over(window_mom))
        .withColumn("variacao_mom_pct",
                   ((col("taxa_recuperacao_contratos_pct") - col("taxa_recuperacao_mes_anterior")) 
                    / col("taxa_recuperacao_mes_anterior") * 100))
        
        # Valores do mesmo mês no ano anterior (YoY)
        .withColumn("taxa_recuperacao_ano_anterior",
                   lag("taxa_recuperacao_contratos_pct", 1).over(window_yoy))
        .withColumn("variacao_yoy_pct",
                   ((col("taxa_recuperacao_contratos_pct") - col("taxa_recuperacao_ano_anterior"))
                    / col("taxa_recuperacao_ano_anterior") * 100))
        
        # Classificação de performance
        .withColumn("classificacao_performance",
                   when(col("variacao_mom_pct") > 10, "EXCELENTE")
                   .when(col("variacao_mom_pct") > 0, "BOM")
                   .when(col("variacao_mom_pct") > -10, "NEUTRO")
                   .otherwise("CRÍTICO"))
        
        # Médias móveis (MA - Moving Average)
        .withColumn("taxa_recuperacao_ma3",  # Média dos últimos 3 meses
                   avg("taxa_recuperacao_contratos_pct").over(
                       window_mom.rowsBetween(-2, 0)))
        .withColumn("taxa_recuperacao_ma6",  # Média dos últimos 6 meses
                   avg("taxa_recuperacao_contratos_pct").over(
                       window_mom.rowsBetween(-5, 0)))
    )

-- ===== VACUUM: Limpeza de versões antigas =====
VACUUM recuperacao_credito.silver.cobranca_enriched RETAIN 168 HOURS;

-- ===== Z-ORDERING: Colocação de dados relacionados =====
OPTIMIZE recuperacao_credito.gold.kpi_recuperacao_mensal
ZORDER BY (ano_mes, regiao, segmento);

-- ===== LIQUID CLUSTERING: Alternativa moderna ao Z-Ordering =====
ALTER TABLE recuperacao_credito.gold.kpi_recuperacao_mensal
CLUSTER BY (ano_mes, regiao);

-- ===== TIME TRAVEL: Audit e recuperação =====
SELECT * FROM recuperacao_credito.gold.kpi_recuperacao_mensal VERSION AS OF 42;

-- ===== CONTROLE DE ACESSO GRANULAR =====
CREATE GROUP IF NOT EXISTS analistas_recuperacao;
CREATE GROUP IF NOT EXISTS cientistas_dados;
CREATE GROUP IF NOT EXISTS engenheiros_dados;

-- Permissões em nível de Schema
GRANT SELECT ON SCHEMA recuperacao_credito.gold TO GROUP analistas_recuperacao;
GRANT SELECT ON SCHEMA recuperacao_credito.silver TO GROUP cientistas_dados;
GRANT ALL PRIVILEGES ON CATALOG recuperacao_credito TO GROUP engenheiros_dados;

-- ===== MASCARAMENTO DE DADOS SENSÍVEIS =====
CREATE FUNCTION recuperacao_credito.mask_cpf(cpf STRING)
RETURNS STRING
RETURN CASE
    WHEN is_member('gestores_c_level') THEN cpf
    ELSE CONCAT(SUBSTRING(cpf, 1, 3), '*****', SUBSTRING(cpf, 9, 3))
END;

-- ===== AUDIT LOG =====
SELECT
    user_identity.email,
    event_time,
    action_name,
    request_params.full_name_arg AS tabela_acessada
FROM system.access.audit
WHERE action_name IN ('getTable', 'commandSubmit')
  AND request_params.full_name_arg LIKE 'recuperacao_credito.gold%'
  AND event_date >= current_date() - INTERVAL 7 DAYS
ORDER BY event_time DESC;

# databricks.yml
bundle:
  name: lakehouse-recuperacao-credito

environments:
  dev:
    default: true
    workspace:
      host: https://adb-123456789.azuredatabricks.net
    variables:
      catalog_name: "recuperacao_credito_dev"
  
  prod:
    workspace:
      host: https://adb-111222333.azuredatabricks.net
    variables:
      catalog_name: "recuperacao_credito"

resources:
  pipelines:
    silver_pipeline:
      name: "[${bundle.environment}] Silver - Cobrança e Clientes"
      target: "${var.catalog_name}.silver"
      libraries:
        - notebook:
            path: ./notebooks/silver/transform_cobranca_clean.py
      
      clusters:
        - label: "default"
          node_type_id: "Standard_DS3_v2"
          autoscale:
            min_workers: 2
            max_workers: 8
      
      continuous: false
      
      notifications:
        - email_recipients:
            - engenharia-dados@empresa.com
          on_failure: true
  
  jobs:
    daily_ingestion:
      name: "[${bundle.environment}] Ingestão Diária - Cobrança"
      
      tasks:
        - task_key: ingest_bronze
          notebook_task:
            notebook_path: ./notebooks/bronze/ingest_api_cobranca.py
          job_cluster_key: "shared_cluster"
        
        - task_key: run_silver_pipeline
          depends_on:
            - task_key: ingest_bronze
          pipeline_task:
            pipeline_id: ${resources.pipelines.silver_pipeline.id}
      
      schedule:
        quartz_cron_expression: "0 0 6 * * ?"
        timezone_id: "America/Sao_Paulo"

# .github/workflows/deploy-prod.yml
name: Deploy to Production

on:
  push:
    branches:
      - main

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Setup Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Run unit tests
        run: pytest tests/unit/ -v --cov=notebooks
  
  deploy:
    runs-on: ubuntu-latest
    needs: test
    environment: production
    
    steps:
      - uses: actions/checkout@v3
      
      - name: Deploy bundle
        run: databricks bundle deploy -e prod
      
      - name: Notify Slack
        if: success()
        run: |
          curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
          -d '{"text":"✅ Deploy em PRODUÇÃO concluído!"}'

┌─────────────────────────────────────────────────────────────────┐
│                    FONTES DE DADOS                              │
│  • APIs REST (sistemas de cobrança)                             │
│  • Bancos SQL (Oracle, SQL Server)                              │
│  • Arquivos (CSV, JSON, Parquet)                                │
│  • Event Streaming (Azure Event Hub)                            │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────────┐
│                INGESTÃO (Azure Data Factory)                    │
│  • Copy Activity para batch                                     │
│  • Event Hub Trigger para streaming                             │
│  • Autoloader para file ingestion                               │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────────┐
│              BRONZE LAYER (ADLS Gen2 + Delta Lake)              │
│  • Dados brutos preservados                                     │
│  • Schema evolution habilitado                                  │
│  • Auditoria: timestamp + source file                           │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────────┐
│       SILVER LAYER (Delta Live Tables + Spark Structured)       │
│  • Limpeza e deduplicação                                       │
│  • Validações de qualidade (@dlt.expect)                        │
│  • Enriquecimento (joins com dimensões)                         │
└─────────────────────┬───────────────────────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────────────────────┐
│          GOLD LAYER (Databricks SQL Warehouse)                  │
│  • Agregações e KPIs                                            │
│  • Otimizações: Z-Order, Data Skipping                          │
│  • Semantic Layer para BI                                       │
└─────────────────────┬───────────────────────────────────────────┘
                      │
        ┌─────────────┴─────────────┐
        │                           │
        ▼                           ▼
┌──────────────────┐      ┌──────────────────┐
│   POWER BI       │      │   ML/AI          │
│  (Direct Lake)   │      │  • MLflow        │
│  • Dashboards    │      │  • Feature Store │
│  • Alerts        │      │  • Predição      │
└──────────────────┘      └──────────────────┘

          GOVERNANÇA (Unity Catalog + Azure Purview)
          • Data Lineage end-to-end
          • Access Control (RBAC/ABAC)
          • PII Classification