From 6904e3969de8827969e4c4059243255e8a5fd383 Mon Sep 17 00:00:00 2001 From: Rafael Pinheiro Date: Wed, 8 Jan 2025 13:27:07 -0300 Subject: [PATCH] migra br_rj_riodejaneiro_bilhetagem para bilhetagem --- .../dicionario_bilhetagem.sql} | 3 +- .../staging/aux_gratuidade.sql} | 4 + .../staging/aux_transacao_particao.sql | 84 +++++++++ .../staging}/staging_cliente.sql | 2 +- .../staging}/staging_endereco.sql | 2 +- .../staging}/staging_gratuidade.sql | 7 +- .../staging}/staging_integracao_transacao.sql | 2 +- .../staging}/staging_linha.sql | 2 +- .../staging}/staging_linha_consorcio.sql | 2 +- ...g_linha_consorcio_operadora_transporte.sql | 2 +- .../staging_linha_sem_ressarcimento.sql | 2 +- .../staging}/staging_ordem_pagamento.sql | 2 +- .../staging_ordem_pagamento_consorcio.sql | 2 +- ...ng_ordem_pagamento_consorcio_operadora.sql | 2 +- .../staging}/staging_ordem_rateio.sql | 2 +- .../staging}/staging_ordem_ressarcimento.sql | 2 +- .../staging_percentual_rateio_integracao.sql | 2 +- .../staging}/staging_transacao.sql | 2 +- queries/models/bilhetagem/transacao.sql | 168 ++++++++++++++++++ .../{transacao.sql => old_transacao.sql} | 3 +- .../staging_conta_bancaria.sql | 33 ---- .../staging_contato_pessoa_juridica.sql | 34 ---- .../staging_servico_motorista.sql | 23 --- queries/models/cadastro/consorcios.sql | 2 +- queries/models/cadastro/operadoras.sql | 10 +- .../models/cadastro/operadoras_contato.sql | 26 --- .../staging/staging_consorcio_jae.sql} | 2 +- .../staging_operadora_transporte_jae.sql} | 2 +- .../staging/aux_transacao_ordem.sql | 2 +- 29 files changed, 287 insertions(+), 144 deletions(-) rename queries/models/{br_rj_riodejaneiro_bilhetagem/dicionario.sql => bilhetagem/dicionario_bilhetagem.sql} (80%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging/gratuidade_aux.sql => bilhetagem/staging/aux_gratuidade.sql} (92%) create mode 100644 queries/models/bilhetagem/staging/aux_transacao_particao.sql rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_cliente.sql (94%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_endereco.sql (95%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_gratuidade.sql (74%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_integracao_transacao.sql (95%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_linha.sql (96%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_linha_consorcio.sql (91%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_linha_consorcio_operadora_transporte.sql (92%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_linha_sem_ressarcimento.sql (87%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_ordem_pagamento.sql (96%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_ordem_pagamento_consorcio.sql (96%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_ordem_pagamento_consorcio_operadora.sql (98%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_ordem_rateio.sql (98%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_ordem_ressarcimento.sql (97%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_percentual_rateio_integracao.sql (95%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging => bilhetagem/staging}/staging_transacao.sql (97%) create mode 100644 queries/models/bilhetagem/transacao.sql rename queries/models/br_rj_riodejaneiro_bilhetagem/{transacao.sql => old_transacao.sql} (99%) delete mode 100644 queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_conta_bancaria.sql delete mode 100644 queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_contato_pessoa_juridica.sql delete mode 100644 queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_servico_motorista.sql delete mode 100644 queries/models/cadastro/operadoras_contato.sql rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging/staging_consorcio.sql => cadastro/staging/staging_consorcio_jae.sql} (90%) rename queries/models/{br_rj_riodejaneiro_bilhetagem_staging/staging_operadora_transporte.sql => cadastro/staging/staging_operadora_transporte_jae.sql} (92%) diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem/dicionario.sql b/queries/models/bilhetagem/dicionario_bilhetagem.sql similarity index 80% rename from queries/models/br_rj_riodejaneiro_bilhetagem/dicionario.sql rename to queries/models/bilhetagem/dicionario_bilhetagem.sql index 7c303eca9..8278c241a 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem/dicionario.sql +++ b/queries/models/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") }} \ No newline at end of file + {{ source("source_jae", "dicionario") }} \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/gratuidade_aux.sql b/queries/models/bilhetagem/staging/aux_gratuidade.sql similarity index 92% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/gratuidade_aux.sql rename to queries/models/bilhetagem/staging/aux_gratuidade.sql index 92d448fe1..0998c378e 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/gratuidade_aux.sql +++ b/queries/models/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 new file mode 100644 index 000000000..f3c261795 --- /dev/null +++ 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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_cliente.sql b/queries/models/bilhetagem/staging/staging_cliente.sql similarity index 94% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_cliente.sql rename to queries/models/bilhetagem/staging/staging_cliente.sql index 4feda28c6..1456a8401 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_cliente.sql +++ b/queries/models/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_endereco.sql b/queries/models/bilhetagem/staging/staging_endereco.sql similarity index 95% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_endereco.sql rename to queries/models/bilhetagem/staging/staging_endereco.sql index 051100159..93cacad12 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_endereco.sql +++ b/queries/models/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_gratuidade.sql b/queries/models/bilhetagem/staging/staging_gratuidade.sql similarity index 74% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_gratuidade.sql rename to queries/models/bilhetagem/staging/staging_gratuidade.sql index 799558301..67e7a5c47 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_gratuidade.sql +++ b/queries/models/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_integracao_transacao.sql b/queries/models/bilhetagem/staging/staging_integracao_transacao.sql similarity index 95% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_integracao_transacao.sql rename to queries/models/bilhetagem/staging/staging_integracao_transacao.sql index 211241c49..74f0d97aa 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_integracao_transacao.sql +++ b/queries/models/bilhetagem/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') }} \ No newline at end of file + {{ source('source_jae', 'integracao_transacao') }} \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha.sql b/queries/models/bilhetagem/staging/staging_linha.sql similarity index 96% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha.sql rename to queries/models/bilhetagem/staging/staging_linha.sql index 9485fa599..83a985269 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha.sql +++ b/queries/models/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio.sql b/queries/models/bilhetagem/staging/staging_linha_consorcio.sql similarity index 91% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio.sql rename to queries/models/bilhetagem/staging/staging_linha_consorcio.sql index 019b069ac..b44c5aea9 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio.sql +++ b/queries/models/bilhetagem/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio_operadora_transporte.sql b/queries/models/bilhetagem/staging/staging_linha_consorcio_operadora_transporte.sql similarity index 92% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio_operadora_transporte.sql rename to queries/models/bilhetagem/staging/staging_linha_consorcio_operadora_transporte.sql index d067140d1..0a2f972cb 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_consorcio_operadora_transporte.sql +++ b/queries/models/bilhetagem/staging/staging_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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_sem_ressarcimento.sql b/queries/models/bilhetagem/staging/staging_linha_sem_ressarcimento.sql similarity index 87% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_sem_ressarcimento.sql rename to queries/models/bilhetagem/staging/staging_linha_sem_ressarcimento.sql index 826772cf4..631e21f8a 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_linha_sem_ressarcimento.sql +++ b/queries/models/bilhetagem/staging/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento.sql b/queries/models/bilhetagem/staging/staging_ordem_pagamento.sql similarity index 96% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento.sql rename to queries/models/bilhetagem/staging/staging_ordem_pagamento.sql index b59e649cc..bb07b4f1b 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento.sql +++ b/queries/models/bilhetagem/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio.sql b/queries/models/bilhetagem/staging/staging_ordem_pagamento_consorcio.sql similarity index 96% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio.sql rename to queries/models/bilhetagem/staging/staging_ordem_pagamento_consorcio.sql index e32b075c5..2276a9219 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio.sql +++ b/queries/models/bilhetagem/staging/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio_operadora.sql b/queries/models/bilhetagem/staging/staging_ordem_pagamento_consorcio_operadora.sql similarity index 98% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio_operadora.sql rename to queries/models/bilhetagem/staging/staging_ordem_pagamento_consorcio_operadora.sql index 65202cafb..7ea754c5e 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_pagamento_consorcio_operadora.sql +++ b/queries/models/bilhetagem/staging/staging_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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_rateio.sql b/queries/models/bilhetagem/staging/staging_ordem_rateio.sql similarity index 98% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_rateio.sql rename to queries/models/bilhetagem/staging/staging_ordem_rateio.sql index 65858e699..6dc6d2ece 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_rateio.sql +++ b/queries/models/bilhetagem/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_ressarcimento.sql b/queries/models/bilhetagem/staging/staging_ordem_ressarcimento.sql similarity index 97% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_ressarcimento.sql rename to queries/models/bilhetagem/staging/staging_ordem_ressarcimento.sql index 2e809805c..e68fb4502 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_ordem_ressarcimento.sql +++ b/queries/models/bilhetagem/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_percentual_rateio_integracao.sql b/queries/models/bilhetagem/staging/staging_percentual_rateio_integracao.sql similarity index 95% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_percentual_rateio_integracao.sql rename to queries/models/bilhetagem/staging/staging_percentual_rateio_integracao.sql index 480a5472a..0a9d4f24c 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_percentual_rateio_integracao.sql +++ b/queries/models/bilhetagem/staging/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/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_transacao.sql b/queries/models/bilhetagem/staging/staging_transacao.sql similarity index 97% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_transacao.sql rename to queries/models/bilhetagem/staging/staging_transacao.sql index 4cc69fc2d..a5cec5855 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_transacao.sql +++ b/queries/models/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") }} \ No newline at end of file + {{ source("source_jae", "transacao") }} \ No newline at end of file diff --git a/queries/models/bilhetagem/transacao.sql b/queries/models/bilhetagem/transacao.sql new file mode 100644 index 000000000..06e36b7e2 --- /dev/null +++ b/queries/models/bilhetagem/transacao.sql @@ -0,0 +1,168 @@ +{{ + config( + materialized="incremental", + partition_by={"field": "data", "data_type": "date", "granularity": "day"}, + incremental_strategy="insert_overwrite", + ) +}} + + +{% set transacao_ordem = ref("aux_transacao_id_ordem_pagamento") %} +{% set aux_transacao_particao = ref("aux_transacao_particao") %} + +{% set incremental_filter %} + 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')}}") +{% endset %} + +{% if execute %} + {% if is_incremental() %} + {% set transacao_partitions_query %} + select distinct + particao + from + ( + select + array_concat_agg(particoes) as particoes + from + {{ aux_transacao_particao }} + ), + unnest(particoes) as particao + + union distinct + + select + concat("'", parse_date("%Y%m%d", partition_id), "'") as particao + from + `rj-smtr.{{ transacao_ordem.schema }}.INFORMATION_SCHEMA.PARTITIONS` + where + table_name = "{{ transacao_ordem.identifier }}" + and partition_id != "__NULL__" + and datetime(last_modified_time, "America/Sao_Paulo") between datetime("{{var('date_range_start')}}") and (datetime("{{var('date_range_end')}}")) + + {% endset %} + + {% set transacao_partitions = run_query(transacao_partitions_query).columns[0].values() %} + {% endif %} +{% endif %} + +with transacao_staging as ( + select + * + from {{ ref('staging_transacao') }} + {% if is_incremental() %} + where {{ incremental_filter }} + {% endif %} +), +tipo_transacao as ( + select chave as id_tipo_transacao, valor as tipo_transacao, + from {{ ref('dicionario_bilhetagem') }} + {# from `rj-smtr.br_rj_riodejaneiro_bilhetagem.dicionario` #} + where id_tabela = "transacao" and coluna = "id_tipo_transacao" +), +gratuidade as ( + select + cast(id_cliente as string) as id_cliente, + tipo_gratuidade, + data_inicio_validade, + data_fim_validade + from {{ ref("aux_gratuidade") }} +), +tipo_pagamento as ( + select chave as id_tipo_pagamento, valor as tipo_pagamento + from {{ ref('dicionario_bilhetagem') }} + {# from `rj-smtr.br_rj_riodejaneiro_bilhetagem.dicionario` #} + where id_tabela = "transacao" and coluna = "id_tipo_pagamento" +), +integracao as ( + select id_transacao, valor_rateio, datetime_processamento_integracao + from {{ ref("integracao") }} + {# from `rj-smtr.br_rj_riodejaneiro_bilhetagem.integracao` #} + {% if is_incremental() %} + where + {% if transacao_partitions | length > 0 %} + data in ({{ transacao_partitions | join(", ") }}) + {% else %} data = "2000-01-01" + {% endif %} + {% endif %} + qualify + row_number() over ( + partition by id_transacao + order by datetime_processamento_integracao desc + ) + = 1 +), +transacao_ordem as ( + select * + from {{ ref("aux_transacao_id_ordem_pagamento") }} + {% if is_incremental() %} + where + {% if transacao_partitions | length > 0 %} + data in ({{ transacao_partitions | join(", ") }}) + {% else %} data = "2000-01-01" + {% endif %} + {% endif %} +), +transacao_nova as ( + select + extract(date from data_transacao) as data, + extract(hour from data_transacao) as hora, + data_transacao as datetime_transacao, + data_processamento as datetime_processamento, + t.timestamp_captura as datetime_captura, + m.modo, + dc.id_consorcio, + dc.consorcio, + do.id_operadora, + t.cd_operadora as id_operadora_jae, + do.operadora, + t.cd_linha as id_servico_jae, + l.nr_linha as servico_jae, + l.nm_linha as descricao_servico_jae, + sentido, + case + when m.modo = "VLT" + then substring(t.veiculo_id, 1, 3) + when m.modo = "BRT" + then null + else t.veiculo_id + end as id_veiculo, + t.numero_serie_validador as id_validador, + t.id_cliente as id_cliente, + t.pan_hash as hash_cartao, + id as id_transacao, + tp.tipo_pagamento, + tt.tipo_transacao, + g.tipo_gratuidade, + g.deficiencia_permanente, + g.rede_ensino, + latitude_trx as latitude, + longitude_trx as longitude, + st_geogpoint(longitude_trx, latitude_trx) as geo_point_transacao, + valor_transacao + from transacao as t + left join + {{ source("cadastro", "modos") }} m + on t.id_tipo_modal = m.id_modo + and m.fonte = "jae" + left join + {{ ref("operadoras") }} do + -- `rj-smtr.cadastro.operadoras` do + on t.cd_operadora = do.id_operadora_jae + left join + {{ ref("consorcios") }} dc + -- `rj-smtr.cadastro.consorcios` dc + on t.cd_consorcio = dc.id_consorcio_jae + left join {{ ref("staging_linha") }} l on t.cd_linha = l.cd_linha + left join tipo_transacao tt on tt.id_tipo_transacao = t.tipo_transacao + left join tipo_pagamento tp on t.id_tipo_midia = tp.id_tipo_pagamento + left join + gratuidade g + on t.id_cliente = g.id_cliente + and t.data_transacao >= g.data_inicio_validade + and (t.data_transacao < g.data_fim_validade or g.data_fim_validade is null) + left join + {{ ref("staging_linha_sem_ressarcimento") }} lsr + on t.cd_linha = lsr.id_linha + where lsr.id_linha is null and date(data_transacao) >= "2023-07-17" +) \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem/transacao.sql b/queries/models/br_rj_riodejaneiro_bilhetagem/old_transacao.sql similarity index 99% rename from queries/models/br_rj_riodejaneiro_bilhetagem/transacao.sql rename to queries/models/br_rj_riodejaneiro_bilhetagem/old_transacao.sql index 49ef7c41f..cae110006 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem/transacao.sql +++ b/queries/models/br_rj_riodejaneiro_bilhetagem/old_transacao.sql @@ -1,4 +1,3 @@ --- depends_on: {{ ref('operadoras_contato') }} -- depends_on: {{ ref('servico_operadora') }} -- depends_on: {{ ref('transacao_riocard') }} {{ @@ -100,7 +99,7 @@ with tipo_gratuidade, data_inicio_validade, data_fim_validade - from {{ ref("gratuidade_aux") }} + from {{ ref("aux_gratuidade") }} -- `rj-smtr.br_rj_riodejaneiro_bilhetagem_staging.gratuidade_aux` -- TODO: FILTRAR PARTIÇÕES DE FORMA EFICIENTE ), diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_conta_bancaria.sql b/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_conta_bancaria.sql deleted file mode 100644 index 8bb6fc5e4..000000000 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_conta_bancaria.sql +++ /dev/null @@ -1,33 +0,0 @@ -{{ - config( - alias='conta_bancaria', - ) -}} - -WITH - conta_bancaria AS ( - SELECT - data, - SAFE_CAST(CD_CLIENTE AS STRING) AS cd_cliente, - timestamp_captura, - SAFE_CAST(JSON_VALUE(content, '$.CD_AGENCIA') AS STRING) AS cd_agencia, - SAFE_CAST(JSON_VALUE(content, '$.CD_TIPO_CONTA') AS STRING) AS cd_tipo_conta, - SAFE_CAST(JSON_VALUE(content, '$.NM_BANCO') AS STRING) AS nm_banco, - SAFE_CAST(JSON_VALUE(content, '$.NR_BANCO') AS STRING) AS nr_banco, - SAFE_CAST(JSON_VALUE(content, '$.NR_CONTA') AS STRING) AS nr_conta, - FROM - {{ source("br_rj_riodejaneiro_bilhetagem_staging", "conta_bancaria") }} - ), - conta_bancaria_rn AS ( - SELECT - *, - ROW_NUMBER() OVER (PARTITION BY cd_cliente ORDER BY timestamp_captura DESC) AS rn - FROM - conta_bancaria - ) -SELECT - * EXCEPT(rn) -FROM - conta_bancaria_rn -WHERE - rn = 1 \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_contato_pessoa_juridica.sql b/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_contato_pessoa_juridica.sql deleted file mode 100644 index dbf02068d..000000000 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_contato_pessoa_juridica.sql +++ /dev/null @@ -1,34 +0,0 @@ -{{ - config( - alias='contato_pessoa_juridica', - ) -}} - -WITH - contato_pessoa_juridica AS ( - SELECT - data, - SAFE_CAST(NR_SEQ_CONTATO AS STRING) AS nr_seq_contato, - SAFE_CAST(CD_CLIENTE AS STRING) AS cd_cliente, - timestamp_captura, - DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez', SAFE_CAST(JSON_VALUE(content, '$.DT_INCLUSAO') AS STRING)), "America/Sao_Paulo") AS datetime_inclusao, - SAFE_CAST(JSON_VALUE(content, '$.NM_CONTATO') AS STRING) AS nm_contato, - SAFE_CAST(JSON_VALUE(content, '$.NR_RAMAL') AS STRING) AS nr_ramal, - SAFE_CAST(JSON_VALUE(content, '$.NR_TELEFONE') AS STRING) AS nr_telefone, - SAFE_CAST(JSON_VALUE(content, '$.TX_EMAIL') AS STRING) AS tx_email, - FROM - {{ source("br_rj_riodejaneiro_bilhetagem_staging", "contato_pessoa_juridica") }} - ), - contato_pessoa_juridica_rn AS ( - SELECT - *, - ROW_NUMBER() OVER (PARTITION BY nr_seq_contato, cd_cliente ORDER BY timestamp_captura DESC) AS rn - FROM - contato_pessoa_juridica - ) -SELECT - * EXCEPT(rn) -FROM - contato_pessoa_juridica_rn -WHERE - rn = 1 \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_servico_motorista.sql b/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_servico_motorista.sql deleted file mode 100644 index e50322b6e..000000000 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_servico_motorista.sql +++ /dev/null @@ -1,23 +0,0 @@ -{{ - config( - alias='servico_motorista', - ) -}} - -SELECT - data, - SAFE_CAST(NR_LOGICO_MIDIA AS STRING) AS nr_logico_midia, - SAFE_CAST(ID_SERVICO AS STRING) AS id_servico, - DATETIME(PARSE_TIMESTAMP('%Y-%m-%d %H:%M:%S%Ez', timestamp_captura), "America/Sao_Paulo") AS timestamp_captura, - SAFE_CAST(JSON_VALUE(content, '$.CD_LINHA') AS STRING) AS cd_linha, - SAFE_CAST(JSON_VALUE(content, '$.CD_OPERADORA') AS STRING) AS cd_operadora, - SAFE_CAST(JSON_VALUE(content, '$.CD_STATUS') AS STRING) AS cd_status, - DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez', SAFE_CAST(JSON_VALUE(content, '$.DT_ABERTURA') AS STRING)), 'America/Sao_Paulo') AS dt_abertura, - DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez', SAFE_CAST(JSON_VALUE(content, '$.DT_FECHAMENTO') AS STRING)), 'America/Sao_Paulo') AS dt_fechamento, - SAFE_CAST(JSON_VALUE(content, '$.ID_VEICULO') AS STRING) AS id_veiculo, - SAFE_CAST(JSON_VALUE(content, '$.NR_LOGICO_MIDIA_FECHAMENTO') AS STRING) AS nr_logico_midia_fechamento, - SAFE_CAST(JSON_VALUE(content, '$.SN_DEVICE') AS STRING) AS sn_device, - SAFE_CAST(JSON_VALUE(content, '$.TP_GERACAO') AS STRING) AS tp_geracao, - SAFE_CAST(JSON_VALUE(content, '$.VL_TARIFA_LINHA') AS FLOAT64) AS vl_tarifa_linha -FROM - {{ source('br_rj_riodejaneiro_bilhetagem_staging', 'servico_motorista') }} \ No newline at end of file diff --git a/queries/models/cadastro/consorcios.sql b/queries/models/cadastro/consorcios.sql index 6ff02358e..55f7e8917 100644 --- a/queries/models/cadastro/consorcios.sql +++ b/queries/models/cadastro/consorcios.sql @@ -32,7 +32,7 @@ WITH stu AS ( s.id_consorcio AS id_consorcio_stu, j.cd_consorcio AS id_consorcio_jae, s.processo AS id_processo - FROM {{ ref("staging_consorcio") }} AS j + FROM {{ ref("staging_consorcio_jae") }} AS j FULL OUTER JOIN stu AS s ON diff --git a/queries/models/cadastro/operadoras.sql b/queries/models/cadastro/operadoras.sql index 47fc19969..0759b2433 100644 --- a/queries/models/cadastro/operadoras.sql +++ b/queries/models/cadastro/operadoras.sql @@ -19,15 +19,15 @@ with end as tipo_documento, c.nr_documento, c.nm_cliente, - cb.cd_agencia, + {# cb.cd_agencia, cb.cd_tipo_conta, cb.nm_banco, cb.nr_banco, - cb.nr_conta - from {{ ref("staging_operadora_transporte") }} as ot + cb.nr_conta #} + from {{ ref("staging_operadora_transporte_jae") }} as ot join {{ ref("staging_cliente") }} as c on ot.cd_cliente = c.cd_cliente - left join - {{ ref("staging_conta_bancaria") }} as cb on ot.cd_cliente = cb.cd_cliente + {# left join + {{ ref("staging_conta_bancaria") }} as cb on ot.cd_cliente = cb.cd_cliente #} join {{ source("cadastro", "modos") }} m on ot.cd_tipo_modal = m.id_modo diff --git a/queries/models/cadastro/operadoras_contato.sql b/queries/models/cadastro/operadoras_contato.sql deleted file mode 100644 index 8de108498..000000000 --- a/queries/models/cadastro/operadoras_contato.sql +++ /dev/null @@ -1,26 +0,0 @@ -{{ - config( - materialized="table", - ) -}} - - -SELECT - d.id_operadora, - cpj.nm_contato AS contato, - cpj.nr_ramal AS ramal, - COALESCE(cpj.nr_telefone, c.nr_telefone) AS telefone, - COALESCE(cpj.tx_email, c.tx_email) AS email -FROM - {{ ref("staging_cliente") }} AS c -LEFT JOIN - {{ ref("staging_contato_pessoa_juridica") }} cpj -ON - c.cd_cliente = cpj.cd_cliente -JOIN - {{ ref("staging_operadora_transporte") }} AS ot -ON - ot.cd_cliente = c.cd_cliente -JOIN - {{ ref("operadoras") }} d -ON d.id_operadora_jae = ot.cd_operadora_transporte \ No newline at end of file diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_consorcio.sql b/queries/models/cadastro/staging/staging_consorcio_jae.sql similarity index 90% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_consorcio.sql rename to queries/models/cadastro/staging/staging_consorcio_jae.sql index 8cf67a473..bdc2ffc6b 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_consorcio.sql +++ b/queries/models/cadastro/staging/staging_consorcio_jae.sql @@ -13,7 +13,7 @@ WITH DATETIME(PARSE_TIMESTAMP('%Y-%m-%dT%H:%M:%S%Ez', SAFE_CAST(JSON_VALUE(content, '$.DT_INCLUSAO') AS STRING)), "America/Sao_Paulo") AS datetime_inclusao, SAFE_CAST(JSON_VALUE(content, '$.NM_CONSORCIO') AS STRING) AS nm_consorcio FROM - {{ source("br_rj_riodejaneiro_bilhetagem_staging", "consorcio") }} + {{ source("source_jae", "consorcio") }} ), consorcio_rn AS ( SELECT diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_operadora_transporte.sql b/queries/models/cadastro/staging/staging_operadora_transporte_jae.sql similarity index 92% rename from queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_operadora_transporte.sql rename to queries/models/cadastro/staging/staging_operadora_transporte_jae.sql index 8e68db6c7..03d1d272c 100644 --- a/queries/models/br_rj_riodejaneiro_bilhetagem_staging/staging_operadora_transporte.sql +++ b/queries/models/cadastro/staging/staging_operadora_transporte_jae.sql @@ -17,7 +17,7 @@ WITH SAFE_CAST(JSON_VALUE(content, '$.IN_SITUACAO_ATIVIDADE') AS STRING) AS in_situacao_atividade, SAFE_CAST(JSON_VALUE(content, '$.DS_TIPO_MODAL') AS STRING) AS ds_tipo_modal FROM - {{ source("br_rj_riodejaneiro_bilhetagem_staging", "operadora_transporte") }} + {{ source("source_jae", "operadora_transporte") }} ), operadora_transporte_rn AS ( SELECT diff --git a/queries/models/validacao_dados_jae/staging/aux_transacao_ordem.sql b/queries/models/validacao_dados_jae/staging/aux_transacao_ordem.sql index c0479f482..9bbb88e59 100644 --- a/queries/models/validacao_dados_jae/staging/aux_transacao_ordem.sql +++ b/queries/models/validacao_dados_jae/staging/aux_transacao_ordem.sql @@ -17,7 +17,7 @@ -- cd_operadora, -- ROW_NUMBER() OVER (PARTITION BY id_servico, nr_logico_midia ORDER BY timestamp_captura DESC) AS rn -- FROM --- {{ ref("staging_servico_motorista") }} +{# -- {{ ref("staging_servico_motorista") }} #} -- {% if is_incremental() %} -- WHERE -- DATE(data) BETWEEN DATE_SUB(DATE("{{var('date_range_start')}}"), INTERVAL 1 DAY) AND DATE("{{var('date_range_end')}}")