diff --git a/dbt_project.yml b/dbt_project.yml index 6656b49c..f406aba7 100644 --- a/dbt_project.yml +++ b/dbt_project.yml @@ -144,6 +144,10 @@ vars: rho_max_processing_interval: 7 rho_sppo_start_date: "2021-03-01" + + ### TG #### + tg_transacao_staging: "`rj-smtr-staging.br_rj_riodejaneiro_onibus_tg.card_transactions`" + tg_dicionario_staging: "`rj-smtr-staging.br_rj_riodejaneiro_onibus_tg.dicionario`" ### Bilhetagem ### var_percentual_tarifa_cbd: 0.04 quantidade_integracoes_max: 5 @@ -220,6 +224,9 @@ models: br_rj_riodejaneiro_rdo: +materialized: view +schema: br_rj_riodejaneiro_rdo + br_rj_riodejaneiro_onibus_tg: + +materialized: view + +schema: br_rj_riodejaneiro_onibus_tg br_rj_riodejaneiro_rdo_staging: +materialized: view +schema: br_rj_riodejaneiro_rdo_staging diff --git a/models/br_rj_riodejaneiro_onibus_tg/schema.yaml b/models/br_rj_riodejaneiro_onibus_tg/schema.yaml new file mode 100644 index 00000000..588ee5c7 --- /dev/null +++ b/models/br_rj_riodejaneiro_onibus_tg/schema.yaml @@ -0,0 +1,49 @@ +version: 2 + +models: + - name: tg_transacao + description: "Microdados das transações de cartão. Cada registro é uma transação." + columns: + - name: data + description: "Data da transação (partição)" + - name: id_empresa + description: "Código da empresa que opera o veículo de origem da transação" + - name: id_veiculo + description: "Número do veículo configurado no equipamento (apenas número)" + - name: id_cartao + description: "Número único criptografado do cartão RioCard" + - name: tipo_cartao + description: "Emissor e número do aplicação, indica qual o tipo do cartão (ex: gratuidade, vale transporte). Categorias podem ser encontradas na tabela dicionário." + - name: sequencial_transacao_cartao + description: "Número sequencial histórico de transações do cartão" + - name: datetime + description: "Datetime da transação em GMT-3" + - name: tipo_embarque + description: "Tipo de embarque atual, indica se é integração ou não" + - name: tipo_debito + description: "Tipo de débito no cartão, indica o tipo de tarifa executada (ex: cheia, com desconto). Categorias podem ser encontradas na tabela dicionário." + - name: mensagem_debito + description: "Mensagem de estado do débito no cartão. Categorias podem ser encontradas na tabela dicionário." + - name: tarifa + description: "Valor da tarifa configurado no equipamento (R$)" + - name: tarifa_anterior + description: "Valor da tarifa anterior da viagem anterior (R$)" + - name: debito + description: "Valor debitado na viagem atual (R$)" + - name: desconto + description: "Valor promocional ou de desconto da tarifa (R$)" + - name: total_integracao + description: "Soma do valor total pago em uma integração (R$)" + - name: tg_dicionario + description: "Dicionário de dados" + columns: + - name: chave + description: "Chave" + - name: cobertura_temporal + description: "Cobertura temporal" + - name: id_tabela + description: "Nome da tabela" + - name: nome_coluna + description: "Nome da coluna" + - name: valor + description: "Valor" \ No newline at end of file diff --git a/models/br_rj_riodejaneiro_onibus_tg/tg_dicionario.sql b/models/br_rj_riodejaneiro_onibus_tg/tg_dicionario.sql new file mode 100644 index 00000000..83f4354b --- /dev/null +++ b/models/br_rj_riodejaneiro_onibus_tg/tg_dicionario.sql @@ -0,0 +1,15 @@ +{{ + config( + alias="dicionario", + materialized="table", + ) +}} + +SELECT + SAFE_CAST(chave AS STRING) AS chave, + SAFE_CAST(cobertura_temporal AS STRING) AS cobertura_temporal, + SAFE_CAST(id_tabela AS STRING) AS id_tabela, + SAFE_CAST(nome_coluna AS STRING) AS nome_coluna, + SAFE_CAST(valor AS STRING) AS valor +FROM + {{var('tg_dicionario_staging')}} \ No newline at end of file diff --git a/models/br_rj_riodejaneiro_onibus_tg/tg_transacao.sql b/models/br_rj_riodejaneiro_onibus_tg/tg_transacao.sql new file mode 100644 index 00000000..96133db6 --- /dev/null +++ b/models/br_rj_riodejaneiro_onibus_tg/tg_transacao.sql @@ -0,0 +1,104 @@ +{{ + config( + alias="transacao", + materialized="incremental", + partition_by={"field": "data", "data_type": "date", "granularity": "day"}, + incremental_strategy="insert_overwrite", + ) +}} + +WITH + tg AS ( + SELECT + SAFE_CAST(DATE_ADD(DATE '2002-12-31', INTERVAL data DAY) AS DATE) AS data, + SAFE_CAST(codigo_empresa AS STRING) AS id_empresa, + SAFE_CAST(num_onibus AS STRING) AS id_veiculo, + SAFE_CAST(uid AS STRING) AS id_cartao, + SAFE_CAST(emissor_aplicacao || "_" || aplicacao AS STRING) AS tipo_cartao, + SAFE_CAST(tsn AS STRING) AS sequencial_transacao_cartao, + SAFE_CAST(DATETIME(TIMESTAMP_ADD(TIMESTAMP(DATE_ADD(DATE '2002-12-31', INTERVAL data DAY)), INTERVAL hora SECOND)) AS DATETIME) AS datetime, + SAFE_CAST(tipo_embarque AS STRING) AS tipo_embarque, + SAFE_CAST(SAFE_CAST(tipo_debito AS FLOAT64) AS STRING) AS tipo_debito, + SAFE_CAST(SAFE_CAST(mensagem_debito AS FLOAT64) AS STRING) AS mensagem_debito, + SAFE_CAST(valor_tarifa/100 AS FLOAT64) AS tarifa, + SAFE_CAST(valor_tarifa_anterior/100 AS FLOAT64) AS tarifa_anterior, + SAFE_CAST(valor_debitado/100 AS FLOAT64) AS debito, + SAFE_CAST(SAFE_CAST(valor_promo_desconto AS INT64)/100 AS FLOAT64) AS desconto, + SAFE_CAST(SAFE_CAST(valor_acumulado AS INT64)/100 AS FLOAT64) AS total_integracao, + SAFE_CAST(garagem AS STRING) AS garagem + FROM + {{ var('tg_transacao_staging') }} + WHERE + {% if var("date_range_start") == "None" %} + + {% set date_range_start = "2022-06-01" %} + + {% else %} + + {% set date_range_start = var("date_range_start") %} + + {% endif %} + + ano BETWEEN EXTRACT(YEAR FROM DATE("{{ date_range_start }}")) AND EXTRACT(YEAR FROM DATE("{{ var("date_range_end") }}")) + AND status = 0 -- Apenas transações com sucesso + AND (emissor_aplicacao BETWEEN 1 AND 15) -- Range de emissores de cartão válidos (fonte: RioCard) + AND (aplicacao BETWEEN 0 AND 1023) -- Range de aplicações válidas (fonte: RioCard) + + ), + tg_filtra_data AS ( + SELECT + * + FROM + tg + WHERE + data BETWEEN DATE("{{ date_range_start }}") AND DATE("{{ var("date_range_end") }}") + ), + tg_soma_debito AS ( + SELECT + data, + id_empresa, + id_veiculo, + id_cartao, + tipo_cartao, + sequencial_transacao_cartao, + datetime, + tipo_embarque, + tipo_debito, + mensagem_debito, + tarifa, + tarifa_anterior, + ROUND(SUM(debito), 2) AS debito, + desconto, + total_integracao, + garagem + FROM + tg_filtra_data + GROUP BY + data, + id_empresa, + id_veiculo, + id_cartao, + tipo_cartao, + sequencial_transacao_cartao, + datetime, + tipo_embarque, + tipo_debito, + mensagem_debito, + tarifa, + tarifa_anterior, + desconto, + total_integracao, + garagem), + tg_rn_garagem AS ( + SELECT + *, + ROW_NUMBER() OVER (PARTITION BY datetime, id_cartao ) AS rn_garagem + FROM + tg_soma_debito ) +SELECT + * EXCEPT(garagem, + rn_garagem) +FROM + tg_rn_garagem +WHERE + rn_garagem = 1 \ No newline at end of file