From 1f9dfb99a83baf7bad425ed2212f9437368add3d Mon Sep 17 00:00:00 2001
From: Adam Kariv <adam.kariv@gmail.com>
Date: Mon, 15 Jul 2024 19:50:28 +0300
Subject: [PATCH] Fix for supports

---
 .../pipelines/supports/collate_payments.py    | 34 +++++++
 .../supports/fix_values_by_request_year.py    | 16 ----
 .../pipelines/supports/pipeline-spec.yaml     | 88 +++++++------------
 3 files changed, 68 insertions(+), 70 deletions(-)
 create mode 100644 datapackage_pipelines_budgetkey/pipelines/supports/collate_payments.py

diff --git a/datapackage_pipelines_budgetkey/pipelines/supports/collate_payments.py b/datapackage_pipelines_budgetkey/pipelines/supports/collate_payments.py
new file mode 100644
index 00000000..ffa623e5
--- /dev/null
+++ b/datapackage_pipelines_budgetkey/pipelines/supports/collate_payments.py
@@ -0,0 +1,34 @@
+import decimal
+
+from datapackage_pipelines.wrapper import process
+
+def fix_decimal(value):
+    if isinstance(value, decimal.Decimal):
+        value = float(value)
+    return value
+
+def process_row(row, row_index,
+                spec, resource_index,
+                parameters, stats):
+    fields = parameters['fields']
+    field = parameters['field']
+    if spec['name'] == parameters['resource']:
+        rec = dict(
+            (k, fix_decimal(row[k]))
+            for k in fields
+        )
+        row[field] = rec
+    return row
+
+
+def modify_datapackage(dp, parameters, stats):
+    for resource in dp['resources']:
+        if resource['name'] == parameters['resource']:
+            resource['schema']['fields'].append({
+                'name': parameters['field'],
+                'type': 'object',
+            })
+    return dp
+
+process(modify_datapackage=modify_datapackage,
+        process_row=process_row)
\ No newline at end of file
diff --git a/datapackage_pipelines_budgetkey/pipelines/supports/fix_values_by_request_year.py b/datapackage_pipelines_budgetkey/pipelines/supports/fix_values_by_request_year.py
index 33db131a..bac6c221 100644
--- a/datapackage_pipelines_budgetkey/pipelines/supports/fix_values_by_request_year.py
+++ b/datapackage_pipelines_budgetkey/pipelines/supports/fix_values_by_request_year.py
@@ -5,27 +5,11 @@
 
 
 def process_row(row, *_):
-    amount_total = 0
-    payments = []
-    for deet in row['payments']:
-        new_deet = {}
-        amount_total += deet['amount_total'] if deet['amount_total'] is not None else 0
-        for k, v in deet.items():
-            if isinstance(v, Decimal):
-                v = float(v)
-            new_deet[k] = v
-        payments.append(new_deet)
-    row['payments'] = payments
-    row['amount_total'] = amount_total
     row['short_id'] = (row['entity_id'] or slugify(row['recipient'], max_length=32)).strip()
     return row
 
 
 def modify_datapackage(dp, *_):
-    dp['resources'][0]['schema']['fields'].append({
-        'name': 'amount_total',
-        'type': 'number'
-    })
     dp['resources'][0]['schema']['fields'].append({
         'name': 'short_id',
         'type': 'string',
diff --git a/datapackage_pipelines_budgetkey/pipelines/supports/pipeline-spec.yaml b/datapackage_pipelines_budgetkey/pipelines/supports/pipeline-spec.yaml
index a443a417..043b54ad 100644
--- a/datapackage_pipelines_budgetkey/pipelines/supports/pipeline-spec.yaml
+++ b/datapackage_pipelines_budgetkey/pipelines/supports/pipeline-spec.yaml
@@ -406,6 +406,21 @@ by-request-year:
           support_title:
             es:title: true
 
+    - run: collate_payments
+      parameters:
+        resource: supports
+        field: payments
+        fields: 
+          - resolved_budget_codes
+          - support_title
+          - amount_paid
+          - amount_approved
+          - recipient_entity_id
+          - year_paid
+          - supporting_ministry
+          - amount_advance
+          - amount_total
+
     - run: join
       parameters:
         source:
@@ -414,7 +429,8 @@ by-request-year:
             - budget_code
             - year_requested
             - recipient
-            - request_type
+            - entity_id
+            - support_title
           delete: true
         target:
           name: supports
@@ -423,76 +439,40 @@ by-request-year:
           budget_code: null
           year_requested: null
           recipient: null
+          support_title: null
           request_type: null
           entity_name: null
           entity_id: null
           entity_kind: null
-          supporting_ministry: null
-          year_paid: null
-          support_title: null
+          supporting_ministry:
+            aggregate: last          
           recipient_entity_id: null
-          amount_approved: null
-          amount_paid: null
-          amount_advance: null
-          amount_total: null
+          amount_approved:
+            aggregate: max
+          amount_paid:
+            aggregate: sum
+          amount_total:
+            aggregate: sum
           resolved_budget_codes: null
           last_payment_year:
             name: year_paid
-            aggregate: last
+            aggregate: max
           last_payment_amount:
             name: amount_paid
             aggregate: last
-
-    - run: collate
-      parameters:
-        resource: supports
-        key:
-          - budget_code
-          - year_requested
-          - recipient
-          - request_type
-          - entity_name
-          - entity_id
-          - entity_kind
-          - last_payment_year
-          - last_payment_amount
-        collated-field-name: payments
-
-    - run: join
-      parameters:
-        source:
-          name: supports
-          key:
-            - budget_code
-            - year_requested
-            - recipient
-            - request_type
-            - entity_name
-            - entity_id
-            - entity_kind
-            - last_payment_year
-            - last_payment_amount
-          delete: true
-        target:
-          name: supports
-          key: null
-        fields:
-          budget_code: null
-          year_requested: null
-          recipient: null
-          request_type: null
-          entity_name: null
-          entity_id: null
-          entity_kind: null
-          last_payment_year: null
-          last_payment_amount: null
           payments:
             aggregate: array
 
     # - run: match_criteria
     - run: fix_values_by_request_year
-    - run: collate_schema
     - run: calc-support-score
+    - run: set_types
+      parameters:
+        payments:
+          type: array
+          es:itemType: object
+          es:index: false
+            
     - run: set_types
     - run: set_primary_key
       parameters: