[WIP] Adiciona dados de br_rj_riodejaneiro_onibus_tg

dbt_project.yml

@@ -130,6 +130,11 @@ 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`"
+
 models:
   +persist_docs:
     relation: true
@@ -159,3 +164,6 @@ 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

models/br_rj_riodejaneiro_onibus_tg/schema.yaml

@@ -0,0 +1,63 @@
+version: 2
+
+models:
+  - name: tg_transacao
+    description: "Microdados das transações de cartão. Cada registro é uma transação."
+    columns:
+      - name: tipo_registro
+        description: "Tipo de registro de acordo com a especificação TG/PRODATA (Valor Fixo: 002)"
+      - name: versao_registro
+        description: "Versão deste registro"
+      - name: timestamp_transacao
+        description: "Timestamp da transação em GMT-3"
+      - name: numero_aplicacao
+        description: "Número do aplicação"
+      - name: sequencial_transacao
+        description: "Número sequencial da transação no veículo naquele dia"
+      - name: valor_tarifa
+        description: "Valor da tarifa configurado no equipamento (R$)"
+      - name: valor_tarifa_anterior
+        description: "Valor da tarifa anterior da viagem anterior (R$)"
+      - name: valor_debitado
+        description: "Valor debitado na viagem atual (R$)"
+      - name: status
+        description: "Estado da transação, indicando falha ou sucesso"
+      - name: id_veiculo
+        description: "Número do veículo configurado no equipamento (apenas número)"
+      - name: tipo_embarque
+        description: "Tipo de embarque atual, indica se é integração ou não"
+      - name: diferenca_valor_debitado
+        description: "Diferença do valor debitado na transação (R$)"
+      - name: valor_promo_desconto
+        description: "Valor promocional ou de desconto da tarifa (R$)"
+      - name: valor_acumulado
+        description: "Soma do valor total pago em uma integração (R$)"
+      - name: tipo_debito
+        description: "Tipo de débito no cartão"
+      - name: mensagem_debito
+        description: "Mensagem de estado do débito no cartão"
+      - name: arquivo_origem
+        description: "Nome do arquivo txt de origem do registro: TG[Código da Empresa]_[id_veiculo]_[Número do Validador]_[Número de Controle Sequencial]_[Código da Empresa]_[Timestamp de geração do arquivo]_[Versão do Validador (Software) - XX.XX]"
+      - name: garagem
+        description: "Garagem onde o arquivo de origem foi descarregado"
+      - name: data_arquivo_origem
+        description: "Data de criação do arquivo binário de origem extraída do nome do arquivo"
+      - name: codigo_empresa
+        description: "Código da empresa que opera o veículo de origem da transação"
+      - name: id_cartao
+        description: "Número único criptografado do cartão RioCard"
+      - name: data
+        description: "Data da transação (partição)"
+  - 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"

models/br_rj_riodejaneiro_onibus_tg/tg_dicionario.sql

@@ -0,0 +1,12 @@
+{{ 
+  config(alias='dicionario') 
+}}
+
+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')}}

models/br_rj_riodejaneiro_onibus_tg/tg_transacao.sql

@@ -0,0 +1,62 @@
+{{ 
+  config( 
+    alias="transacao",
+    materialized="incremental",
+    partition_by={"field": "data", "data_type": "date", "granularity": "day"},
+    incremental_strategy="insert_overwrite",
+  ) 
+}}
+
+SELECT
+  SAFE_CAST(tipo_registro AS STRING) AS tipo_registro,
+  SAFE_CAST(versao_registro AS STRING) AS versao_registro,
+  SAFE_CAST(DATETIME(TIMESTAMP_ADD(TIMESTAMP(DATE_ADD(DATE '2002-12-31', INTERVAL data DAY)), INTERVAL hora SECOND), "America/Sao_Paulo") AS DATETIME) AS timestamp_transacao,
+  SAFE_CAST(emissor_aplicacao || "_" || aplicacao AS STRING) AS numero_aplicacao,
+  SAFE_CAST(tsn AS INT64) AS sequencial_transacao,
+  SAFE_CAST(status AS INT64) AS status,
+  SAFE_CAST(num_onibus AS STRING) AS id_veiculo,
+  SAFE_CAST(tipo_embarque AS INT64) AS tipo_embarque,
+  SAFE_CAST(SAFE_CAST(tipo_debito AS FLOAT64) AS INT64) AS tipo_debito,
+  SAFE_CAST(SAFE_CAST(mensagem_debito AS FLOAT64) AS INT64) AS mensagem_debito,
+  SAFE_CAST(valor_tarifa/100 AS FLOAT64) AS valor_tarifa,
+  SAFE_CAST(valor_tarifa_anterior/100 AS FLOAT64) AS valor_tarifa_anterior,
+  SAFE_CAST(valor_debitado/100 AS FLOAT64) AS valor_debitado,
+  SAFE_CAST(COALESCE(SAFE_CAST(diferenca_valor_debitado AS INT64)/100, 0) AS FLOAT64) AS diferenca_valor_debitado,
+  SAFE_CAST(SAFE_CAST(valor_promo_desconto AS INT64)/100 AS FLOAT64) AS valor_promo_desconto,
+  SAFE_CAST(SAFE_CAST(valor_acumulado AS INT64)/100 AS FLOAT64) AS valor_acumulado,
+  SAFE_CAST(origin_file AS STRING) AS arquivo_origem,
+  SAFE_CAST(garagem AS STRING) AS garagem,
+  SAFE_CAST(file_date AS STRING) AS data_arquivo_origem,
+  SAFE_CAST(codigo_empresa AS INT64) AS codigo_empresa,
+  SAFE_CAST(uid AS STRING) AS id_cartao,
+  SAFE_CAST(DATE_ADD(DATE '2002-12-31', INTERVAL data DAY) AS DATE) AS data
+FROM
+  {{var('tg_transacao_staging')}}
+WHERE
+  (ano BETWEEN 2022 AND 2023) -- Apenas dados de 2022 e 2023
+  AND mes >= 1
+  AND dia >= 1
+  AND status = 0 -- Apenas transações com sucesso
+  AND (emissor_aplicacao BETWEEN 1 AND 15)
+  AND (aplicacao BETWEEN 0 AND 1023)
+  AND dia <= EXTRACT(DAY FROM DATE("{{ var("date_range_end") }}"))
+  AND mes <= EXTRACT(MONTH FROM DATE("{{ var("date_range_end") }}"))
+  AND ano <= EXTRACT(YEAR FROM DATE("{{ var("date_range_end") }}"))
+
+  {% if is_incremental() %}
+
+  {% if var("date_range_start") == "None" %}
+
+  {% set date_range_start = run_query("SELECT gr FROM (SELECT IF(MAX(data) > DATE('" ~ var("date_range_end") ~ "'), DATE('" ~ var("date_range_end") ~ "'), MAX(data)) AS gr FROM " ~ this ~ ")").columns[0].values()[0] %}
+
+  {% else %}
+
+  {% set date_range_start = var("date_range_start") %}
+
+  {% endif %}
+
+    AND dia >= EXTRACT(DAY FROM DATE("{{ date_range_start }}"))
+    AND mes >= EXTRACT(MONTH FROM DATE("{{ date_range_start }}"))
+    AND ano >= EXTRACT(YEAR FROM DATE("{{ date_range_start }}"))
+
+  {% endif %}