Descricao: Adiciona a pipeline para calculo de viagens do subsídio

1. subsidio_sppo_planejado: Tratamento da operacao planejada da quinzena
2. subsidio_sppo_viagens: Calculo das viagens da quinzena

TODO:

• [ ] Adicionar opção para rodar as viagens da quinzena completa (ou um período de datas) a inves de um unico dia: na pratica, chamar run_dbt_model multiplas vezes

Descrição: Adiciona captura das APIs do SIGMOB e o flow de materialização do dataset br_rj_riodejaneiro_sigmob. As queries que serão materializadas se encontram no repo de queries da SMTR, e ainda estão em WIP. Algumas das tasks criadas podem ser incorporadas as tasks mais gerais relativas ao uso do DBT (i.e. pipelines.rj_smtr.br_rj_riodejaneiro_sigmob.run_dbt_schema). a task de backfill (build_incremental_model) ainda está num estado não tão refinado, testei usando a flag --vars da CLI do DBT para parametrizar as ranges de materialização, mas encontrei algumas complicações que pareceram desnecessárias, mas uma solução mais elegante ainda pode ser desenvolvida e incorporada as tasks do utils geral deste repositório. Ainda é necessário testar as tasks na cloud do Prefect, mas por enquanto estamos aguardando a finalização do desenvolvimento do SQL contido no repositorio de queries. A captura, porém, espera-se que já funcione. Além disso, a maioria dos parâmetros necessários para execução dos flows têm valores default definidos dentro de cada flow em vez de estarem sendo providos pelo schedule, caso isso não esteja de acordo com as melhores práticas, é fácil de lidar.

Changelog:

• new: pipelines/rj_smtr/init.py
• new: pipelines/rj_smtr/br_rj_riodejaneiro_sigmob/constants.py
• new: pipelines/rj_smtr/br_rj_riodejaneiro_sigmob/flows.py
• new: pipelines/rj_smtr/br_rj_riodejaneiro_sigmob/schedules.py
• new: pipelines/rj_smtr/br_rj_riodejaneiro_sigmob/tasks.py
• new: pipelines/rj_smtr/br_rj_riodejaneiro_sigmob/utils.py
• modified: pipelines/rj_smtr/tasks.py
• modified: pipelines/rj_smtr/utils.py

(1) Refatoração do flow captura_sppo_v2: Ajustes para uso da captura_sppo_v2 na recaptura *

Changelog:

• adição de 2 parâmetros opcionais de recaptura - bool indicando se é recaptura + erro que gerou a recaptura
• modificação do parametro timestamp - quando há recaptura, utiliza parametro opcional datetime_filter, cc usa timestamp atual; fixa formatacao em isoformat
• upload_logs_to_bq: mantém log do erro anterior que foi recapturado
• run_config: limita uso de cpu e memoria

(2) Refatoração do flow recaptura: Ajustes para uso da captura_sppo_v2 na recaptura

Changelog:

• create_flow_run: usa o flow de captura_sppo_v2 para criar múltiplas runs do mesmo, substituindo o uso do método .map em cada task
• query_logs: aumenta limite de 40 para 60 recapturas por run; retorna dicionário com timestamp e erro + flag de recaptura, novos parametros que são passados no create_flow_run para a captura_sppo_v2
• run_config: limita uso de cpu e memoria

(3) Refatoração do flow materialize: Corrije atraso na materialização quando há falhas

Changelog:

• get_materialization_date_range: parametriza o delay da materialização de acordo com o flow + varia start_ts para preencher buracos de materialização (caso haja algum), fixando end_ts com base no delay_hour

• Detalhamento do flow de captura:

• Criação do flow subsidio_sppo_apuracao com o objetivo de criação das views do subsídio no schema homônimo;
• Ajuste da task get_run_dates para adaptar às necessidades do flow subsidio_sppo_apuracao e utilizar a data atual considerando a timezone (estava considerado UTC);
• Teste do flow smtr_materialize_to_datario_viagem_sppo_flow. Em funcionamento com ou sem parâmetros;
• Teste do subsidio_sppo_apuracao. Em funcionamento com ou sem parâmetros.

Adiciona a pipeline para incorporar realocação de linhas, realizadas pelas empresas operadoras, visando corrigir o relacionamento entre linhas e veículos. Consequentemente, impacta na identificação de viagens e no subsídio.

Descrição: PR para migração da pipeline de captura dos arquivos de RDO/RHO que hoje rodam no dagster. As funcionalidades são basicamente as mesmas, porém com alguma adaptação nas estruturas de saídas das funções para simplificar o código e adequar aos paradigmas do prefect. No primeiro commit, está sendo adicionada uma versão funcional para a captura do RHO do SPPO, única tabela que ainda não temos no BigQuery. O flow é parametrizado de forma genérica para que possamos usar a mesma definição, apenas variando parâmetros, para as capturas do RDO pro SPPO e dos RHO/RDO do STPL. TODO:

• [x] Adicionar schedules para os outros arquivos
• [ ] Ajustar os table_id's em rj-smtr.constants
• [x] Adicionar os secrets para conexão com FTP ao vault
• [ ] Definir a frequência com a qual cada flow será lançado
• [ ] Dimensionar limites e requisições de memórias para o deploy

Descrição: Adiciona os flows de captura do RDO/RHO para os modais SPPO e STPL via FTP. A captura funciona buscando dentro do FTP os diretórios relativos ao modal e então, buscando todos os arquivos cujo nome inicie com o tipo do relatório ( parâmetro report type). A busca é limitada por padrão (parâmetro dump=False) para apenas arquivos entre o (datetime atual + 11,5 horas) e as 24h anteriores a isso, porém o flow pode ser utilizado também para realização de um dump de todos os arquivos presentes num dado momento (limitados a partir do ano de 2022). Após isso, seguimos com a sequência padrão de download local dos arquivos (originalmente em txt) e posterior pré tratamento para padronização de colunas e conversão para csv, seguido de upload para o bucket, sem upload de logs no momento. Devido às restrições impostas pelo uso de tabelas do tipo external_table e de não termos ainda pronto a infra para fácil geração das biglake tables, após a captura se segue uma etapa de materialização.

Descrição: Migração da captura de GPS do SPPO para o prefect. A estrutura é basicamente idêntica à que está em https://github.com/RJ-SMTR/maestro/tree/main/repositories/capturas/br_rj_riodejaneiro_onibus_gps. Utiliza as tasks já testadas para a captura de gps do STPL em pipelines/rj-smtr/tasks.py, adicionando aos utils da SMTR uma função log_critical, que basicamente serve para postar mensagens no discord da SMTR visando diminuir o flood no servidor do escritório. Adiciona uma nova checagem do argumento kind em pipelines/rj_smtr/tasks.py::get_raw() para usar uma nova estrutura de autenticação.

TODO:

• [x] Adicionar schedules
• [x] Adicionar webhook do canal critical aos secrets
• [ ] Testar a postagem em caso de falha

Closes #126

Changelog:

• [x] registros_logs: Adicionar busca pelos minutos sem timestamp_captura (além de erros capturados nos logs)
• [x] upload_logs: Corrige status para registro dos logs

TODO:

• [x] Merge de ajustes nas falhas de captura https://github.com/prefeitura-rio/pipelines/pull/127
• [ ] Alterar schedule para ~2:05
• [ ] Testar a pipeline completa (recaptura + materialização)

• Create function generate_dump_url_schedules to do the same as generate_dump_db_schedules, but in the url context;
• Create flow and schedules to dump data from Google Sheets and use in the project SMFP - Dashboard de Metas

Descrição

O flow de materialização é um run_dbt_model padrão porém recebendo as variáveis run_dates (sempre em uma lista), para que possa materializar a tabela com base direta na data de partição usada para a captura, além de não possuir schedule próprio e somente ser ativada quando a captura baixar pelo menos um arquivo e fazer seu upload sem erros

Recursos

• Queries: (...)

Código para testar as tasks

from pipelines.utils.policy_matrix.tasks import *

project_ids = ["rj-escritorio-dev"]

discovery_api = get_discovery_api.run(mode="prod")
policies = []
for project_id in project_ids:
    policies.append(get_iam_policy.run(project_id=project_id, discovery_api=discovery_api))
policies = merge_iam_policies.run(project_ids=project_ids, policies=policies)
roles_matrix = generate_roles_matrix.run(policies=policies)
dataframe = roles_matrix_to_pandas_dataframe.run(roles_matrix=roles_matrix)

Tarefas

• [x] Adicionar parâmetros
• [ ] Aguardar merge de https://github.com/basedosdados/mais/pull/1469
• [ ] Fazer upgrade da versão do basedosdados
• [ ] Resolver problemas de linting