diff --git a/rj_smtr/constants.html b/rj_smtr/constants.html index d1481c213..8dd518587 100644 --- a/rj_smtr/constants.html +++ b/rj_smtr/constants.html @@ -188,11 +188,564 @@

Module pipelines.rj_smtr.constants

# SUBSÍDIO SUBSIDIO_SPPO_DATASET_ID = "projeto_subsidio_sppo" + SUBSIDIO_SPPO_SECRET_PATH = "projeto_subsidio_sppo" SUBSIDIO_SPPO_TABLE_ID = "viagem_completa" + SUBSIDIO_SPPO_CODE_OWNERS = ["rodrigo"] # SUBSÍDIO DASHBOARD SUBSIDIO_SPPO_DASHBOARD_DATASET_ID = "dashboard_subsidio_sppo" SUBSIDIO_SPPO_DASHBOARD_TABLE_ID = "sumario_servico_dia" + SUBSIDIO_SPPO_DATA_CHECKS_PARAMS = { + "check_gps_capture": { + "query": """WITH + t AS ( + SELECT + DATETIME(timestamp_array) AS timestamp_array + FROM + UNNEST( GENERATE_TIMESTAMP_ARRAY( TIMESTAMP("{start_timestamp}"), TIMESTAMP("{end_timestamp}"), INTERVAL {interval} minute) ) AS timestamp_array + WHERE + timestamp_array < TIMESTAMP("{end_timestamp}") ), + logs_table AS ( + SELECT + SAFE_CAST(DATETIME(TIMESTAMP(timestamp_captura), "America/Sao_Paulo") AS DATETIME) timestamp_captura, + SAFE_CAST(sucesso AS BOOLEAN) sucesso, + SAFE_CAST(erro AS STRING) erro, + SAFE_CAST(DATA AS DATE) DATA + FROM + rj-smtr-staging.{dataset_id}_staging.{table_id}_logs AS t ), + logs AS ( + SELECT + *, + TIMESTAMP_TRUNC(timestamp_captura, minute) AS timestamp_array + FROM + logs_table + WHERE + DATA BETWEEN DATE(TIMESTAMP("{start_timestamp}")) + AND DATE(TIMESTAMP("{end_timestamp}")) + AND timestamp_captura BETWEEN "{start_timestamp}" + AND "{end_timestamp}" ) + SELECT + COALESCE(logs.timestamp_captura, t.timestamp_array) AS timestamp_captura, + logs.erro + FROM + t + LEFT JOIN + logs + ON + logs.timestamp_array = t.timestamp_array + WHERE + logs.sucesso IS NOT TRUE""", + "order_columns": ["timestamp_captura"], + }, + "check_gps_treatment": { + "query": """ + WITH + data_hora AS ( + SELECT + EXTRACT(date + FROM + timestamp_array) AS DATA, + EXTRACT(hour + FROM + timestamp_array) AS hora, + FROM + UNNEST(GENERATE_TIMESTAMP_ARRAY("{start_timestamp}", "{end_timestamp}", INTERVAL 1 hour)) AS timestamp_array ), + gps_raw AS ( + SELECT + EXTRACT(date + FROM + timestamp_gps) AS DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_raw + FROM + `rj-smtr.br_rj_riodejaneiro_onibus_gps.registros` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 ), + gps_filtrada AS ( + SELECT + EXTRACT(date + FROM + timestamp_gps) AS DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_filtrada + FROM + `rj-smtr.br_rj_riodejaneiro_onibus_gps.sppo_aux_registros_filtrada` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 ), + gps_sppo AS ( + SELECT + DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_treated + FROM + `rj-smtr.br_rj_riodejaneiro_veiculos.gps_sppo` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2), + gps_join AS ( + SELECT + *, + SAFE_DIVIDE(q_gps_filtrada, q_gps_raw) as indice_tratamento_raw, + SAFE_DIVIDE(q_gps_treated, q_gps_filtrada) as indice_tratamento_filtrada, + CASE + WHEN q_gps_raw = 0 OR q_gps_filtrada = 0 OR q_gps_treated = 0 -- Hipótese de perda de dados no tratamento + OR q_gps_raw IS NULL OR q_gps_filtrada IS NULL OR q_gps_treated IS NULL -- Hipótese de perda de dados no tratamento + OR (q_gps_raw <= q_gps_filtrada) OR (q_gps_filtrada < q_gps_treated) -- Hipótese de duplicação de dados + OR (COALESCE(SAFE_DIVIDE(q_gps_filtrada, q_gps_raw), 0) < 0.96) -- Hipótese de perda de dados no tratamento (superior a 3%) + OR (COALESCE(SAFE_DIVIDE(q_gps_treated, q_gps_filtrada), 0) < 0.96) -- Hipótese de perda de dados no tratamento (superior a 3%) + THEN FALSE + ELSE + TRUE + END + AS status + FROM + data_hora + LEFT JOIN + gps_raw + USING + (DATA, + hora) + LEFT JOIN + gps_filtrada + USING + (DATA, + hora) + LEFT JOIN + gps_sppo + USING + (DATA, + hora)) + SELECT + * + FROM + gps_join + WHERE + status IS FALSE + """, + "order_columns": ["DATA", "hora"], + }, + "check_sppo_veiculo_dia": { + "query": """ + WITH + count_dist_status AS ( + SELECT + DATA, + COUNT(DISTINCT status) AS q_dist_status, + NULL AS q_duplicated_status, + NULL AS q_null_status + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1 + HAVING + COUNT(DISTINCT status) = 1 ), + count_duplicated_status AS ( + SELECT + DATA, + id_veiculo, + COUNT(*) AS q_status, + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 + HAVING + COUNT(*) > 1 ), + count_duplicated_status_agg AS ( + SELECT + DATA, + NULL AS q_dist_status, + SUM(q_status) AS q_duplicated_status, + NULL AS q_null_status + FROM + count_duplicated_status + GROUP BY + 1), + count_null_status AS ( + SELECT + DATA, + NULL AS q_dist_status, + NULL AS q_duplicated_status, + COUNT(*) AS q_null_status + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + AND status IS NULL + GROUP BY + 1 ) + SELECT + * + FROM + count_dist_status + + UNION ALL + + SELECT + * + FROM + count_duplicated_status_agg + + UNION ALL + + SELECT + * + FROM + count_null_status + """, + "order_columns": ["DATA"], + }, + "accepted_values_valor_penalidade": { + "query": """ + WITH + all_values AS ( + SELECT + DISTINCT valor_penalidade AS value_field, + COUNT(*) AS n_records + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + valor_penalidade ) + SELECT + * + FROM + all_values + WHERE + value_field NOT IN ( + SELECT + valor + FROM + `rj-smtr`.`dashboard_subsidio_sppo`.`valor_tipo_penalidade` ) + """, + "order_columns": ["n_records"], + }, + "teto_pagamento_valor_subsidio_pago": { + "query": """ + WITH + {table_id} AS ( + SELECT + * + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")), + subsidio_parametros AS ( + SELECT + data_inicio, + data_fim, + MAX(subsidio_km) AS subsidio_km_teto + FROM + `rj-smtr`.`dashboard_subsidio_sppo`.`subsidio_parametros` + WHERE + subsidio_km > 0 + GROUP BY + 1, + 2) + SELECT + * + FROM + {table_id} AS s + LEFT JOIN + subsidio_parametros AS p + ON + s.data BETWEEN p.data_inicio + AND p.data_fim + WHERE + NOT({expression}) + """, + "order_columns": ["data"], + }, + "expression_is_true": { + "query": """ + SELECT + * + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + (DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")) + AND NOT({expression}) + """, + "order_columns": ["data"], + }, + "unique_combination": { + "query": """ + SELECT + {expression} + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + {expression} + HAVING + COUNT(*) > 1 + """, + }, + "teste_completude": { + "query": """ + WITH + time_array AS ( + SELECT + * + FROM + UNNEST(GENERATE_DATE_ARRAY(DATE("{start_timestamp}"), DATE("{end_timestamp}"))) AS DATA ), + {table_id} AS ( + SELECT + DATA, + COUNT(*) AS q_registros + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1 ) + SELECT + DATA, + q_registros + FROM + time_array + LEFT JOIN + {table_id} + USING + (DATA) + WHERE + q_registros IS NULL + OR q_registros = 0 + """, + "order_columns": ["DATA"], + }, + "teste_sumario_servico_dia_tipo_soma_km": { + "query": """ + WITH + kms AS ( + SELECT + * EXCEPT(km_apurada), + km_apurada, + ROUND(COALESCE(km_apurada_registrado_com_ar_inoperante,0) + COALESCE(km_apurada_n_licenciado,0) + COALESCE(km_apurada_autuado_ar_inoperante,0) + COALESCE(km_apurada_autuado_seguranca,0) + COALESCE(km_apurada_autuado_limpezaequipamento,0) + COALESCE(km_apurada_licenciado_sem_ar_n_autuado,0) + COALESCE(km_apurada_licenciado_com_ar_n_autuado,0),2) AS km_apurada2 + FROM + `rj-smtr.dashboard_subsidio_sppo.sumario_servico_dia_tipo` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")) + SELECT + *, + ABS(km_apurada2-km_apurada) AS dif + FROM + kms + WHERE + ABS(km_apurada2-km_apurada) > 0.015 + """, + "order_columns": ["dif"], + }, + } + SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST = { + "general": { + "Todos os dados de GPS foram capturados": { + "test": "check_gps_capture", + "params": { + "interval": 1, + "dataset_id": GPS_SPPO_RAW_DATASET_ID, + "table_id": GPS_SPPO_RAW_TABLE_ID, + }, + }, + "Todos os dados de realocação foram capturados": { + "test": "check_gps_capture", + "params": { + "interval": 10, + "dataset_id": GPS_SPPO_RAW_DATASET_ID, + "table_id": GPS_SPPO_REALOCACAO_RAW_TABLE_ID, + }, + }, + "Todos os dados de GPS foram devidamente tratados": { + "test": "check_gps_treatment", + }, + "Todos os dados de status dos veículos foram devidamente tratados": { + "test": "check_sppo_veiculo_dia", + }, + } + } + SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST = { + "sumario_servico_dia": { + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todos serviços com valores de penalidade aceitos": { + "test": "accepted_values_valor_penalidade" + }, + "Todos serviços abaixo do teto de pagamento de valor do subsídio": { + "test": "teto_pagamento_valor_subsidio_pago", + "expression": "ROUND(valor_subsidio_pago/subsidio_km_teto,2) <= ROUND(km_apurada+0.01,2)", + }, + "Todos serviços são únicos em cada data": { + "test": "unique_combination", + "expression": "data, servico", + }, + "Todos serviços possuem data não nula": { + "expression": "data IS NOT NULL", + }, + "Todos serviços possuem tipo de dia não nulo": { + "expression": "tipo_dia IS NOT NULL", + }, + "Todos serviços possuem consórcio não nulo": { + "expression": "consorcio IS NOT NULL", + }, + "Todas as datas possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todos serviços com quantidade de viagens não nula e maior ou igual a zero": { + "expression": "viagens IS NOT NULL AND viagens >= 0", + }, + "Todos serviços com quilometragem apurada não nula e maior ou igual a zero": { + "expression": "km_apurada IS NOT NULL AND km_apurada >= 0", + }, + "Todos serviços com quilometragem planejada não nula e maior ou igual a zero": { + "expression": "km_planejada IS NOT NULL AND km_planejada >= 0", + }, + "Todos serviços com Percentual de Operação Diário (POD) não nulo e maior ou igual a zero": { + "expression": "perc_km_planejada IS NOT NULL AND perc_km_planejada >= 0", + }, + "Todos serviços com valor de subsídio pago não nulo e maior ou igual a zero": { + "expression": "valor_subsidio_pago IS NOT NULL AND valor_subsidio_pago >= 0", + }, + }, + "sumario_servico_dia_tipo_sem_glosa": { + "Todas as somas dos tipos de quilometragem são equivalentes a quilometragem total": { + "test": "teste_sumario_servico_dia_tipo_soma_km" + }, + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todos serviços abaixo do teto de pagamento de valor do subsídio": { + "test": "teto_pagamento_valor_subsidio_pago", + "expression": "ROUND(valor_total_subsidio/subsidio_km_teto,2) <= ROUND(distancia_total_subsidio+0.01,2)", + }, + "Todos serviços são únicos em cada data": { + "test": "unique_combination", + "expression": "data, servico", + }, + "Todos serviços possuem data não nula": { + "expression": "data IS NOT NULL", + }, + "Todos serviços possuem tipo de dia não nulo": { + "expression": "tipo_dia IS NOT NULL", + }, + "Todos serviços possuem consórcio não nulo": { + "expression": "consorcio IS NOT NULL", + }, + "Todas as datas possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todos serviços com quantidade de viagens não nula e maior ou igual a zero": { + "expression": "viagens_subsidio IS NOT NULL AND viagens_subsidio >= 0", + }, + "Todos serviços com quilometragem apurada não nula e maior ou igual a zero": { + "expression": "distancia_total_subsidio IS NOT NULL AND distancia_total_subsidio >= 0", + }, + "Todos serviços com quilometragem planejada não nula e maior ou igual a zero": { + "expression": "distancia_total_planejada IS NOT NULL AND distancia_total_planejada >= 0", + }, + "Todos serviços com Percentual de Operação Diário (POD) não nulo e maior ou igual a zero": { + "expression": "perc_distancia_total_subsidio IS NOT NULL AND perc_distancia_total_subsidio >= 0", + }, + "Todos serviços com valor total de subsídio não nulo e maior ou igual a zero": { + "expression": "valor_total_subsidio IS NOT NULL AND valor_total_subsidio >= 0", + }, + "Todos serviços com viagens por veículos não licenciados não nulo e maior ou igual a zero": { + "expression": "viagens_n_licenciado IS NOT NULL AND viagens_n_licenciado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos não licenciados não nulo e maior ou igual a zero": { + "expression": "km_apurada_n_licenciado IS NOT NULL AND km_apurada_n_licenciado >= 0", + }, + "Todos serviços com viagens por veículos autuados por ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_ar_inoperante IS NOT NULL AND viagens_autuado_ar_inoperante >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_ar_inoperante IS NOT NULL AND km_apurada_autuado_ar_inoperante >= 0", + }, + "Todos serviços com viagens por veículos autuados por segurança não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_seguranca IS NOT NULL AND viagens_autuado_seguranca >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por segurança não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_seguranca IS NOT NULL AND km_apurada_autuado_seguranca >= 0", + }, + "Todos serviços com viagens por veículos autuados por limpeza/equipamento não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_limpezaequipamento IS NOT NULL AND viagens_autuado_limpezaequipamento >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por limpeza/equipamento não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_limpezaequipamento IS NOT NULL AND km_apurada_autuado_limpezaequipamento >= 0", + }, + "Todos serviços com viagens por veículos sem ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "viagens_licenciado_sem_ar_n_autuado IS NOT NULL AND viagens_licenciado_sem_ar_n_autuado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos sem ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "km_apurada_licenciado_sem_ar_n_autuado IS NOT NULL AND km_apurada_licenciado_sem_ar_n_autuado >= 0", + }, + "Todos serviços com viagens por veículos com ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "viagens_licenciado_com_ar_n_autuado IS NOT NULL AND viagens_licenciado_com_ar_n_autuado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos com ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "km_apurada_licenciado_com_ar_n_autuado IS NOT NULL AND km_apurada_licenciado_com_ar_n_autuado >= 0", + }, + "Todos serviços com viagens por veículos registrados com ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "viagens_registrado_com_ar_inoperante IS NOT NULL AND viagens_registrado_com_ar_inoperante >= 0", + }, + "Todos serviços com quilometragem apurada por veículos registrados com ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "km_apurada_registrado_com_ar_inoperante IS NOT NULL AND km_apurada_registrado_com_ar_inoperante >= 0", + }, + }, + "viagens_remuneradas": { + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todas viagens são únicas": { + "test": "unique_combination", + "expression": "id_viagem", + }, + "Todas viagens possuem data": { + "expression": "data IS NOT NULL", + }, + "Todas viagens possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todas viagens possuem ID não nulo": { + "expression": "id_viagem IS NOT NULL", + }, + "Todas viagens possuem indicador de viagem remunerada não nulo e verdadeiro/falso": { + "expression": "indicador_viagem_remunerada IS NOT NULL AND indicador_viagem_remunerada IN (TRUE, FALSE)", + }, + "Todas viagens com distância planejada não nula e maior ou igual a zero": { + "expression": "distancia_planejada IS NOT NULL AND distancia_planejada >= 0", + }, + "Todas viagens com valor de subsídio por km não nulo e maior ou igual a zero": { + "expression": "subsidio_km IS NOT NULL AND subsidio_km >= 0", + }, + }, + } # BILHETAGEM BILHETAGEM_DATASET_ID = "br_rj_riodejaneiro_bilhetagem" @@ -1033,11 +1586,564 @@

Classes

# SUBSÍDIO SUBSIDIO_SPPO_DATASET_ID = "projeto_subsidio_sppo" + SUBSIDIO_SPPO_SECRET_PATH = "projeto_subsidio_sppo" SUBSIDIO_SPPO_TABLE_ID = "viagem_completa" + SUBSIDIO_SPPO_CODE_OWNERS = ["rodrigo"] # SUBSÍDIO DASHBOARD SUBSIDIO_SPPO_DASHBOARD_DATASET_ID = "dashboard_subsidio_sppo" SUBSIDIO_SPPO_DASHBOARD_TABLE_ID = "sumario_servico_dia" + SUBSIDIO_SPPO_DATA_CHECKS_PARAMS = { + "check_gps_capture": { + "query": """WITH + t AS ( + SELECT + DATETIME(timestamp_array) AS timestamp_array + FROM + UNNEST( GENERATE_TIMESTAMP_ARRAY( TIMESTAMP("{start_timestamp}"), TIMESTAMP("{end_timestamp}"), INTERVAL {interval} minute) ) AS timestamp_array + WHERE + timestamp_array < TIMESTAMP("{end_timestamp}") ), + logs_table AS ( + SELECT + SAFE_CAST(DATETIME(TIMESTAMP(timestamp_captura), "America/Sao_Paulo") AS DATETIME) timestamp_captura, + SAFE_CAST(sucesso AS BOOLEAN) sucesso, + SAFE_CAST(erro AS STRING) erro, + SAFE_CAST(DATA AS DATE) DATA + FROM + rj-smtr-staging.{dataset_id}_staging.{table_id}_logs AS t ), + logs AS ( + SELECT + *, + TIMESTAMP_TRUNC(timestamp_captura, minute) AS timestamp_array + FROM + logs_table + WHERE + DATA BETWEEN DATE(TIMESTAMP("{start_timestamp}")) + AND DATE(TIMESTAMP("{end_timestamp}")) + AND timestamp_captura BETWEEN "{start_timestamp}" + AND "{end_timestamp}" ) + SELECT + COALESCE(logs.timestamp_captura, t.timestamp_array) AS timestamp_captura, + logs.erro + FROM + t + LEFT JOIN + logs + ON + logs.timestamp_array = t.timestamp_array + WHERE + logs.sucesso IS NOT TRUE""", + "order_columns": ["timestamp_captura"], + }, + "check_gps_treatment": { + "query": """ + WITH + data_hora AS ( + SELECT + EXTRACT(date + FROM + timestamp_array) AS DATA, + EXTRACT(hour + FROM + timestamp_array) AS hora, + FROM + UNNEST(GENERATE_TIMESTAMP_ARRAY("{start_timestamp}", "{end_timestamp}", INTERVAL 1 hour)) AS timestamp_array ), + gps_raw AS ( + SELECT + EXTRACT(date + FROM + timestamp_gps) AS DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_raw + FROM + `rj-smtr.br_rj_riodejaneiro_onibus_gps.registros` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 ), + gps_filtrada AS ( + SELECT + EXTRACT(date + FROM + timestamp_gps) AS DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_filtrada + FROM + `rj-smtr.br_rj_riodejaneiro_onibus_gps.sppo_aux_registros_filtrada` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 ), + gps_sppo AS ( + SELECT + DATA, + EXTRACT(hour + FROM + timestamp_gps) AS hora, + COUNT(*) AS q_gps_treated + FROM + `rj-smtr.br_rj_riodejaneiro_veiculos.gps_sppo` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2), + gps_join AS ( + SELECT + *, + SAFE_DIVIDE(q_gps_filtrada, q_gps_raw) as indice_tratamento_raw, + SAFE_DIVIDE(q_gps_treated, q_gps_filtrada) as indice_tratamento_filtrada, + CASE + WHEN q_gps_raw = 0 OR q_gps_filtrada = 0 OR q_gps_treated = 0 -- Hipótese de perda de dados no tratamento + OR q_gps_raw IS NULL OR q_gps_filtrada IS NULL OR q_gps_treated IS NULL -- Hipótese de perda de dados no tratamento + OR (q_gps_raw <= q_gps_filtrada) OR (q_gps_filtrada < q_gps_treated) -- Hipótese de duplicação de dados + OR (COALESCE(SAFE_DIVIDE(q_gps_filtrada, q_gps_raw), 0) < 0.96) -- Hipótese de perda de dados no tratamento (superior a 3%) + OR (COALESCE(SAFE_DIVIDE(q_gps_treated, q_gps_filtrada), 0) < 0.96) -- Hipótese de perda de dados no tratamento (superior a 3%) + THEN FALSE + ELSE + TRUE + END + AS status + FROM + data_hora + LEFT JOIN + gps_raw + USING + (DATA, + hora) + LEFT JOIN + gps_filtrada + USING + (DATA, + hora) + LEFT JOIN + gps_sppo + USING + (DATA, + hora)) + SELECT + * + FROM + gps_join + WHERE + status IS FALSE + """, + "order_columns": ["DATA", "hora"], + }, + "check_sppo_veiculo_dia": { + "query": """ + WITH + count_dist_status AS ( + SELECT + DATA, + COUNT(DISTINCT status) AS q_dist_status, + NULL AS q_duplicated_status, + NULL AS q_null_status + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1 + HAVING + COUNT(DISTINCT status) = 1 ), + count_duplicated_status AS ( + SELECT + DATA, + id_veiculo, + COUNT(*) AS q_status, + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1, + 2 + HAVING + COUNT(*) > 1 ), + count_duplicated_status_agg AS ( + SELECT + DATA, + NULL AS q_dist_status, + SUM(q_status) AS q_duplicated_status, + NULL AS q_null_status + FROM + count_duplicated_status + GROUP BY + 1), + count_null_status AS ( + SELECT + DATA, + NULL AS q_dist_status, + NULL AS q_duplicated_status, + COUNT(*) AS q_null_status + FROM + rj-smtr.veiculo.sppo_veiculo_dia + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + AND status IS NULL + GROUP BY + 1 ) + SELECT + * + FROM + count_dist_status + + UNION ALL + + SELECT + * + FROM + count_duplicated_status_agg + + UNION ALL + + SELECT + * + FROM + count_null_status + """, + "order_columns": ["DATA"], + }, + "accepted_values_valor_penalidade": { + "query": """ + WITH + all_values AS ( + SELECT + DISTINCT valor_penalidade AS value_field, + COUNT(*) AS n_records + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + valor_penalidade ) + SELECT + * + FROM + all_values + WHERE + value_field NOT IN ( + SELECT + valor + FROM + `rj-smtr`.`dashboard_subsidio_sppo`.`valor_tipo_penalidade` ) + """, + "order_columns": ["n_records"], + }, + "teto_pagamento_valor_subsidio_pago": { + "query": """ + WITH + {table_id} AS ( + SELECT + * + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")), + subsidio_parametros AS ( + SELECT + data_inicio, + data_fim, + MAX(subsidio_km) AS subsidio_km_teto + FROM + `rj-smtr`.`dashboard_subsidio_sppo`.`subsidio_parametros` + WHERE + subsidio_km > 0 + GROUP BY + 1, + 2) + SELECT + * + FROM + {table_id} AS s + LEFT JOIN + subsidio_parametros AS p + ON + s.data BETWEEN p.data_inicio + AND p.data_fim + WHERE + NOT({expression}) + """, + "order_columns": ["data"], + }, + "expression_is_true": { + "query": """ + SELECT + * + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + (DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")) + AND NOT({expression}) + """, + "order_columns": ["data"], + }, + "unique_combination": { + "query": """ + SELECT + {expression} + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + {expression} + HAVING + COUNT(*) > 1 + """, + }, + "teste_completude": { + "query": """ + WITH + time_array AS ( + SELECT + * + FROM + UNNEST(GENERATE_DATE_ARRAY(DATE("{start_timestamp}"), DATE("{end_timestamp}"))) AS DATA ), + {table_id} AS ( + SELECT + DATA, + COUNT(*) AS q_registros + FROM + `rj-smtr`.`{dataset_id}`.`{table_id}` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}") + GROUP BY + 1 ) + SELECT + DATA, + q_registros + FROM + time_array + LEFT JOIN + {table_id} + USING + (DATA) + WHERE + q_registros IS NULL + OR q_registros = 0 + """, + "order_columns": ["DATA"], + }, + "teste_sumario_servico_dia_tipo_soma_km": { + "query": """ + WITH + kms AS ( + SELECT + * EXCEPT(km_apurada), + km_apurada, + ROUND(COALESCE(km_apurada_registrado_com_ar_inoperante,0) + COALESCE(km_apurada_n_licenciado,0) + COALESCE(km_apurada_autuado_ar_inoperante,0) + COALESCE(km_apurada_autuado_seguranca,0) + COALESCE(km_apurada_autuado_limpezaequipamento,0) + COALESCE(km_apurada_licenciado_sem_ar_n_autuado,0) + COALESCE(km_apurada_licenciado_com_ar_n_autuado,0),2) AS km_apurada2 + FROM + `rj-smtr.dashboard_subsidio_sppo.sumario_servico_dia_tipo` + WHERE + DATA BETWEEN DATE("{start_timestamp}") + AND DATE("{end_timestamp}")) + SELECT + *, + ABS(km_apurada2-km_apurada) AS dif + FROM + kms + WHERE + ABS(km_apurada2-km_apurada) > 0.015 + """, + "order_columns": ["dif"], + }, + } + SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST = { + "general": { + "Todos os dados de GPS foram capturados": { + "test": "check_gps_capture", + "params": { + "interval": 1, + "dataset_id": GPS_SPPO_RAW_DATASET_ID, + "table_id": GPS_SPPO_RAW_TABLE_ID, + }, + }, + "Todos os dados de realocação foram capturados": { + "test": "check_gps_capture", + "params": { + "interval": 10, + "dataset_id": GPS_SPPO_RAW_DATASET_ID, + "table_id": GPS_SPPO_REALOCACAO_RAW_TABLE_ID, + }, + }, + "Todos os dados de GPS foram devidamente tratados": { + "test": "check_gps_treatment", + }, + "Todos os dados de status dos veículos foram devidamente tratados": { + "test": "check_sppo_veiculo_dia", + }, + } + } + SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST = { + "sumario_servico_dia": { + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todos serviços com valores de penalidade aceitos": { + "test": "accepted_values_valor_penalidade" + }, + "Todos serviços abaixo do teto de pagamento de valor do subsídio": { + "test": "teto_pagamento_valor_subsidio_pago", + "expression": "ROUND(valor_subsidio_pago/subsidio_km_teto,2) <= ROUND(km_apurada+0.01,2)", + }, + "Todos serviços são únicos em cada data": { + "test": "unique_combination", + "expression": "data, servico", + }, + "Todos serviços possuem data não nula": { + "expression": "data IS NOT NULL", + }, + "Todos serviços possuem tipo de dia não nulo": { + "expression": "tipo_dia IS NOT NULL", + }, + "Todos serviços possuem consórcio não nulo": { + "expression": "consorcio IS NOT NULL", + }, + "Todas as datas possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todos serviços com quantidade de viagens não nula e maior ou igual a zero": { + "expression": "viagens IS NOT NULL AND viagens >= 0", + }, + "Todos serviços com quilometragem apurada não nula e maior ou igual a zero": { + "expression": "km_apurada IS NOT NULL AND km_apurada >= 0", + }, + "Todos serviços com quilometragem planejada não nula e maior ou igual a zero": { + "expression": "km_planejada IS NOT NULL AND km_planejada >= 0", + }, + "Todos serviços com Percentual de Operação Diário (POD) não nulo e maior ou igual a zero": { + "expression": "perc_km_planejada IS NOT NULL AND perc_km_planejada >= 0", + }, + "Todos serviços com valor de subsídio pago não nulo e maior ou igual a zero": { + "expression": "valor_subsidio_pago IS NOT NULL AND valor_subsidio_pago >= 0", + }, + }, + "sumario_servico_dia_tipo_sem_glosa": { + "Todas as somas dos tipos de quilometragem são equivalentes a quilometragem total": { + "test": "teste_sumario_servico_dia_tipo_soma_km" + }, + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todos serviços abaixo do teto de pagamento de valor do subsídio": { + "test": "teto_pagamento_valor_subsidio_pago", + "expression": "ROUND(valor_total_subsidio/subsidio_km_teto,2) <= ROUND(distancia_total_subsidio+0.01,2)", + }, + "Todos serviços são únicos em cada data": { + "test": "unique_combination", + "expression": "data, servico", + }, + "Todos serviços possuem data não nula": { + "expression": "data IS NOT NULL", + }, + "Todos serviços possuem tipo de dia não nulo": { + "expression": "tipo_dia IS NOT NULL", + }, + "Todos serviços possuem consórcio não nulo": { + "expression": "consorcio IS NOT NULL", + }, + "Todas as datas possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todos serviços com quantidade de viagens não nula e maior ou igual a zero": { + "expression": "viagens_subsidio IS NOT NULL AND viagens_subsidio >= 0", + }, + "Todos serviços com quilometragem apurada não nula e maior ou igual a zero": { + "expression": "distancia_total_subsidio IS NOT NULL AND distancia_total_subsidio >= 0", + }, + "Todos serviços com quilometragem planejada não nula e maior ou igual a zero": { + "expression": "distancia_total_planejada IS NOT NULL AND distancia_total_planejada >= 0", + }, + "Todos serviços com Percentual de Operação Diário (POD) não nulo e maior ou igual a zero": { + "expression": "perc_distancia_total_subsidio IS NOT NULL AND perc_distancia_total_subsidio >= 0", + }, + "Todos serviços com valor total de subsídio não nulo e maior ou igual a zero": { + "expression": "valor_total_subsidio IS NOT NULL AND valor_total_subsidio >= 0", + }, + "Todos serviços com viagens por veículos não licenciados não nulo e maior ou igual a zero": { + "expression": "viagens_n_licenciado IS NOT NULL AND viagens_n_licenciado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos não licenciados não nulo e maior ou igual a zero": { + "expression": "km_apurada_n_licenciado IS NOT NULL AND km_apurada_n_licenciado >= 0", + }, + "Todos serviços com viagens por veículos autuados por ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_ar_inoperante IS NOT NULL AND viagens_autuado_ar_inoperante >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_ar_inoperante IS NOT NULL AND km_apurada_autuado_ar_inoperante >= 0", + }, + "Todos serviços com viagens por veículos autuados por segurança não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_seguranca IS NOT NULL AND viagens_autuado_seguranca >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por segurança não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_seguranca IS NOT NULL AND km_apurada_autuado_seguranca >= 0", + }, + "Todos serviços com viagens por veículos autuados por limpeza/equipamento não nulo e maior ou igual a zero": { + "expression": "viagens_autuado_limpezaequipamento IS NOT NULL AND viagens_autuado_limpezaequipamento >= 0", + }, + "Todos serviços com quilometragem apurada por veículos autuados por limpeza/equipamento não nulo e maior ou igual a zero": { + "expression": "km_apurada_autuado_limpezaequipamento IS NOT NULL AND km_apurada_autuado_limpezaequipamento >= 0", + }, + "Todos serviços com viagens por veículos sem ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "viagens_licenciado_sem_ar_n_autuado IS NOT NULL AND viagens_licenciado_sem_ar_n_autuado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos sem ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "km_apurada_licenciado_sem_ar_n_autuado IS NOT NULL AND km_apurada_licenciado_sem_ar_n_autuado >= 0", + }, + "Todos serviços com viagens por veículos com ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "viagens_licenciado_com_ar_n_autuado IS NOT NULL AND viagens_licenciado_com_ar_n_autuado >= 0", + }, + "Todos serviços com quilometragem apurada por veículos com ar condicionado e não autuado não nulo e maior ou igual a zero": { + "expression": "km_apurada_licenciado_com_ar_n_autuado IS NOT NULL AND km_apurada_licenciado_com_ar_n_autuado >= 0", + }, + "Todos serviços com viagens por veículos registrados com ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "viagens_registrado_com_ar_inoperante IS NOT NULL AND viagens_registrado_com_ar_inoperante >= 0", + }, + "Todos serviços com quilometragem apurada por veículos registrados com ar condicionado inoperante não nulo e maior ou igual a zero": { + "expression": "km_apurada_registrado_com_ar_inoperante IS NOT NULL AND km_apurada_registrado_com_ar_inoperante >= 0", + }, + }, + "viagens_remuneradas": { + "Todas as datas possuem dados": {"test": "teste_completude"}, + "Todas viagens são únicas": { + "test": "unique_combination", + "expression": "id_viagem", + }, + "Todas viagens possuem data": { + "expression": "data IS NOT NULL", + }, + "Todas viagens possuem serviço não nulo": { + "expression": "servico IS NOT NULL", + }, + "Todas viagens possuem ID não nulo": { + "expression": "id_viagem IS NOT NULL", + }, + "Todas viagens possuem indicador de viagem remunerada não nulo e verdadeiro/falso": { + "expression": "indicador_viagem_remunerada IS NOT NULL AND indicador_viagem_remunerada IN (TRUE, FALSE)", + }, + "Todas viagens com distância planejada não nula e maior ou igual a zero": { + "expression": "distancia_planejada IS NOT NULL AND distancia_planejada >= 0", + }, + "Todas viagens com valor de subsídio por km não nulo e maior ou igual a zero": { + "expression": "subsidio_km IS NOT NULL AND subsidio_km >= 0", + }, + }, + } # BILHETAGEM BILHETAGEM_DATASET_ID = "br_rj_riodejaneiro_bilhetagem" @@ -2079,6 +3185,10 @@

Class variables

+
var SUBSIDIO_SPPO_CODE_OWNERS
+
+
+
var SUBSIDIO_SPPO_DASHBOARD_DATASET_ID
@@ -2091,6 +3201,18 @@

Class variables

+
var SUBSIDIO_SPPO_DATA_CHECKS_PARAMS
+
+
+
+
var SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST
+
+
+
+
var SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST
+
+
+
var SUBSIDIO_SPPO_RECURSOS_DATASET_ID
@@ -2119,6 +3241,10 @@

Class variables

+
var SUBSIDIO_SPPO_SECRET_PATH
+
+
+
var SUBSIDIO_SPPO_TABLE_ID
@@ -2298,9 +3424,13 @@

STU_MODE_MAPPING
  • STU_TABLE_CAPTURE_PARAMS
  • STU_TYPE_MAPPING
  • +
  • SUBSIDIO_SPPO_CODE_OWNERS
  • SUBSIDIO_SPPO_DASHBOARD_DATASET_ID
  • SUBSIDIO_SPPO_DASHBOARD_TABLE_ID
  • SUBSIDIO_SPPO_DATASET_ID
  • +
  • SUBSIDIO_SPPO_DATA_CHECKS_PARAMS
  • +
  • SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST
  • +
  • SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST
  • SUBSIDIO_SPPO_RECURSOS_DATASET_ID
  • SUBSIDIO_SPPO_RECURSOS_MATERIALIZACAO_PARAMS
  • SUBSIDIO_SPPO_RECURSOS_TABLE_IDS
  • @@ -2308,6 +3438,7 @@

    SUBSIDIO_SPPO_RECURSO_API_SECRET_PATH
  • SUBSIDIO_SPPO_RECURSO_CAPTURE_PARAMS
  • SUBSIDIO_SPPO_RECURSO_TABLE_CAPTURE_PARAMS
  • +
  • SUBSIDIO_SPPO_SECRET_PATH
  • SUBSIDIO_SPPO_TABLE_ID
  • TASK_MAX_RETRIES
  • TASK_RETRY_DELAY
  • diff --git a/rj_smtr/projeto_subsidio_sppo/flows.html b/rj_smtr/projeto_subsidio_sppo/flows.html index ce74179dc..90e623e57 100644 --- a/rj_smtr/projeto_subsidio_sppo/flows.html +++ b/rj_smtr/projeto_subsidio_sppo/flows.html @@ -33,7 +33,7 @@

    Module pipelines.rj_smtr.projeto_subsidio_sppo.flowsModule pipelines.rj_smtr.projeto_subsidio_sppo.flowsModule pipelines.rj_smtr.projeto_subsidio_sppo.flowsModule pipelines.rj_smtr.projeto_subsidio_sppo.flowsModule pipelines.rj_smtr.projeto_subsidio_sppo.flowsModule pipelines.rj_smtr.projeto_subsidio_sppo.flows
    +

    Functions

    +
    +
    +def SPPO_VEICULO_DIA_RUN_WAIT_FALSE() +
    +
    +
    +
    + +Expand source code + +
    lambda: [None], checkpoint=False, name="assign_none_to_previous_runs"
    +
    +
    +
    @@ -328,6 +372,11 @@

    Index

  • pipelines.rj_smtr.projeto_subsidio_sppo
  • +
  • Functions

    + +
  • diff --git a/rj_smtr/projeto_subsidio_sppo/tasks.html b/rj_smtr/projeto_subsidio_sppo/tasks.html index 5969cc281..43ae3a3cd 100644 --- a/rj_smtr/projeto_subsidio_sppo/tasks.html +++ b/rj_smtr/projeto_subsidio_sppo/tasks.html @@ -32,15 +32,158 @@

    Module pipelines.rj_smtr.projeto_subsidio_sppo.tasks + return param is None + + +@task +def subsidio_data_quality_check( + mode: str, params: dict, code_owners: list = None, check_params: dict = None +) -> bool: + """ + Verifica qualidade de dados para o processo de apuração de subsídio + + Args: + mode (str): Modo de execução (pre ou pos) + params (dict): Parameters for the checks + code_owners (list): Code owners to be notified + check_params (dict): queries and order columns for the checks + + Returns: + test_check (bool): True if all checks passed, False otherwise + """ + + if mode not in ["pre", "pos"]: + raise ValueError("Mode must be 'pre' or 'pos'") + + if check_params is None: + check_params = smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_PARAMS.value + + if code_owners is None: + code_owners = smtr_constants.SUBSIDIO_SPPO_CODE_OWNERS.value + + checks = dict() + + request_params = { + "start_timestamp": f"""{params["start_date"]} 00:00:00""", + "end_timestamp": ( + datetime.strptime(params["end_date"], "%Y-%m-%d") + timedelta(hours=27) + ).strftime("%Y-%m-%d %H:%M:%S"), + } + + if mode == "pos": + request_params["end_timestamp"] = f"""{params["end_date"]} 00:00:00""" + request_params[ + "dataset_id" + ] = smtr_constants.SUBSIDIO_SPPO_DASHBOARD_DATASET_ID.value + + checks_list = ( + smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST.value + if mode == "pre" + else smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST.value + ) + + for ( + table_id, + test_check_list, + ) in checks_list.items(): + checks[table_id] = perform_checks_for_table( + table_id, request_params, test_check_list, check_params + ) + + log(checks) + + date_range = ( + params["start_date"] + if params["start_date"] == params["end_date"] + else f'{params["start_date"]} a {params["end_date"]}' + ) + + webhook_url = get_vault_secret( + secret_path=smtr_constants.SUBSIDIO_SPPO_SECRET_PATH.value + )["data"]["discord_data_check_webhook"] + + test_check = all( + table["status"] for sublist in checks.values() for table in sublist + ) + + formatted_messages = [ + ":green_circle: " if test_check else ":red_circle: ", + f"**{mode.capitalize()}-Data Quality Checks - Apuração de Subsídio - {date_range}**\n\n", + ] + + if "general" in checks: + formatted_messages.extend( + f'{":white_check_mark:" if check["status"] else ":x:"} {check["desc"]}\n' + for check in checks["general"] + ) + + format_send_discord_message(formatted_messages, webhook_url) + + for table_id, checks_ in checks.items(): + if table_id != "general": + formatted_messages = [ + f"*{table_id}:*\n" + + "\n".join( + f'{":white_check_mark:" if check["status"] else ":x:"} {check["desc"]}' + for check in checks_ + ) + ] + format_send_discord_message(formatted_messages, webhook_url) + + formatted_messages = ["\n\n"] + + if mode == "pre": + formatted_messages.append( + "" + if test_check + else """:warning: **Status:** Necessidade de revisão dos dados de entrada!\n""" + ) + + if mode == "pos": + formatted_messages.append( + ":tada: **Status:** Sucesso" + if test_check + else ":warning: **Status:** Testes falharam. Necessidade de revisão dos dados finais!\n" + ) + + if not test_check: + at_code_owners = [ + f' - <@{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n' + if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"] == "user" + else f' - <@!{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n' + if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"] + == "user_nickname" + else f' - <#{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n' + if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"] == "channel" + else f' - <@&{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n' + for code_owner in code_owners + ] + + formatted_messages.extend(at_code_owners) + + format_send_discord_message(formatted_messages, webhook_url) + + return test_check
    @@ -67,6 +210,159 @@

    Functions

    return param is None

    +
    +def subsidio_data_quality_check(mode: str, params: dict, code_owners: list = None, check_params: dict = None) ‑> bool +
    +
    +

    Verifica qualidade de dados para o processo de apuração de subsídio

    +

    Args

    +
    +
    mode : str
    +
    Modo de execução (pre ou pos)
    +
    params : dict
    +
    Parameters for the checks
    +
    code_owners : list
    +
    Code owners to be notified
    +
    check_params : dict
    +
    queries and order columns for the checks
    +
    +

    Returns

    +

    test_check (bool): True if all checks passed, False otherwise

    +
    + +Expand source code + +
    @task
    +def subsidio_data_quality_check(
    +    mode: str, params: dict, code_owners: list = None, check_params: dict = None
    +) -> bool:
    +    """
    +    Verifica qualidade de dados para o processo de apuração de subsídio
    +
    +    Args:
    +        mode (str): Modo de execução (pre ou pos)
    +        params (dict): Parameters for the checks
    +        code_owners (list): Code owners to be notified
    +        check_params (dict): queries and order columns for the checks
    +
    +    Returns:
    +        test_check (bool): True if all checks passed, False otherwise
    +    """
    +
    +    if mode not in ["pre", "pos"]:
    +        raise ValueError("Mode must be 'pre' or 'pos'")
    +
    +    if check_params is None:
    +        check_params = smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_PARAMS.value
    +
    +    if code_owners is None:
    +        code_owners = smtr_constants.SUBSIDIO_SPPO_CODE_OWNERS.value
    +
    +    checks = dict()
    +
    +    request_params = {
    +        "start_timestamp": f"""{params["start_date"]} 00:00:00""",
    +        "end_timestamp": (
    +            datetime.strptime(params["end_date"], "%Y-%m-%d") + timedelta(hours=27)
    +        ).strftime("%Y-%m-%d %H:%M:%S"),
    +    }
    +
    +    if mode == "pos":
    +        request_params["end_timestamp"] = f"""{params["end_date"]} 00:00:00"""
    +        request_params[
    +            "dataset_id"
    +        ] = smtr_constants.SUBSIDIO_SPPO_DASHBOARD_DATASET_ID.value
    +
    +    checks_list = (
    +        smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_PRE_LIST.value
    +        if mode == "pre"
    +        else smtr_constants.SUBSIDIO_SPPO_DATA_CHECKS_POS_LIST.value
    +    )
    +
    +    for (
    +        table_id,
    +        test_check_list,
    +    ) in checks_list.items():
    +        checks[table_id] = perform_checks_for_table(
    +            table_id, request_params, test_check_list, check_params
    +        )
    +
    +    log(checks)
    +
    +    date_range = (
    +        params["start_date"]
    +        if params["start_date"] == params["end_date"]
    +        else f'{params["start_date"]} a {params["end_date"]}'
    +    )
    +
    +    webhook_url = get_vault_secret(
    +        secret_path=smtr_constants.SUBSIDIO_SPPO_SECRET_PATH.value
    +    )["data"]["discord_data_check_webhook"]
    +
    +    test_check = all(
    +        table["status"] for sublist in checks.values() for table in sublist
    +    )
    +
    +    formatted_messages = [
    +        ":green_circle: " if test_check else ":red_circle: ",
    +        f"**{mode.capitalize()}-Data Quality Checks - Apuração de Subsídio - {date_range}**\n\n",
    +    ]
    +
    +    if "general" in checks:
    +        formatted_messages.extend(
    +            f'{":white_check_mark:" if check["status"] else ":x:"} {check["desc"]}\n'
    +            for check in checks["general"]
    +        )
    +
    +    format_send_discord_message(formatted_messages, webhook_url)
    +
    +    for table_id, checks_ in checks.items():
    +        if table_id != "general":
    +            formatted_messages = [
    +                f"*{table_id}:*\n"
    +                + "\n".join(
    +                    f'{":white_check_mark:" if check["status"] else ":x:"} {check["desc"]}'
    +                    for check in checks_
    +                )
    +            ]
    +            format_send_discord_message(formatted_messages, webhook_url)
    +
    +    formatted_messages = ["\n\n"]
    +
    +    if mode == "pre":
    +        formatted_messages.append(
    +            ""
    +            if test_check
    +            else """:warning: **Status:** Necessidade de revisão dos dados de entrada!\n"""
    +        )
    +
    +    if mode == "pos":
    +        formatted_messages.append(
    +            ":tada: **Status:** Sucesso"
    +            if test_check
    +            else ":warning: **Status:** Testes falharam. Necessidade de revisão dos dados finais!\n"
    +        )
    +
    +    if not test_check:
    +        at_code_owners = [
    +            f'    - <@{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n'
    +            if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"] == "user"
    +            else f'    - <@!{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n'
    +            if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"]
    +            == "user_nickname"
    +            else f'    - <#{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n'
    +            if constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["type"] == "channel"
    +            else f'    - <@&{constants.OWNERS_DISCORD_MENTIONS.value[code_owner]["user_id"]}>\n'
    +            for code_owner in code_owners
    +        ]
    +
    +        formatted_messages.extend(at_code_owners)
    +
    +    format_send_discord_message(formatted_messages, webhook_url)
    +
    +    return test_check
    +
    +
    @@ -132,6 +428,7 @@

    Index

  • Functions

  • diff --git a/rj_smtr/tasks.html b/rj_smtr/tasks.html index 4e33a94f0..1bfb50d0d 100644 --- a/rj_smtr/tasks.html +++ b/rj_smtr/tasks.html @@ -40,7 +40,6 @@

    Module pipelines.rj_smtr.tasks

    from typing import Dict, List, Union, Iterable, Any import io - from basedosdados import Storage, Table import basedosdados as bd from dbt_client import DbtClient diff --git a/rj_smtr/utils.html b/rj_smtr/utils.html index efb1e0edc..58beb465c 100644 --- a/rj_smtr/utils.html +++ b/rj_smtr/utils.html @@ -43,26 +43,23 @@

    Module pipelines.rj_smtr.utils

    import io import json import zipfile +import time import pytz import requests import basedosdados as bd from basedosdados import Table, Storage from basedosdados.upload.datatypes import Datatype -import math import pandas as pd from google.cloud.storage.blob import Blob from google.cloud import bigquery import pymysql import psycopg2 import psycopg2.extras -import time - from prefect.schedules.clocks import IntervalClock from pipelines.constants import constants as emd_constants - from pipelines.rj_smtr.implicit_ftp import ImplicitFtpTls from pipelines.rj_smtr.constants import constants @@ -73,7 +70,6 @@

    Module pipelines.rj_smtr.utils

    get_redis_client, ) - # Set BD config to run on cloud # bd.config.from_file = True @@ -1158,7 +1154,116 @@

    Module pipelines.rj_smtr.utils

    log(f"Request concluído, tamanho dos dados: {len(data)}.") - return error, data, filetype
    + return error, data, filetype + + +def perform_check(desc: str, check_params: dict, request_params: dict) -> dict: + """ + Perform a check on a query + + Args: + desc (str): The check description + check_params (dict): The check parameters + * query (str): SQL query to be executed + * order_columns (list): order columns for query log results, in case of failure (optional) + request_params (dict): The request parameters + + Returns: + dict: The check status + """ + try: + q = check_params["query"].format(**request_params) + order_columns = check_params.get("order_columns", None) + except KeyError as e: + raise ValueError(f"Missing key in check_params: {e}") from e + + log(q) + df = bd.read_sql(q) + + check_status = df.empty + + check_status_dict = {"desc": desc, "status": check_status} + + log(f"Check status:\n{check_status_dict}") + + if not check_status: + log(f"Data info:\n{data_info_str(df)}") + log( + f"Sorted data:\n{df.sort_values(by=order_columns) if order_columns else df}" + ) + + return check_status_dict + + +def perform_checks_for_table( + table_id: str, request_params: dict, test_check_list: dict, check_params: dict +) -> dict: + """ + Perform checks for a table + + Args: + table_id (str): The table id + request_params (dict): The request parameters + test_check_list (dict): The test check list + check_params (dict): The check parameters + + Returns: + dict: The checks + """ + request_params["table_id"] = table_id + checks = list() + + for description, test_check in test_check_list.items(): + request_params["expression"] = test_check.get("expression", "") + checks.append( + perform_check( + description, + check_params.get(test_check.get("test", "expression_is_true")), + request_params | test_check.get("params", {}), + ) + ) + + return checks + + +def format_send_discord_message(formatted_messages: list, webhook_url: str): + """ + Format and send a message to discord + + Args: + formatted_messages (list): The formatted messages + webhook_url (str): The webhook url + + Returns: + None + """ + formatted_message = "".join(formatted_messages) + log(formatted_message) + msg_ext = len(formatted_message) + if msg_ext > 2000: + log( + f"** Message too long ({msg_ext} characters), will be split into multiple messages **" + ) + # Split message into lines + lines = formatted_message.split("\n") + message_chunks = [] + chunk = "" + for line in lines: + if len(chunk) + len(line) + 1 > 2000: # +1 for the newline character + message_chunks.append(chunk) + chunk = "" + chunk += line + "\n" + message_chunks.append(chunk) # Append the last chunk + for chunk in message_chunks: + send_discord_message( + message=chunk, + webhook_url=webhook_url, + ) + else: + send_discord_message( + message=formatted_message, + webhook_url=webhook_url, + )
    @@ -1866,6 +1971,64 @@

    Returns

    return data +
    +def format_send_discord_message(formatted_messages: list, webhook_url: str) +
    +
    +

    Format and send a message to discord

    +

    Args

    +
    +
    formatted_messages : list
    +
    The formatted messages
    +
    webhook_url : str
    +
    The webhook url
    +
    +

    Returns

    +

    None

    +
    + +Expand source code + +
    def format_send_discord_message(formatted_messages: list, webhook_url: str):
    +    """
    +    Format and send a message to discord
    +
    +    Args:
    +        formatted_messages (list): The formatted messages
    +        webhook_url (str): The webhook url
    +
    +    Returns:
    +        None
    +    """
    +    formatted_message = "".join(formatted_messages)
    +    log(formatted_message)
    +    msg_ext = len(formatted_message)
    +    if msg_ext > 2000:
    +        log(
    +            f"** Message too long ({msg_ext} characters), will be split into multiple messages **"
    +        )
    +        # Split message into lines
    +        lines = formatted_message.split("\n")
    +        message_chunks = []
    +        chunk = ""
    +        for line in lines:
    +            if len(chunk) + len(line) + 1 > 2000:  # +1 for the newline character
    +                message_chunks.append(chunk)
    +                chunk = ""
    +            chunk += line + "\n"
    +        message_chunks.append(chunk)  # Append the last chunk
    +        for chunk in message_chunks:
    +            send_discord_message(
    +                message=chunk,
    +                webhook_url=webhook_url,
    +            )
    +    else:
    +        send_discord_message(
    +            message=formatted_message,
    +            webhook_url=webhook_url,
    +        )
    +
    +
    def generate_df_and_save(data: dict, fname: pathlib.Path)
    @@ -2569,6 +2732,125 @@

    Args

    return data +
    +def perform_check(desc: str, check_params: dict, request_params: dict) ‑> dict +
    +
    +

    Perform a check on a query

    +

    Args

    +
    +
    desc : str
    +
    The check description
    +
    check_params : dict
    +
    The check parameters +* query (str): SQL query to be executed +* order_columns (list): order columns for query log results, in case of failure (optional)
    +
    request_params : dict
    +
    The request parameters
    +
    +

    Returns

    +
    +
    dict
    +
    The check status
    +
    +
    + +Expand source code + +
    def perform_check(desc: str, check_params: dict, request_params: dict) -> dict:
    +    """
    +    Perform a check on a query
    +
    +    Args:
    +        desc (str): The check description
    +        check_params (dict): The check parameters
    +            * query (str): SQL query to be executed
    +            * order_columns (list): order columns for query log results, in case of failure (optional)
    +        request_params (dict): The request parameters
    +
    +    Returns:
    +        dict: The check status
    +    """
    +    try:
    +        q = check_params["query"].format(**request_params)
    +        order_columns = check_params.get("order_columns", None)
    +    except KeyError as e:
    +        raise ValueError(f"Missing key in check_params: {e}") from e
    +
    +    log(q)
    +    df = bd.read_sql(q)
    +
    +    check_status = df.empty
    +
    +    check_status_dict = {"desc": desc, "status": check_status}
    +
    +    log(f"Check status:\n{check_status_dict}")
    +
    +    if not check_status:
    +        log(f"Data info:\n{data_info_str(df)}")
    +        log(
    +            f"Sorted data:\n{df.sort_values(by=order_columns) if order_columns else df}"
    +        )
    +
    +    return check_status_dict
    +
    +
    +
    +def perform_checks_for_table(table_id: str, request_params: dict, test_check_list: dict, check_params: dict) ‑> dict +
    +
    +

    Perform checks for a table

    +

    Args

    +
    +
    table_id : str
    +
    The table id
    +
    request_params : dict
    +
    The request parameters
    +
    test_check_list : dict
    +
    The test check list
    +
    check_params : dict
    +
    The check parameters
    +
    +

    Returns

    +
    +
    dict
    +
    The checks
    +
    +
    + +Expand source code + +
    def perform_checks_for_table(
    +    table_id: str, request_params: dict, test_check_list: dict, check_params: dict
    +) -> dict:
    +    """
    +    Perform checks for a table
    +
    +    Args:
    +        table_id (str): The table id
    +        request_params (dict): The request parameters
    +        test_check_list (dict): The test check list
    +        check_params (dict): The check parameters
    +
    +    Returns:
    +        dict: The checks
    +    """
    +    request_params["table_id"] = table_id
    +    checks = list()
    +
    +    for description, test_check in test_check_list.items():
    +        request_params["expression"] = test_check.get("expression", "")
    +        checks.append(
    +            perform_check(
    +                description,
    +                check_params.get(test_check.get("test", "expression_is_true")),
    +                request_params | test_check.get("params", {}),
    +            )
    +        )
    +
    +    return checks
    +
    +
    def read_raw_data(filepath: str, reader_args: dict = None) ‑> tuple[str, pandas.core.frame.DataFrame]
    @@ -2988,6 +3270,7 @@

    Index

  • execute_db_query
  • filter_data
  • filter_null
  • +
  • format_send_discord_message
  • generate_df_and_save
  • generate_execute_schedules
  • get_datetime_range
  • @@ -3000,6 +3283,8 @@

    Index

  • get_upload_storage_blob
  • log_critical
  • map_dict_keys
  • +
  • perform_check
  • +
  • perform_checks_for_table
  • read_raw_data
  • safe_cast
  • save_raw_local_func