Composite score block

README.md

@@ -1 +1 @@
-**Hello world!!!**
+# Block for assigning a composite score

block.py

@@ -1,21 +1,33 @@
-@flowx_block
+import logging
-def example_function(request: dict) -> dict:
+from score_processing import processing
 
-  # Processing logic here...
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
 
-  return {
+def __main__(
-    "meta_info": [
+        hd_score_m1: float,
-      {
+        hd_score_g1: float,
-        "name": "created_date",
+        hd_score_s1: float,
-        "type": "string",
+        hd_score_s2: float,
-        "value": "2024-11-05"
+        hd_score_s3: float,
-      }
+    ) -> dict:
-    ],
+
-    "fields": [
+    data = {
-      {
+        "hd_score_m1": hd_score_m1,
-        "name": "",
+        "hd_score_g1": hd_score_g1,
-        "type": "",
+        "hd_score_s1": hd_score_s1,
-        "value": ""
+        "hd_score_s2": hd_score_s2,
-      }
+        "hd_score_s3": hd_score_s3,
-    ]
     }
+
+    final = processing(data)
+    logger.info(f"Scores of application: {final}")
+
+    return final
+
+# testing :
+# __main__

request_schema.json

@@ -1 +1,27 @@
-{}
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "hd_score_m1": {
+      "type": ["number", "null"],
+      "description": "HD Fraud Score M1"
+    },
+    "hd_score_g1": {
+      "type": ["number", "null"],
+      "description": "HD Fraud Score G1"
+    },
+    "hd_score_s1": {
+      "type": ["number", "null"],
+      "description": "HD Fraud Score S1"
+    },
+    "hd_score_s2": {
+      "type": ["number", "null"],
+      "description": "HD Fraud Score S2"
+    },
+    "hd_score_s3": {
+      "type": ["number", "null"],
+      "description": "HD Fraud Score S3"
+    }
+  },
+  "required": []
+}

requirements.txt

@@ -1 +1 @@
-{}
+

response_schema.json

@@ -1 +1,25 @@
-{}
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "hd_score": {
+      "type": ["number", "null"],
+      "description": "HD fraud Score"
+    },
+    "recommended_action": {
+      "type": ["string", "null"],
+      "description": "Recommended Action"
+    },
+    "description": {
+      "type": ["string", "null"],
+      "description": "Description of Recommended Action"
+    },
+    "hd_action_reasoncode": {
+      "type": ["string", "null"],
+      "description": "HD Action Reasoncode"
+    }
+  }
+}
+
+
+

score_processing.py

@@ -0,0 +1,75 @@
+import logging
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def processing(data: dict) -> dict:
+    score_threshold = 1200
+
+    # Compute FINAL_score based on conditions
+    try:
+        for key in ["hd_score_s1", "hd_score_s2", "hd_score_g1", "hd_score_s3"]:
+            if key in data and data[key] > score_threshold:
+                hd_score = data[key]
+                break
+        else:
+            hd_score = data.get("hd_score_m1", None)  # Default if no conditions are met
+
+        logging.info(f"hd_score calculated: {hd_score}")
+    except Exception as e:
+        logging.error(f"Error processing hd_score calculations: {e}")
+        return {}
+
+
+    # Compute recommended_action
+    try:
+        recommended_action = "Decline Application" if hd_score > score_threshold else "Pass Application"
+        logging.info(f"recommended_action: {recommended_action}")
+    except Exception as e:
+        logging.error(f"Error assigning recommended_action: {e}")
+        return {}
+
+    # Compute description
+    try:
+        description = (
+            f"HD Fraud Score is above the risk threshold {score_threshold}, Recommended action: Decline Application"
+            if hd_score > score_threshold
+            else f"HD Fraud Score is below the risk threshold {score_threshold}, Recommended action: Pass Application"
+        )
+        logging.info(f"description: {description}")
+    except Exception as e:
+        logging.error(f"Error assigning description: {e}")
+        return {}
+
+    # Compute action_reasoncode
+    try:
+        action_reason_mapping = {
+            "hd_score_s1": "S1 Triggered",
+            "hd_score_s2": "S2 Triggered",
+            "hd_score_g1": "G1 Triggered",
+            "hd_score_s3": "S3 Triggered",
+            "hd_score_m1": "M1 Triggered",
+        }
+
+        hd_action_reasoncode = next(
+            (reason for key, reason in action_reason_mapping.items() if data[key] > score_threshold),
+            "Pass"
+        )
+        logging.info(f"action_reasoncode: {hd_action_reasoncode}")
+    except Exception as e:
+        logging.error(f"Error compiling action_reasoncode: {e}")
+        return {}
+
+    return {
+        "hd_score": hd_score,
+        "recommended_action": recommended_action,
+        "description": description,
+        "hd_action_reasoncode": hd_action_reasoncode,
+    }
+
+

test_block.py

@@ -0,0 +1,17 @@
+import unittest
+from block import __main__
+
+data = {'hd_score_m1': 1173.0, 'hd_score_g1': 1203.0, 'hd_score_s1': 1240.0, 'hd_score_s2': 0, 'hd_score_s3': 0}
+
+class TestBlock(unittest.TestCase):
+    def test_main_success(self):
+        # blockResult = main(data)
+        blockResult = __main__(**data) 
+
+        # breakpoint()
+        self.assertIsInstance(blockResult, dict, "Result should be a dictionary.")
+        self.assertIn("hd_score", blockResult, "Result dictionary should contain 'hd_score' if success.")
+
+
+if __name__ == "__main__":
+    unittest.main()