[WIP] Cria captura de tickets de atendimento do MoviDesk

pipelines/capture/movidesk/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog - source_movidesk
+
+## [1.0.0] - 2024-12-26
+
+### Adicionado
+
+- Cria flow de captura dos tickets do MoviDesk (https://github.com/prefeitura-rio/pipelines_rj_smtr/pull/) # adicionar numero PR

pipelines/capture/movidesk/constants.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+"""
+Valores constantes para captura de dados da Rio Ônibus
+"""
+
+from datetime import datetime
+from enum import Enum
+
+from pipelines.schedules import create_daily_cron
+from pipelines.utils.gcp.bigquery import SourceTable
+
+
+class constants(Enum):  # pylint: disable=c0103
+    """
+    Valores constantes para captura de dados do MoviDesk
+    """
+
+    MOVIDESK_SOURCE_NAME = "movidesk"
+    MOVIDESK_SECRET_PATH = "sppo_subsidio_recursos_api"  # movidesk_api
+    TICKETS_BASE_URL = "https://api.movidesk.com/public/v1/tickets"
+    TICKETS_TABLE_ID = "tickets"
+    TICKETS_SOURCE = SourceTable(
+        source_name=MOVIDESK_SOURCE_NAME,
+        table_id=TICKETS_TABLE_ID,
+        first_timestamp=datetime(2024, 12, 1, 0, 0, 0),
+        schedule_cron=create_daily_cron(hour=0),
+        partition_date_only=True,
+        # max_recaptures=5,
+        primary_keys=["protocol"],
+    )

pipelines/capture/movidesk/flows.py

@@ -0,0 +1,15 @@
+# -*- coding: utf-8 -*-
+"""Flows de captura dos tickets do MoviDesk"""
+from pipelines.capture.movidesk.constants import constants
+from pipelines.capture.movidesk.tasks import create_tickets_extractor
+from pipelines.capture.templates.flows import create_default_capture_flow
+from pipelines.constants import constants as smtr_constants
+from pipelines.utils.prefect import set_default_parameters
+
+CAPTURA_TICKETS = create_default_capture_flow(
+    flow_name="movidesk: tickets - captura",
+    source=constants.TICKETS_SOURCE.value,
+    create_extractor_task=create_tickets_extractor,
+    agent_label=smtr_constants.RJ_SMTR_AGENT_LABEL.value,
+)
+set_default_parameters(CAPTURA_TICKETS, {"recapture": True})

pipelines/capture/movidesk/tasks.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""Tasks de captura dos tickets do MoviDesk"""
+from datetime import datetime, timedelta
+from functools import partial
+
+from prefect import task
+
+from pipelines.capture.movidesk.constants import constants
+from pipelines.constants import constants as smtr_constants
+from pipelines.utils.extractors.api import get_raw_api_top_skip
+from pipelines.utils.gcp.bigquery import SourceTable
+from pipelines.utils.secret import get_secret
+
+
+@task(
+    max_retries=smtr_constants.MAX_RETRIES.value,
+    retry_delay=timedelta(seconds=smtr_constants.RETRY_DELAY.value),
+)
+def create_tickets_extractor(
+    source: SourceTable,  # pylint: disable=W0613
+    timestamp: datetime,
+):
+    """Cria a extração dos tickets do MoviDesk"""
+
+    start = datetime.strftime(timestamp - timedelta(days=1), "%Y-%m-%dT%H:%M:%S.%MZ")
+    end = datetime.strftime(timestamp, "%Y-%m-%dT%H:%M:%S.%MZ")
+    token = get_secret(constants.MOVIDESK_SECRET_PATH.value)["token"]
+    service = "Viagem Individual"
+    params = {
+        "token": token,
+        "$select": "id,protocol,createdDate,lastUpdate",
+        "$filter": f"serviceFirstLevel eq '{service} - Recurso Viagens Subsídio'and (lastUpdate ge {start} and lastUpdate lt {end} or createdDate ge {start} and createdDate lt {end})",  # noqa
+        "$expand": "customFieldValues,customFieldValues($expand=items)",
+        "$orderby": "createdDate asc",
+    }
+
+    return partial(
+        get_raw_api_top_skip,
+        url=constants.TICKETS_BASE_URL.value,
+        headers=None,
+        params=params,
+        top_param_name="$top",
+        skip_param_name="$skip",
+        page_size=1000,
+        max_page=10,
+    )

pipelines/flows.py

@@ -3,6 +3,7 @@
 Imports all flows for every project so we can register all of them.
 """
 from pipelines.capture.jae.flows import *  # noqa
+from pipelines.capture.movidesk.flows import *  # noqa
 from pipelines.capture.rioonibus.flows import *  # noqa
 from pipelines.capture.sonda.flows import *  # noqa
 from pipelines.exemplo import *  # noqa