Merge branch 'master' into staging/sme_frequencia

Author: mergify[bot]

pipelines/rj_smtr/br_rj_riodejaneiro_bilhetagem/flows.py

@@ -63,6 +63,25 @@
 
 bilhetagem_transacao_captura.schedule = every_minute
 
+
+bilhetagem_transacao_riocard_captura = deepcopy(default_capture_flow)
+bilhetagem_transacao_riocard_captura.name = (
+    "SMTR: Bilhetagem Transação RioCard - Captura"
+)
+bilhetagem_transacao_riocard_captura.storage = GCS(emd_constants.GCS_FLOWS_BUCKET.value)
+bilhetagem_transacao_riocard_captura.run_config = KubernetesRun(
+    image=emd_constants.DOCKER_IMAGE.value,
+    labels=[emd_constants.RJ_SMTR_AGENT_LABEL.value],
+)
+
+bilhetagem_transacao_riocard_captura = set_default_parameters(
+    flow=bilhetagem_transacao_riocard_captura,
+    default_parameters=constants.BILHETAGEM_GENERAL_CAPTURE_DEFAULT_PARAMS.value
+    | constants.BILHETAGEM_TRANSACAO_RIOCARD_CAPTURE_PARAMS.value,
+)
+
+bilhetagem_transacao_riocard_captura.schedule = every_minute
+
 # BILHETAGEM FISCALIZAÇÃO - CAPTURA A CADA 5 MINUTOS #
 
 bilhetagem_fiscalizacao_captura = deepcopy(default_capture_flow)
@@ -179,6 +198,26 @@
     default_parameters=bilhetagem_materializacao_transacao_parameters,
 )
 
+
+bilhetagem_materializacao_transacao_riocard = deepcopy(default_materialization_flow)
+bilhetagem_materializacao_transacao_riocard.name = (
+    "SMTR: Bilhetagem Transação RioCard - Materialização"
+)
+bilhetagem_materializacao_transacao_riocard.storage = GCS(
+    emd_constants.GCS_FLOWS_BUCKET.value
+)
+bilhetagem_materializacao_transacao_riocard.run_config = KubernetesRun(
+    image=emd_constants.DOCKER_IMAGE.value,
+    labels=[emd_constants.RJ_SMTR_AGENT_LABEL.value],
+)
+
+bilhetagem_materializacao_transacao_riocard = set_default_parameters(
+    flow=bilhetagem_materializacao_transacao_riocard,
+    default_parameters=constants.BILHETAGEM_MATERIALIZACAO_TRANSACAO_RIOCARD_PARAMS.value,
+)
+
+bilhetagem_materializacao_transacao_riocard.schedule = every_day_hour_five
+
 # Ordem Pagamento
 
 bilhetagem_materializacao_ordem_pagamento = deepcopy(default_materialization_flow)
@@ -318,6 +357,21 @@
             raise_final_state=True,
         )
 
+        run_recaptura_transacao_riocard = create_flow_run(
+            flow_name=bilhetagem_recaptura.name,
+            project_name=emd_constants.PREFECT_DEFAULT_PROJECT.value,
+            labels=LABELS,
+            parameters=constants.BILHETAGEM_TRANSACAO_RIOCARD_CAPTURE_PARAMS.value,
+            upstream_tasks=[wait_recaptura_transacao_true],
+        )
+
+        wait_recaptura_transacao_riocard_true = wait_for_flow_run(
+            run_recaptura_transacao_riocard,
+            stream_states=True,
+            stream_logs=True,
+            raise_final_state=True,
+        )
+
         # Recaptura Fiscalização
 
         run_recaptura_fiscalizacao = create_flow_run(

pipelines/rj_smtr/constants.py

@@ -777,6 +777,25 @@ class constants(Enum):  # pylint: disable=c0103
         "interval_minutes": 1,
     }
 
+    BILHETAGEM_TRANSACAO_RIOCARD_CAPTURE_PARAMS = {
+        "table_id": "transacao_riocard",
+        "partition_date_only": False,
+        "extract_params": {
+            "database": "transacao_db",
+            "query": """
+                SELECT
+                    *
+                FROM
+                    transacao_riocard
+                WHERE
+                    data_processamento >= '{start}'
+                    AND data_processamento < '{end}'
+            """,
+        },
+        "primary_key": ["id"],
+        "interval_minutes": 1,
+    }
+
     BILHETAGEM_FISCALIZACAO_CAPTURE_PARAMS = {
         "table_id": "fiscalizacao",
         "partition_date_only": False,
@@ -1210,6 +1229,17 @@ class constants(Enum):  # pylint: disable=c0103
         "exclude": "integracao matriz_integracao stops_gtfs2 routes_gtfs2 feed_info_gtfs2",
     }
 
+    BILHETAGEM_MATERIALIZACAO_TRANSACAO_RIOCARD_PARAMS = {
+        "dataset_id": "dashboard_controle_vinculo_jae_riocard",
+        "table_id": "veiculo_indicadores_dia",
+        "upstream": True,
+        "dbt_vars": {
+            "run_date": {},
+            "version": {},
+        },
+        "exclude": "+gps_sppo +sppo_licenciamento +gps_validador",
+    }
+
     BILHETAGEM_MATERIALIZACAO_ORDEM_PAGAMENTO_PARAMS = {
         "dataset_id": BILHETAGEM_DATASET_ID,
         "table_id": "ordem_pagamento_dia",

pipelines/rj_smtr/tasks.py

@@ -224,7 +224,7 @@ def create_dbt_run_vars(
             day_datetime=timestamp,
         )
 
-        final_vars.append([d.copy() for d in date_var])
+        final_vars += [d.copy() for d in date_var]
 
         log(f"run_date created: {date_var}")