migra br_rj_riodejaneiro_bilhetagem para bilhetagem

prefeitura-rio · Jan 8, 2025 · 6904e39 · 6904e39
1 parent 3764d41
commit 6904e39
Show file tree

Hide file tree

Showing 29 changed files with 287 additions and 144 deletions.
diff --git a/...rj_riodejaneiro_bilhetagem/dicionario.sql → ...dels/bilhetagem/dicionario_bilhetagem.sql b/...rj_riodejaneiro_bilhetagem/dicionario.sql → ...dels/bilhetagem/dicionario_bilhetagem.sql
@@ -1,6 +1,7 @@
 {{
   config(
     materialized="table",
+    alias="dicionario",
   )
 }}
 SELECT
@@ -10,4 +11,4 @@ SELECT
   SAFE_CAST(coluna AS STRING) AS coluna,
   SAFE_CAST(valor AS STRING) AS valor
 FROM
-  {{ source("br_rj_riodejaneiro_bilhetagem_staging", "dicionario") }}
+  {{ source("source_jae", "dicionario") }}
diff --git a/...iro_bilhetagem_staging/gratuidade_aux.sql → ...els/bilhetagem/staging/aux_gratuidade.sql b/...iro_bilhetagem_staging/gratuidade_aux.sql → ...els/bilhetagem/staging/aux_gratuidade.sql
@@ -24,6 +24,8 @@ WITH gratuidade_complete_partitions AS (
         CAST(CAST(cd_cliente AS FLOAT64) AS INT64) AS id_cliente,
         id AS id_gratuidade,
         tipo_gratuidade,
+        deficiencia_permanente,
+        rede_ensino,
         data_inclusao AS data_inicio_validade,
         timestamp_captura
     FROM
@@ -47,6 +49,8 @@ SELECT
     id_cliente,
     id_gratuidade,
     tipo_gratuidade,
+    deficiencia_permanente,
+    rede_ensino,
     data_inicio_validade,
     LEAD(data_inicio_validade) OVER (PARTITION BY id_cliente ORDER BY data_inicio_validade) AS data_fim_validade,
     timestamp_captura

diff --git a/queries/models/bilhetagem/staging/aux_transacao_particao.sql b/queries/models/bilhetagem/staging/aux_transacao_particao.sql
@@ -0,0 +1,84 @@
+{{
+    config(
+        materialized="incremental",
+        partition_by={"field": "data", "data_type": "date", "granularity": "day"},
+        incremental_strategy="insert_overwrite",
+    )
+}}
+
+with
+    transacao as (
+        select distinct
+            data, hora, timestamp_captura, date(data_transacao) as data_transacao
+        from {{ ref("staging_transacao") }}
+        {% if is_incremental() %}
+            where
+                {{
+                    generate_date_hour_partition_filter(
+                        var("date_range_start"), var("date_range_end")
+                    )
+                }}
+                and timestamp_captura
+                between datetime("{{var('date_range_start')}}") and datetime(
+                    "{{var('date_range_end')}}"
+                )
+        {% endif %}
+    ),
+    integracao as (
+        select distinct
+            data,
+            extract(hour from timestamp_captura) as hora,
+            timestamp_captura,
+            date(data_transacao) as data_transacao
+        from
+            {{ ref("staging_integracao_transacao") }},
+            unnest(
+                [
+                    data_transacao_t0,
+                    data_transacao_t1,
+                    data_transacao_t2,
+                    data_transacao_t3,
+                    data_transacao_t4
+                ]
+            ) as data_transacao
+        where
+            data_transacao is not null
+        {% if is_incremental() %}
+                and date(data) between date("{{var('date_range_start')}}") and date(
+                    "{{var('date_range_end')}}"
+                )
+                and timestamp_captura
+                between datetime("{{var('date_range_start')}}") and datetime(
+                    "{{var('date_range_end')}}"
+                )
+        {% endif %}
+    ),
+    novos_dados as (
+        select *
+        from transacao
+
+        union distinct
+
+        select *
+        from integracao
+    ),
+    particao_completa as (
+        select
+           data, hora, timestamp_captura, array_agg(data_transacao) as particoes , 0 as priority
+        from novos_dados
+        group by 1, 2, 3
+
+        {% if is_incremental() %}
+            union all
+
+            select
+                *, 1 as priority
+            from {{ this }}
+        {% endif %}
+    )
+    select
+        * except(priority)
+    from particao_completa
+    qualify
+        row_number() over (partition by data, hora, timestamp_captura order by priority) = 1
+
diff --git a/...ro_bilhetagem_staging/staging_cliente.sql → ...ls/bilhetagem/staging/staging_cliente.sql b/...ro_bilhetagem_staging/staging_cliente.sql → ...ls/bilhetagem/staging/staging_cliente.sql
@@ -21,7 +21,7 @@ WITH
             SAFE_CAST(JSON_VALUE(content, '$.NR_TELEFONE') AS STRING) AS nr_telefone,
             SAFE_CAST(JSON_VALUE(content, '$.DT_CADASTRO') AS STRING) AS dt_cadastro
         FROM
-            {{ source("br_rj_riodejaneiro_bilhetagem_staging", "cliente") }}
+            {{ source("source_jae", "cliente") }}
     ),
     cliente_rn AS (
         SELECT

diff --git a/...o_bilhetagem_staging/staging_endereco.sql → ...s/bilhetagem/staging/staging_endereco.sql b/...o_bilhetagem_staging/staging_endereco.sql → ...s/bilhetagem/staging/staging_endereco.sql
@@ -22,7 +22,7 @@ WITH
             SAFE_CAST(JSON_VALUE(content, '$.TX_COMPLEMENTO_LOGRADOURO') AS STRING) AS tx_complemento_logradouro,
             SAFE_CAST(JSON_VALUE(content, '$.TX_LOGRADOURO') AS STRING) AS tx_logradouro
         FROM
-            {{ source("br_rj_riodejaneiro_bilhetagem_staging", "endereco") }}
+            {{ source("source_jae", "endereco") }}
     ),
     endereco_rn AS (
         SELECT

diff --git a/...bilhetagem_staging/staging_gratuidade.sql → ...bilhetagem/staging/staging_gratuidade.sql b/...bilhetagem_staging/staging_gratuidade.sql → ...bilhetagem/staging/staging_gratuidade.sql
@@ -12,7 +12,10 @@ SELECT
     DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%E*S%Ez', SAFE_CAST(JSON_VALUE(content, '$.data_inclusao') AS STRING)), 'America/Sao_Paulo') AS data_inclusao,
     SAFE_CAST(JSON_VALUE(content, '$.id_status_gratuidade') AS STRING) AS id_status_gratuidade,
     SAFE_CAST(JSON_VALUE(content, '$.id_tipo_gratuidade') AS STRING) AS id_tipo_gratuidade,
-    SAFE_CAST(JSON_VALUE(content, '$.tipo_gratuidade') AS STRING) AS tipo_gratuidade
+    SAFE_CAST(JSON_VALUE(content, '$.tipo_gratuidade') AS STRING) AS tipo_gratuidade,
+    SAFE_CAST(JSON_VALUE(content, '$.deficiencia_permanente') AS bool) AS deficiencia_permanente,
+    SAFE_CAST(JSON_VALUE(content, '$.rede_ensino') AS STRING) AS rede_ensino,
+
 FROM
-  {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'gratuidade') }}
+  {{ source('source_jae', 'gratuidade') }}
 
diff --git a/..._staging/staging_integracao_transacao.sql → .../staging/staging_integracao_transacao.sql b/..._staging/staging_integracao_transacao.sql → .../staging/staging_integracao_transacao.sql
@@ -31,4 +31,4 @@ SELECT
   SAFE_CAST(JSON_VALUE(content, '$.valor_transacao_total') AS FLOAT64) AS valor_transacao_total,
   SAFE_CAST(JSON_VALUE(content, '$.tx_adicional') AS STRING) AS tx_adicional
 FROM
-  {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'integracao_transacao') }}
+  {{ source('source_jae', 'integracao_transacao') }}
diff --git a/...eiro_bilhetagem_staging/staging_linha.sql → ...dels/bilhetagem/staging/staging_linha.sql b/...eiro_bilhetagem_staging/staging_linha.sql → ...dels/bilhetagem/staging/staging_linha.sql
@@ -28,7 +28,7 @@ WITH
             SAFE_CAST(JSON_VALUE(content, '$.GTFS_ROUTE_ID') AS STRING) AS gtfs_route_id,
             SAFE_CAST(JSON_VALUE(content, '$.GTFS_STOP_ID') AS STRING) AS gtfs_stop_id
         FROM
-            {{ source("br_rj_riodejaneiro_bilhetagem_staging", "linha") }}
+            {{ source("source_jae", "linha") }}
     ),
     linha_rn AS (
         SELECT

diff --git a/...tagem_staging/staging_linha_consorcio.sql → ...tagem/staging/staging_linha_consorcio.sql b/...tagem_staging/staging_linha_consorcio.sql → ...tagem/staging/staging_linha_consorcio.sql
@@ -14,7 +14,7 @@ WITH linha_consorcio AS (
     PARSE_DATE("%Y-%m-%d", SAFE_CAST(JSON_VALUE(content, '$.DT_INICIO_VALIDADE') AS STRING)) AS dt_inicio_validade,
     PARSE_DATE("%Y-%m-%d", SAFE_CAST(JSON_VALUE(content, '$.DT_FIM_VALIDADE') AS STRING)) AS dt_fim_validade
   FROM
-    {{ source("br_rj_riodejaneiro_bilhetagem_staging", "linha_consorcio") }}
+    {{ source("source_jae", "linha_consorcio") }}
 ),
 linha_consorcio_rn AS (
   SELECT

diff --git a/..._linha_consorcio_operadora_transporte.sql → ..._linha_consorcio_operadora_transporte.sql b/..._linha_consorcio_operadora_transporte.sql → ..._linha_consorcio_operadora_transporte.sql
@@ -15,7 +15,7 @@ WITH linha_consorcio_operadora_transporte AS (
     PARSE_DATE("%Y-%m-%d", SAFE_CAST(JSON_VALUE(content, '$.DT_INICIO_VALIDADE') AS STRING)) AS dt_inicio_validade,
     PARSE_DATE("%Y-%m-%d", SAFE_CAST(JSON_VALUE(content, '$.DT_FIM_VALIDADE') AS STRING)) AS dt_fim_validade
   FROM
-    {{ source("br_rj_riodejaneiro_bilhetagem_staging", "linha_consorcio_operadora_transporte") }}
+    {{ source("source_jae", "linha_consorcio_operadora_transporte") }}
 ),
 linha_consorcio_operadora_transporte_rn AS (
   SELECT

diff --git a/...aging/staging_linha_sem_ressarcimento.sql → ...aging/staging_linha_sem_ressarcimento.sql b/...aging/staging_linha_sem_ressarcimento.sql → ...aging/staging_linha_sem_ressarcimento.sql
@@ -11,7 +11,7 @@ WITH linha_sem_ressarcimento AS (
     DATETIME(PARSE_TIMESTAMP('%Y-%m-%d %H:%M:%S%Ez', timestamp_captura), "America/Sao_Paulo") AS timestamp_captura,
     DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%E*S%Ez', SAFE_CAST(JSON_VALUE(content, '$.dt_inclusao') AS STRING)), 'America/Sao_Paulo') AS dt_inclusao
   FROM
-    {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'linha_sem_ressarcimento') }}
+    {{ source('source_jae', 'linha_sem_ressarcimento') }}
 )
 SELECT
   * EXCEPT(rn)

diff --git a/...tagem_staging/staging_ordem_pagamento.sql → ...tagem/staging/staging_ordem_pagamento.sql b/...tagem_staging/staging_ordem_pagamento.sql → ...tagem/staging/staging_ordem_pagamento.sql
@@ -29,7 +29,7 @@ WITH ordem_pagamento AS (
     SAFE_CAST(JSON_VALUE(content, '$.valor_taxa') AS NUMERIC) AS valor_taxa,
     SAFE_CAST(JSON_VALUE(content, '$.valor_vendaabordo') AS NUMERIC) AS valor_vendaabordo
   FROM
-    {{ source("br_rj_riodejaneiro_bilhetagem_staging", "ordem_pagamento") }}
+    {{ source("source_jae", "ordem_pagamento") }}
 ),
 ordem_pagamento_rn AS (
   SELECT

diff --git a/...ing/staging_ordem_pagamento_consorcio.sql → ...ing/staging_ordem_pagamento_consorcio.sql b/...ing/staging_ordem_pagamento_consorcio.sql → ...ing/staging_ordem_pagamento_consorcio.sql
@@ -29,7 +29,7 @@ WITH ordem_pagamento_consorcio AS (
     SAFE_CAST(JSON_VALUE(content, '$.valor_taxa') AS NUMERIC) AS valor_taxa,
     SAFE_CAST(JSON_VALUE(content, '$.valor_vendaabordo') AS NUMERIC) AS valor_vendaabordo
   FROM
-      {{ source("br_rj_riodejaneiro_bilhetagem_staging", "ordem_pagamento_consorcio") }}
+      {{ source("source_jae", "ordem_pagamento_consorcio") }}
 ),
 ordem_pagamento_consorcio_rn AS (
   SELECT

diff --git a/...g_ordem_pagamento_consorcio_operadora.sql → ...g_ordem_pagamento_consorcio_operadora.sql b/...g_ordem_pagamento_consorcio_operadora.sql → ...g_ordem_pagamento_consorcio_operadora.sql
@@ -68,7 +68,7 @@ with
         from
             {{
                 source(
-                    "br_rj_riodejaneiro_bilhetagem_staging",
+                    "source_jae",
                     "ordem_pagamento_consorcio_operadora",
                 )
             }}

diff --git a/...lhetagem_staging/staging_ordem_rateio.sql → ...lhetagem/staging/staging_ordem_rateio.sql b/...lhetagem_staging/staging_ordem_rateio.sql → ...lhetagem/staging/staging_ordem_rateio.sql
@@ -43,7 +43,7 @@ WITH ordem_rateio AS (
         SAFE_CAST(JSON_VALUE(content, '$.valor_rateio_compensacao_debito_t4') AS NUMERIC) AS valor_rateio_compensacao_debito_t4,
         SAFE_CAST(JSON_VALUE(content, '$.valor_rateio_compensacao_debito_total') AS NUMERIC) AS valor_rateio_compensacao_debito_total
   FROM
-    {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'ordem_rateio') }}
+    {{ source('source_jae', 'ordem_rateio') }}
 )
 SELECT
   * EXCEPT(rn)

diff --git a/...m_staging/staging_ordem_ressarcimento.sql → ...m/staging/staging_ordem_ressarcimento.sql b/...m_staging/staging_ordem_ressarcimento.sql → ...m/staging/staging_ordem_ressarcimento.sql
@@ -35,7 +35,7 @@ WITH
             SAFE_CAST(JSON_VALUE(content, '$.valor_taxa') AS NUMERIC) AS valor_taxa,
             SAFE_CAST(JSON_VALUE(content, '$.valor_vendaabordo') AS NUMERIC) AS valor_vendaabordo
         FROM
-            {{ source("br_rj_riodejaneiro_bilhetagem_staging", "ordem_ressarcimento") }}
+            {{ source("source_jae", "ordem_ressarcimento") }}
     ),
     ordem_ressarcimento_rn AS (
         SELECT

diff --git a/.../staging_percentual_rateio_integracao.sql → .../staging_percentual_rateio_integracao.sql b/.../staging_percentual_rateio_integracao.sql → .../staging_percentual_rateio_integracao.sql
@@ -23,7 +23,7 @@ WITH percentual_rateio_integracao AS (
     SAFE_CAST(JSON_VALUE(content, '$.perc_rateio_integracao_t4') AS FLOAT64) AS perc_rateio_integracao_t4,
     SAFE_CAST(JSON_VALUE(content, '$.perc_rateio_origem') AS FLOAT64) AS perc_rateio_origem
   FROM
-    {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'percentual_rateio_integracao') }}
+    {{ source('source_jae', 'percentual_rateio_integracao') }}
 )
 SELECT
   * EXCEPT(rn)

diff --git a/..._bilhetagem_staging/staging_transacao.sql → .../bilhetagem/staging/staging_transacao.sql b/..._bilhetagem_staging/staging_transacao.sql → .../bilhetagem/staging/staging_transacao.sql
@@ -41,4 +41,4 @@ SELECT
     SAFE_CAST(JSON_VALUE(content, '$.vl_saldo') AS FLOAT64) AS vl_saldo,
     SAFE_CAST(JSON_VALUE(content, '$.id_tipo_modal') AS STRING) AS id_tipo_modal
 FROM
-    {{ source("br_rj_riodejaneiro_bilhetagem_staging", "transacao") }}
+    {{ source("source_jae", "transacao") }}
-Original file line number
+Diff line change
@@ Expand Up / @@ -68,7 +68,7 @@ with @@
             from
                 {{
                     source(
-                        "br_rj_riodejaneiro_bilhetagem_staging",
+                        "source_jae",
                         "ordem_pagamento_consorcio_operadora",
                     )
                 }}
@@ Expand Down @@