modernize ticket and accounting

iteration 0
2025-11-24 21:31:43 +01:00
parent c33ab89354
commit b1a70673a4
97 changed files with 13619 additions and 2398 deletions
@@ -0,0 +1,168 @@
+id: syn-extraction
+namespace: dev
+description: Run Avicenna extraction with synapse
+
+labels:
+  env: dev
+  project: avicenna
+
+# triggers:
+#   - id: schedule
+#     type: io.kestra.plugin.core.trigger.Schedule
+#     cron: "*/5 * * * *"
+
+tasks:
+  - id: synapseConfig
+    type: io.kestra.plugin.core.storage.Write
+    content: |
+      pipelines:
+        - name: "patient"
+          source:
+            type: "rest_api"
+            config:
+              base_url: "https://avicenna.dispolive.de"
+              path: "/data/patientenStammdaten"
+              method: POST
+              records_path: "data"
+              auth:
+                strategy: "dispolive"
+                dispolive:
+                  username: Marcel
+                  password: arndt1.!
+              rate_limit:
+                requests_per_second: 5
+                burst: 1
+          destination:
+            type: "csv"
+            config: 
+              path: "./patient.csv"
+          transform:
+            - strategy: keep_fields
+              config:
+                fields: 
+                - "_id"
+                - "name"
+                - "surname"
+                - "street"
+                - "zip"
+                - "city"
+                - "birthday"
+                - "kkId"
+                - "krankenkasse"
+            - strategy: rename_fields
+              config:
+                mapping:
+                  _id: id
+            - strategy: add_fields
+              config:
+                fields: |
+                  {
+                    "jobId": "{{ execution.id }}"
+                  }
+        - name: "attendanceRegistration"
+          sync:
+            strategy: "time_range_iteration"
+            time_range_iteration:
+              start_date: "2025-02-01"
+              end_date: "2025-02-28"
+              date_format: "2006-01-02"
+              interval: 1d
+          source:
+            type: "rest_api"
+            config:
+              base_url: "https://app.planning.nu"
+              {% raw %}path: "/avicennaambulance/api2/attendance-registration/{{.SliceStart}}"{% endraw %}
+              records_path: "attendanceRegistration"
+              auth:
+                strategy: "header"
+                header:
+                  header: "authorization"
+                  token: "Token ${SYN_DYFLEXIS_V2_TOKEN}"
+              rate_limit:
+                requests_per_second: 5
+                burst: 1
+          destination:
+            type: "csv"
+            config: 
+              path: "./_output/02-2025-attendance-registrations.csv"
+  - id: run
+    type: io.kestra.plugin.docker.Run
+    containerImage: thedevilisdero/synapse:latest
+    pullPolicy: ALWAYS
+    inputFiles:
+      synapse.yaml: "{{ outputs.synapseConfig.uri }}"
+    commands:
+      - --config
+      - synapse.yaml
+      - run
+      - patient
+    outputFiles:
+      - patient.csv
+  - id: log
+    type: io.kestra.plugin.core.log.Log
+    message: "{{ read(outputs.run.outputFiles['patient.csv']) }}"
+  - id: deleteSource
+    type: io.kestra.plugin.jdbc.postgresql.Query
+    url: jdbc:postgresql://avicenna_db:5432/avicenna
+    username: johndoe
+    password: randompassword
+    sql: DELETE FROM "PatientSource";
+  - id: copyin
+    type: io.kestra.plugin.jdbc.postgresql.CopyIn
+    url: jdbc:postgresql://avicenna_db:5432/avicenna
+    username: johndoe
+    password: randompassword
+    format: CSV
+    from: "{{ outputs.run.outputFiles['patient.csv'] }}"
+    table: "\"PatientSource\""
+    header: true
+    columns:
+      - birthday
+      - city
+      - id
+      - "\"jobId\""
+      - "\"kkId\""
+      - krankenkasse
+      - name
+      - street
+      - surname
+      - zip
+  # - id: loadStatic
+  #   type: io.kestra.plugin.core.flow.EachSequential
+  #   value: '{{ outputs.getPartitions.vars.partitions }}'
+  # - id: getPartitions
+  #   type: io.kestra.plugin.scripts.python.Script
+  #   taskRunner:
+  #     type: io.kestra.plugin.scripts.runner.docker.Docker
+  #   containerImage: ghcr.io/kestra-io/pydata:latest
+  #   beforeCommands:
+  #     - pip install -U arrow
+  #   script: |
+  #     from kestra import Kestra
+  #     import arrow
+  #     partitions = []
+
+  #     now = arrow.utcnow() #.shift(months=-1)
+  #     partitions.append("from={}&to={}".format(now.span('month')[0].format('YYYY-MM-DD'), now.shift(days=14).format('YYYY-MM-DD')))
+  #     lastMonth = now.shift(months=-1)
+  #     partitions.append("from={}&to={}".format(lastMonth.span('month')[0].format('YYYY-MM-DD'), lastMonth.span('month')[-1].format('YYYY-MM-DD')))
+  #     monthBeforeLast = lastMonth.shift(months=-1)
+  #     partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+  #     monthBeforeLast = monthBeforeLast.shift(months=-1)
+  #     partitions.append("from={}&to={}".format(monthBeforeLast.span('month')[0].format('YYYY-MM-DD'), monthBeforeLast.span('month')[-1].format('YYYY-MM-DD')))
+  #     Kestra.outputs({'partitions': partitions})
+  # - id: loadThreeMonths
+  #   type: io.kestra.plugin.core.flow.EachSequential
+  #   value: '{{ outputs.getPartitions.vars.partitions }}'
+  #   tasks:
+  #     - id: loadTours
+  #       type: "io.kestra.plugin.docker.Run"
+  #       containerImage: thedevilisdero/synapse:latest
+  #     - id: "request"
+  #       type: "io.kestra.plugin.core.http.Request"
+  #       uri: "http://avicenna_data:3000/etl?{{ taskrun.value }}"
+  #       headers: 
+  #         user-agent: "kestra-io"
+  #       options:
+  #         readTimeout: "PT180S"
+  #       method: "GET"
@@ -0,0 +1,99 @@
+id: main_sync_workflow
+namespace: ceo.genius.etl
+
+tasks:
+  # Schritt 1: Definiere alle Entitäten und berechne die Zeiträume für die Synchronisation
+  - id: setup_parameters
+    type: io.kestra.plugin.scripts.python.Script
+    containerImage: python:3.11-slim
+    beforeCommands:
+      - "pip install python-dateutil"
+      - "pip install kestra"
+    script: |
+      from kestra import Kestra
+      import json
+      from datetime import date
+      from dateutil.relativedelta import relativedelta
+
+      today = date.today()
+      
+      # 1. Definiere die Zeiträume: aktueller Monat + die letzten 3 vollen Monate
+      date_ranges = []
+      # Aktueller Monat (vom 1. bis heute)
+      date_ranges.append({
+          "start": today.replace(day=1).strftime("%Y-%m-%d"),
+          "end": today.strftime("%Y-%m-%d")
+      })
+      # Letzte 3 volle Monate
+      for i in range(1, 4):
+          month = today - relativedelta(months=i)
+          start_of_month = month.replace(day=1)
+          end_of_month = (start_of_month + relativedelta(months=1)) - relativedelta(days=1)
+          date_ranges.append({
+              "start": start_of_month.strftime("%Y-%m-%d"),
+              "end": end_of_month.strftime("%Y-%m-%d")
+          })
+
+      # 2. Definiere die Entitäten mit ihrem Schema.
+      # HIER IST DER ZENTRALE ORT, UM NEUE ENTITÄTEN HINZUZUFÜGEN ODER ZU ÄNDERN.
+      config = {
+          "static_entities": [
+              {
+                  "name": "patient",
+                  "db_table": "PatientSource",
+                  "columns": ["birthday", "city", "id", "\"jobId\"", "\"kkId\"", "krankenkasse", "name", "street", "surname", "zip"]
+              }
+              # FÜGE HIER WEITERE STATISCHE ENTITÄTEN HINZU
+              # {
+              #   "name": "neue_entitaet",
+              #   "db_table": "NeueEntitaetSource",
+              #   "columns": ["spalte1", "spalte2"]
+              # }
+          ],
+          "dynamic_entities": [
+              {
+                  "name": "attendanceRegistration",
+                  "db_table": "AttendanceRegistrationSource",
+                  "columns": ["\"attendanceRegistrationId\"", "\"dateTime\"", "\"employeeId\"", "event", "\"jobId\"", "\"personnelNumber\""]
+              }
+              # FÜGE HIER WEITERE DYNAMISCHE ENTITÄTEN HINZU
+          ],
+          "date_ranges": date_ranges
+      }
+      Kestra.outputs({'entity_config': config})
+      print(json.dumps(config))
+
+  - id: sync_static_entities
+    type: io.kestra.plugin.core.flow.EachParallel
+    value: "{{ outputs.setup_parameters.vars.entity_config.static_entities }}"
+    tasks:
+      - id: run_static_subflow
+        type: io.kestra.plugin.core.flow.Subflow
+        namespace: ceo.genius.etl
+        flowId: subflow_static_entity
+        inputs:
+          entity_name: "{{ taskrun.value.name }}"
+          db_table: "{{ taskrun.value.db_table }}"
+          columns: "{{ taskrun.value.columns }}"
+
+  - id: sync_dynamic_entities
+    type: io.kestra.plugin.core.flow.EachParallel
+    value: |
+      {%- set result = [] -%}
+      {%- for entity in outputs.setup_parameters.output.dynamic_entities -%}
+      {%-   for range in outputs.setup_parameters.output.date_ranges -%}
+      {%-     do result.append({ "entity": entity, "range": range }) -%}
+      {%-   endfor -%}
+      {%- endfor -%}
+      {{ result | json }}
+    tasks:
+      - id: run_dynamic_subflow
+        type: io.kestra.plugin.core.flow.Subflow
+        namespace: ceo.genius.etl # Passe dies an, falls die Subflows in einem anderen Namespace liegen
+        flowId: subflow_dynamic_entity
+        inputs:
+          entity_name: "{{ taskrun.value.entity.name }}"
+          db_table: "{{ taskrun.value.entity.db_table }}"
+          columns: "{{ taskrun.value.entity.columns }}"
+          startDate: "{{ taskrun.value.range.start }}"
+          endDate: "{{ taskrun.value.range.end }}"
@@ -0,0 +1,51 @@
+id: subflow_static_entity
+namespace: ceo.genius.etl
+
+inputs:
+  - id: entity_name
+    type: STRING
+  - id: db_table
+    type: STRING
+  - id: columns
+    type: JSON
+
+tasks:
+#  - id: render_synapse_config
+#    type: io.kestra.core.tasks.storages.template
+#    templateUri: "kestra://{{ flow.namespace }}/_templates/{{ inputs.entity_name }}.yml"
+
+  - id: render_synapse_config
+    type: io.kestra.plugin.core.templating.TemplatedTask
+    spec: |
+      {{ read('nsfile://' ~ flow.namespace ~ '/configs/' ~ inputs.entity_name ~ '.yml') }}
+
+  - id: run_synapse
+    type: io.kestra.plugin.docker.Run
+    containerImage: thedevilisdero/synapse:latest
+    pullPolicy: ALWAYS
+    inputFiles:
+      synapse.yaml: "{{ read('nsfile://' ~ flow.namespace ~ '/configs/' ~ inputs.entity_name ~ '.yml') }}"
+    commands:
+      - --config
+      - synapse.yaml
+      - run
+      - "{{ inputs.entity_name }}"
+    outputFiles:
+      - "{{ inputs.entity_name }}.csv"
+
+  - id: delete_from_table
+    type: io.kestra.plugin.jdbc.postgresql.Query
+    url: jdbc:postgresql://avicenna_db:5432/avicenna # ANPASSEN
+    username: johndoe # ANPASSEN
+    password: randompassword # ANPASSEN
+    sql: 'DELETE FROM "{{ inputs.db_table }}";'
+
+  - id: copy_to_table
+    type: io.kestra.plugin.jdbc.postgresql.CopyIn
+    url: jdbc:postgresql://avicenna_db:5432/avicenna # ANPASSEN
+    username: johndoe # ANPASSEN
+    password: randompassword # ANPASSEN
+    from: "{{ outputs.run_synapse.outputFiles[inputs.entity_name ~ '.csv'] }}"
+    table: '"{{ inputs.db_table }}"'
+    header: true
+    columns: "{{ inputs.columns }}"
@@ -58,8 +58,7 @@ async function main() {
  const schichtplanung = await getSchichtplanungForDay(new Date('2025-08-11')).then(response => response.json());
  const chunks = chunk(schichtplanung.map(({id}) => id), 1)
  for(const chunk of chunks) {
-    const res = await deleteSchichtplanungItems(chunk as string[])
-    console.log(res)
+    await deleteSchichtplanungItems(chunk as string[])
    await sleep(200)
  }
 }