Cria validação dos dados de integração da Jaé com base na Matriz da S…

…MTR (#371) * cria nova validação de integrações * add data_inicio_matriz * cria tabela matriz_integracao com base na matriz da smtr * corrige integração com transação duplicada * cria validação das integrações usando a matriz da smtr * cria flow de materialização da nova validação dos dados da Jaé * schedule_cron opcional * cria materialização da matriz de integração da smtr * add matriz_integracao_smtr / add validacao_dados_jae * add validacao_dados_jae * wait integracao e transacao * add change log pr 371 * ajusta schedule * corrige mensagem de erro * descomenta refs --------- Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
prefeitura-rio · Dec 30, 2024 · ee19094 · ee19094
1 parent 3081fd0
commit ee19094
Show file tree

Hide file tree

Showing 30 changed files with 866 additions and 501 deletions.
diff --git a/pipelines/flows.py b/pipelines/flows.py
@@ -26,3 +26,4 @@
 from pipelines.treatment.datario.flows import *  # noqa
 from pipelines.treatment.monitoramento.flows import *  # noqa
 from pipelines.treatment.planejamento.flows import *  # noqa
+from pipelines.treatment.validacao_dados_jae.flows import *  # noqa
diff --git a/pipelines/migration/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md b/pipelines/migration/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md
@@ -1,5 +1,10 @@
 # Changelog - br_rj_riodejaneiro_bilhetagem
 
+## [1.4.9] - 2024-12-30
+
+### Alterado
+- Desativa schedule do flow `bilhetagem_validacao_jae` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
+
 ## [1.4.8] - 2024-12-16
 
 ### Alterado

diff --git a/pipelines/migration/br_rj_riodejaneiro_bilhetagem/flows.py b/pipelines/migration/br_rj_riodejaneiro_bilhetagem/flows.py
@@ -33,10 +33,9 @@
     rename_current_flow_run_now_time,
 )
 from pipelines.migration.utils import set_default_parameters
-from pipelines.schedules import (
+from pipelines.schedules import (  # every_day_hour_seven,
     every_5_minutes,
     every_day_hour_five,
-    every_day_hour_seven,
     every_hour,
     every_minute,
 )
@@ -354,7 +353,7 @@
     handler_skip_if_running,
 ]
 
-bilhetagem_validacao_jae.schedule = every_day_hour_seven
+# bilhetagem_validacao_jae.schedule = every_day_hour_seven
 
 
 # RECAPTURA #

diff --git a/pipelines/migration/br_rj_riodejaneiro_gtfs/CHANGELOG.md b/pipelines/migration/br_rj_riodejaneiro_gtfs/CHANGELOG.md
@@ -1,5 +1,9 @@
 # Changelog - gtfs
 
+## [1.1.8] - 2024-12-30
+
+### Alterado
+- Exclui modelo `matriz_integracao` da materialização (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
 
 ## [1.1.7] - 2024-12-13
 

diff --git a/pipelines/migration/br_rj_riodejaneiro_gtfs/flows.py b/pipelines/migration/br_rj_riodejaneiro_gtfs/flows.py
@@ -238,7 +238,8 @@
             + " "
             + constants.PLANEJAMENTO_MATERIALIZACAO_DATASET_ID.value,
             _vars=dbt_vars,
-            exclude="calendario aux_calendario_manual viagem_planejada_planejamento",
+            exclude="calendario aux_calendario_manual viagem_planejada_planejamento \
+matriz_integracao",
         ).set_upstream(task=wait_captura)
 
         run_dbt_failed = task_value_is_none(wait_run_dbt_model)

diff --git a/pipelines/treatment/datario/CHANGELOG.md b/pipelines/treatment/datario/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Changelog - datario
 
+## [1.0.1] - 2024-12-30
+
+### Alterado
+
+- Remove parâmetro `schedule_cron` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
+
 ## [1.0.0] - 2024-12-16
 
 ### Adicionado

diff --git a/pipelines/treatment/datario/constants.py b/pipelines/treatment/datario/constants.py
@@ -6,7 +6,6 @@
 from datetime import datetime
 from enum import Enum
 
-from pipelines.schedules import create_daily_cron
 from pipelines.treatment.templates.utils import DBTSelector
 
 
@@ -17,6 +16,5 @@ class constants(Enum):  # pylint: disable=c0103
 
     DATARIO_SELECTOR = DBTSelector(
         name="datario",
-        schedule_cron=create_daily_cron(hour=0),
         initial_datetime=datetime(2024, 12, 16),
     )
diff --git a/pipelines/treatment/planejamento/CHANGELOG.md b/pipelines/treatment/planejamento/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Changelog - planejamento
 
+## [1.1.0] - 2024-12-30
+
+### Adicionado
+
+- Cria flow de materialização da matriz de integração da SMTR (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
+
 ## [1.0.1] - 2024-11-25
 
 ### Alterado

diff --git a/pipelines/treatment/planejamento/constants.py b/pipelines/treatment/planejamento/constants.py
@@ -20,3 +20,8 @@ class constants(Enum):  # pylint: disable=c0103
         schedule_cron=create_daily_cron(hour=1),
         initial_datetime=datetime(2024, 9, 1, 0, 0, 0),
     )
+
+    MATRIZ_INTEGRACAO_SMTR_SELECTOR = DBTSelector(
+        name="matriz_integracao_smtr",
+        initial_datetime=datetime(2024, 12, 30, 0, 0, 0),
+    )
diff --git a/pipelines/treatment/planejamento/flows.py b/pipelines/treatment/planejamento/flows.py
@@ -14,3 +14,10 @@
     selector=constants.PLANEJAMENTO_DIARIO_SELECTOR.value,
     agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
 )
+
+MATRIZ_INTEGRACAO_SMTR_MATERIALIZACAO = create_default_materialization_flow(
+    flow_name="matriz_integracao_smtr - materializacao",
+    selector=constants.MATRIZ_INTEGRACAO_SMTR_SELECTOR.value,
+    agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
+    generate_schedule=False,
+)
diff --git a/pipelines/treatment/templates/utils.py b/pipelines/treatment/templates/utils.py
@@ -39,8 +39,8 @@ class DBTSelector:
     def __init__(
         self,
         name: str,
-        schedule_cron: str,
         initial_datetime: datetime,
+        schedule_cron: str = None,
         incremental_delay_hours: int = 0,
     ):
         self.name = name
@@ -98,6 +98,8 @@ def is_up_to_date(self, env: str, timestamp: datetime) -> bool:
         Returns:
             bool: se está atualizado ou não
         """
+        if self.schedule_cron is None:
+            raise ValueError("O selector não possui agendamento")
         last_materialization = self.get_last_materialized_datetime(env=env)
 
         last_schedule = cron_get_last_date(cron_expr=self.schedule_cron, timestamp=timestamp)
@@ -115,6 +117,8 @@ def get_next_schedule_datetime(self, timestamp: datetime) -> datetime:
         Returns:
             datetime: próximo datetime do cron
         """
+        if self.schedule_cron is None:
+            raise ValueError("O selector não possui agendamento")
         return cron_get_next_date(cron_expr=self.schedule_cron, timestamp=timestamp)
 
     def set_redis_materialized_datetime(self, env: str, timestamp: datetime):

diff --git a/pipelines/treatment/validacao_dados_jae/CHANGELOG.md b/pipelines/treatment/validacao_dados_jae/CHANGELOG.md
@@ -0,0 +1,7 @@
+# Changelog - validacao_dados_jae
+
+## [1.0.0] - 2024-12-30
+
+### Adicionado
+
+- Cria flow de materialização da validação dos dados da Jaé (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
diff --git a/pipelines/treatment/validacao_dados_jae/__init__.py b/pipelines/treatment/validacao_dados_jae/__init__.py
diff --git a/pipelines/treatment/validacao_dados_jae/constants.py b/pipelines/treatment/validacao_dados_jae/constants.py
@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+"""
+Valores constantes para materialização da validação dos dados da Jaé
+"""
+
+from datetime import datetime
+from enum import Enum
+
+from pipelines.schedules import create_daily_cron
+from pipelines.treatment.templates.utils import DBTSelector
+
+
+class constants(Enum):  # pylint: disable=c0103
+    """
+    Valores constantes para materialização da validação dos dados da Jaé
+    """
+
+    VALIDACAO_DADOS_JAE_SELECTOR = DBTSelector(
+        name="validacao_dados_jae",
+        schedule_cron=create_daily_cron(hour=7),
+        initial_datetime=datetime(2024, 12, 30, 0, 0, 0),
+    )
diff --git a/pipelines/treatment/validacao_dados_jae/flows.py b/pipelines/treatment/validacao_dados_jae/flows.py
@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""
+Flows de tratamento dos dados da validação dos dados da Jaé
+"""
+
+from pipelines.constants import constants as smtr_constants
+from pipelines.migration.br_rj_riodejaneiro_bilhetagem.constants import (
+    constants as bilhetagem_constants,
+)
+from pipelines.schedules import create_daily_cron, create_hourly_cron
+from pipelines.treatment.templates.flows import create_default_materialization_flow
+from pipelines.treatment.validacao_dados_jae.constants import constants
+
+transacao_materializacao_params = (
+    bilhetagem_constants.BILHETAGEM_MATERIALIZACAO_TRANSACAO_PARAMS.value
+)
+
+integracao_materializacao_params = (
+    bilhetagem_constants.BILHETAGEM_MATERIALIZACAO_INTEGRACAO_PARAMS.value
+)
+
+VALIDACAO_DADOS_JAE_MATERIALIZACAO = create_default_materialization_flow(
+    flow_name="validacao_dados_jae - materializacao",
+    selector=constants.VALIDACAO_DADOS_JAE_SELECTOR.value,
+    agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
+    wait=[
+        {
+            "redis_key": f"{integracao_materializacao_params['dataset_id']}\
+.{integracao_materializacao_params['table_id']}",
+            "dict_key": "last_run_timestamp",
+            "datetime_format": "%Y-%m-%dT%H:%M:%S",
+            "delay_hours": integracao_materializacao_params["dbt_vars"]["date_range"][
+                "delay_hours"
+            ],
+            "schedule_cron": create_daily_cron(hour=5),
+        },
+        {
+            "redis_key": f"{transacao_materializacao_params['dataset_id']}\
+.{transacao_materializacao_params['table_id']}",
+            "dict_key": "last_run_timestamp",
+            "datetime_format": "%Y-%m-%dT%H:%M:%S",
+            "delay_hours": transacao_materializacao_params["dbt_vars"]["date_range"]["delay_hours"],
+            "schedule_cron": create_hourly_cron(),
+        },
+    ],
+)
diff --git a/queries/models/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md b/queries/models/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md
@@ -1,5 +1,14 @@
 # Changelog - bilhetagem
 
+## [2.0.2] - 2024-12-30
+
+### Removido
+- Move `matriz_integracao.sql` para o planejamento (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
+
+
+### Alterado
+- Muda deduplicação de integração para considerar integrações diferentes que compartilham transações nos modelos `transacao.sql` e `integracao.sql` (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/371)
+
 ## [2.0.1] - 2024-12-12
 
 ### Alterado