Cria pipeline de captura dos dados de retorno da Caixa

pipelines/rj_smtr/br_rj_riodejaneiro_bilhetagem/CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog - br_rj_riodejaneiro_bilhetagem
 
+## [1.0.2] - 2024-05-28
+
+### Alterado
+
+- Remove schedule do flow `bilhetagem_validacao_jae`, por conta da arquitetura das tabelas que será alterado (https://github.com/prefeitura-rio/pipelines/pull/691)
+
 ## [1.0.1] - 2024-05-22
 
 ### Corrigido

pipelines/rj_smtr/br_rj_riodejaneiro_bilhetagem/flows.py

@@ -40,7 +40,7 @@
     every_hour,
     every_minute,
     every_day_hour_five,
-    every_day_hour_seven,
+    # every_day_hour_seven,
     every_5_minutes,
 )
 
@@ -322,7 +322,7 @@
     default_parameters=constants.BILHETAGEM_MATERIALIZACAO_VALIDACAO_JAE_PARAMS.value,
 )
 
-bilhetagem_validacao_jae.schedule = every_day_hour_seven
+# bilhetagem_validacao_jae.schedule = every_day_hour_seven
 
 
 # RECAPTURA #

pipelines/rj_smtr/constants.py

@@ -1561,21 +1561,3 @@ class constants(Enum):  # pylint: disable=c0103
     ZIRIX_BASE_URL = "https://integration.systemsatx.com.br/Globalbus/SMTR"
 
     CONTROLE_FINANCEIRO_DATASET_ID = "controle_financeiro"
-
-    CONTROLE_FINANCEIRO_BASE_URL = "https://docs.google.com/spreadsheets/d/1QVfa9b8jzpQr3gac0FIlozmTaVeArtJROA343A2lMVM/\
-export?format=csv&gid="
-
-    CONTROLE_FINANCEIRO_CAPTURE_DEFAULT_PARAMS = {
-        "dataset_id": CONTROLE_FINANCEIRO_DATASET_ID,
-        "source_type": "api-csv",
-        "partition_date_only": True,
-    }
-
-    CONTROLE_FINANCEIRO_CB_CAPTURE_PARAMS = {
-        "extract_params": {"sheet_id": "454453523"},
-        "table_id": "cb",
-    }
-    CONTROLE_FINANCEIRO_CETT_CAPTURE_PARAMS = {
-        "extract_params": {"sheet_id": "0"},
-        "table_id": "cett",
-    }

pipelines/rj_smtr/controle_financeiro/CHANGELOG.md

@@ -1,5 +1,16 @@
 # Changelog - controle_financeiro
 
+
+## [1.1.0] - 2024-05-28
+
+### Adicionado
+
+- Cria flow de captura do arquivo de retorno da Caixa, enviado via api pela CCT (https://github.com/prefeitura-rio/pipelines/pull/691)
+
+### Modificado
+
+- Move as constantes dos flows `controle_cct_cb_captura` e `controle_cct_cett_captura` para o arquivo de constantes dentro da pasta do dataset (https://github.com/prefeitura-rio/pipelines/pull/691)
+
 ## [1.0.0] - 2024-05-22
 
 ### Adicionado

pipelines/rj_smtr/controle_financeiro/constants.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+"""
+Constant values for rj_smtr controle_financeiro
+"""
+
+from enum import Enum
+from pipelines.rj_smtr.constants import constants as smtr_constants
+
+
+class constants(Enum):  # pylint: disable=c0103
+    """
+    Constant values for rj_smtr controle_financeiro
+    """
+
+    SHEETS_BASE_URL = "https://docs.google.com/spreadsheets/d/\
+1QVfa9b8jzpQr3gac0FIlozmTaVeArtJROA343A2lMVM/export?format=csv&gid="
+
+    SHEETS_CAPTURE_DEFAULT_PARAMS = {
+        "dataset_id": smtr_constants.CONTROLE_FINANCEIRO_DATASET_ID.value,
+        "source_type": "api-csv",
+        "partition_date_only": True,
+    }
+
+    SHEETS_CB_CAPTURE_PARAMS = {
+        "extract_params": {
+            "sheet_id": "454453523",
+            "base_url": SHEETS_BASE_URL,
+        },
+        "table_id": "cb",
+    }
+    SHEETS_CETT_CAPTURE_PARAMS = {
+        "extract_params": {
+            "sheet_id": "0",
+            "base_url": SHEETS_BASE_URL,
+        },
+        "table_id": "cett",
+    }
+
+    ARQUIVO_RETORNO_TABLE_ID = "arquivo_retorno"
+
+    CCT_API_SECRET_PATH = "cct_api"
+    CCT_API_BASE_URL = "https://api.cct.mobilidade.rio/api/v1"

pipelines/rj_smtr/controle_financeiro/flows.py

@@ -8,21 +8,46 @@
 from copy import deepcopy
 from prefect.run_configs import KubernetesRun
 from prefect.storage import GCS
+from prefect import Parameter
 
 # EMD Imports #
 
 from pipelines.constants import constants as emd_constants
-
+from pipelines.utils.decorators import Flow
+from pipelines.utils.tasks import (
+    rename_current_flow_run_now_time,
+    get_now_time,
+    get_current_flow_labels,
+    get_current_flow_mode,
+)
+from pipelines.utils.utils import set_default_parameters
 
 # SMTR Imports #
 
 from pipelines.rj_smtr.flows import (
     default_capture_flow,
 )
-from pipelines.rj_smtr.constants import constants
+from pipelines.rj_smtr.tasks import (
+    get_current_timestamp,
+    create_date_hour_partition,
+    parse_timestamp_to_string,
+    create_local_partition_path,
+    upload_raw_data_to_gcs,
+    transform_raw_to_nested_structure,
+    upload_staging_data_to_gcs,
+)
 
-from pipelines.utils.utils import set_default_parameters
-from pipelines.rj_smtr.schedules import every_day
+from pipelines.rj_smtr.constants import constants as smtr_constants
+
+from pipelines.rj_smtr.schedules import every_day, every_friday_seven_thirty
+
+from pipelines.rj_smtr.controle_financeiro.constants import constants
+from pipelines.rj_smtr.controle_financeiro.tasks import (
+    get_cct_arquivo_retorno_redis_key,
+    create_cct_arquivo_retorno_params,
+    get_raw_cct_arquivo_retorno,
+    cct_arquivo_retorno_save_redis,
+)
 
 
 # Flows #
@@ -37,8 +62,8 @@
 
 controle_cct_cb_captura = set_default_parameters(
     flow=controle_cct_cb_captura,
-    default_parameters=constants.CONTROLE_FINANCEIRO_CAPTURE_DEFAULT_PARAMS.value
-    | constants.CONTROLE_FINANCEIRO_CB_CAPTURE_PARAMS.value,
+    default_parameters=constants.SHEETS_CAPTURE_DEFAULT_PARAMS.value
+    | constants.SHEETS_CB_CAPTURE_PARAMS.value,
 )
 controle_cct_cb_captura.schedule = every_day
 
@@ -52,7 +77,91 @@
 
 controle_cct_cett_captura = set_default_parameters(
     flow=controle_cct_cett_captura,
-    default_parameters=constants.CONTROLE_FINANCEIRO_CAPTURE_DEFAULT_PARAMS.value
-    | constants.CONTROLE_FINANCEIRO_CETT_CAPTURE_PARAMS.value,
+    default_parameters=constants.SHEETS_CAPTURE_DEFAULT_PARAMS.value
+    | constants.SHEETS_CETT_CAPTURE_PARAMS.value,
 )
 controle_cct_cett_captura.schedule = every_day
+
+
+with Flow(
+    "SMTR: Controle Financeiro Arquivo Retorno - Captura",
+    code_owners=["caio", "fernanda", "boris", "rodrigo", "rafaelpinheiro"],
+) as arquivo_retorno_captura:
+    start_date = Parameter("start_date", default=None)
+    end_date = Parameter("end_date", default=None)
+
+    rename_flow_run = rename_current_flow_run_now_time(
+        prefix=f"SMTR: Captura {constants.ARQUIVO_RETORNO_TABLE_ID.value}: ",
+        now_time=get_now_time(),
+    )
+    LABELS = get_current_flow_labels()
+    MODE = get_current_flow_mode(LABELS)
+
+    timestamp = get_current_timestamp()
+
+    REDIS_KEY = get_cct_arquivo_retorno_redis_key(mode=MODE)
+
+    headers, params = create_cct_arquivo_retorno_params(
+        redis_key=REDIS_KEY,
+        start_date=start_date,
+        end_date=end_date,
+    )
+
+    partitions = create_date_hour_partition(
+        timestamp,
+        partition_date_only=True,
+    )
+
+    filename = parse_timestamp_to_string(timestamp)
+
+    filepath = create_local_partition_path(
+        dataset_id=smtr_constants.CONTROLE_FINANCEIRO_DATASET_ID.value,
+        table_id=constants.ARQUIVO_RETORNO_TABLE_ID.value,
+        filename=filename,
+        partitions=partitions,
+    )
+
+    raw_filepath = get_raw_cct_arquivo_retorno(
+        headers=headers,
+        params=params,
+        local_filepath=filepath,
+    )
+
+    error = upload_raw_data_to_gcs(
+        error=None,
+        raw_filepath=raw_filepath,
+        table_id=constants.ARQUIVO_RETORNO_TABLE_ID.value,
+        dataset_id=smtr_constants.CONTROLE_FINANCEIRO_DATASET_ID.value,
+        partitions=partitions,
+    )
+
+    error, staging_filepath = transform_raw_to_nested_structure(
+        raw_filepath=raw_filepath,
+        filepath=filepath,
+        error=error,
+        timestamp=timestamp,
+        primary_key=["id"],
+    )
+
+    staging_upload = upload_staging_data_to_gcs(
+        error=error,
+        staging_filepath=staging_filepath,
+        timestamp=timestamp,
+        table_id=constants.ARQUIVO_RETORNO_TABLE_ID.value,
+        dataset_id=smtr_constants.CONTROLE_FINANCEIRO_DATASET_ID.value,
+        partitions=partitions,
+    )
+
+    cct_arquivo_retorno_save_redis(
+        redis_key=REDIS_KEY,
+        raw_filepath=raw_filepath,
+        upstream_tasks=[staging_upload],
+    )
+
+arquivo_retorno_captura.storage = GCS(emd_constants.GCS_FLOWS_BUCKET.value)
+arquivo_retorno_captura.run_config = KubernetesRun(
+    image=emd_constants.DOCKER_IMAGE.value,
+    labels=[emd_constants.RJ_SMTR_DEV_AGENT_LABEL.value],
+)
+
+arquivo_retorno_captura.schedule = every_friday_seven_thirty