Merge pull request #219 from DostEducation/develop

Release v1.0

.env.example

@@ -7,3 +7,5 @@ DB_NAME=
 DB_HOST=
 DB_PORT=
 SECRET_KEY=
+RETRY_LOGS_BATCH_LIMIT=
+MAX_RETRY_ATTEMPTS_FOR_LOGS=

api/mixins.py

@@ -1,10 +1,9 @@
 from __future__ import absolute_import
 
 from api import db
+from datetime import datetime
 
 
 class TimestampMixin(object):
-    created_on = db.Column(db.DateTime, server_default=db.func.now())
-    updated_on = db.Column(
-        db.DateTime, server_onupdate=db.func.now(), server_default=db.func.now()
-    )
+    created_on = db.Column(db.DateTime, default=datetime.now)
+    updated_on = db.Column(db.DateTime, onupdate=datetime.now, default=datetime.now)

api/models/registration.py

@@ -15,9 +15,6 @@ def get_by_phone(self, phone):
     def get_by_user_id(self, user_id):
         return self.filter(Registration.user_id == user_id).first()
 
-    def get_by_user_id(self, user_id):
-        return self.filter(Registration.user_id == user_id).first()
-
     def get_latest_by_phone(self, phone):
         return (
             self.filter(Registration.user_phone.contains(phone[-10:]))

api/models/user_program.py

@@ -6,6 +6,10 @@
 class UserProgramQuery(BaseQuery):
     def upsert_user_program(self, user_id, data):
         user_program_details = self.get_latest_active_user_program(user_id)
+
+        if not user_program_details:
+            user_program_details = self.get_latest_user_program(user_id)
+
         if not user_program_details:
             self.create(user_id, data)
         else:
@@ -51,6 +55,15 @@ def get_latest_active_user_program(self, user_id):
             .first()
         )
 
+    def get_latest_user_program(self, user_id):
+        return (
+            self.filter(
+                UserProgram.user_id == user_id,
+            )
+            .order_by(UserProgram.id.desc())
+            .first()
+        )
+
 
 class UserProgram(TimestampMixin, db.Model):
     query_class = UserProgramQuery

api/models/webhook_transaction_log.py

@@ -16,3 +16,4 @@ class WebhookTransactionLog(TimestampMixin, db.Model):
     id = db.Column(db.Integer, primary_key=True)
     payload = db.Column(db.Text)
     processed = db.Column(db.Boolean, nullable=False)
+    attempts = db.Column(db.Integer, nullable=False, default="0")

api/services/__init__.py

@@ -4,3 +4,4 @@
 from .content_service import *
 from .user_contact_service import *
 from .transaction_log_service import *
+from .user_program_service import *

api/services/call_log_service.py

@@ -55,7 +55,7 @@ def handle_call_log(self, jsonData):
                 else:
                     self.create_call_logs(jsonData)
             else:
-                print("flow_run_uuid is now available.")
+                print("flow_run_uuid is not available.")
         except:
             print("Failed to log the call details")
 
@@ -96,6 +96,9 @@ def create_call_logs(self, jsonData):
                 flow_category=jsonData["flow_category"]
                 if "flow_category" in jsonData
                 else self.flow_category,
+                created_on=jsonData["log_created_on"]
+                if jsonData.get("log_created_on", None)
+                else datetime.now(),
             )
             helpers.save(new_call_log)
             self.call_log = new_call_log

api/services/content_service.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from api import models, helpers
 
 
@@ -49,6 +50,9 @@ def add_user_module_content(self, jsonData):
                     program_module_id=program_module_details.id,
                     user_program_id=user_program_details.id,
                     status="complete",
+                    created_on=jsonData["log_created_on"]
+                    if jsonData.get("log_created_on", None)
+                    else datetime.now(),
                 )
                 helpers.save(user_module_content_data)
             return user_module_content_data.id

api/services/prompt_service.py

@@ -1,4 +1,5 @@
 from api import models, db, helpers, app
+from datetime import datetime
 
 
 class PromptService(object):
@@ -59,6 +60,8 @@ def handle_prompt_response(self, jsonData):
                     ivr_prompt_data["prompt_name"] = prompt_name
                     ivr_prompt_data["prompt_response"] = prompt_response
                     ivr_prompt_data["keypress"] = data[key]["value"]
+                    if jsonData.get("log_created_on", None):
+                        ivr_prompt_data["log_created_on"] = jsonData["log_created_on"]
                     self.add_prompt_response(ivr_prompt_details, ivr_prompt_data)
 
     def if_exists(self, ivr_prompt_response_details, prompt_name, prompt_response):
@@ -84,6 +87,9 @@ def add_prompt_response(self, ivr_prompt_details, data):
                 else None,
                 call_log_id=self.call_log_id,
                 keypress=keypress,
+                created_on=data["log_created_on"]
+                if data.get("log_created_on", None)
+                else datetime.now(),
             )
             helpers.save(ivr_prompt_response)
         except:

api/services/transaction_log_service.py

@@ -1,5 +1,5 @@
 # This file is treated as service layer
-from api import models, helpers
+from api import models, helpers, app
 import json
 
 
@@ -14,3 +14,19 @@ def create_new_webhook_log(self, jsonData):
     def mark_webhook_log_as_processed(self, webhook_log):
         webhook_log.processed = True
         helpers.save(webhook_log)
+
+    def get_failed_webhook_transaction_log(self):
+        failed_webhook_transaction_logs = (
+            models.WebhookTransactionLog.query.filter(
+                models.WebhookTransactionLog.processed == False
+            )
+            .filter(
+                models.WebhookTransactionLog.attempts
+                < app.config["MAX_RETRY_ATTEMPTS_FOR_LOGS"]
+            )
+            .order_by(models.WebhookTransactionLog.created_on)
+            .limit(app.config["RETRY_LOGS_BATCH_LIMIT"])
+            .all()
+        )
+
+        return failed_webhook_transaction_logs

api/services/user_program_service.py

@@ -0,0 +1,24 @@
+from api import models, helpers, db
+
+
+class UserProgramService(object):
+    def __init__(self):
+        self.user_id = None
+        self.user_phone = None
+        self.user_program_data = None
+
+    def set_init_data(self, jsonData):
+        user_phone = helpers.fetch_by_key("urn", jsonData["contact"])
+        self.user_phone = helpers.sanitize_phone_string(user_phone)
+        user = models.User.query.get_by_phone(self.user_phone)
+        self.user_id = user.id
+
+    def mark_user_program_as_completed(self, JsonData):
+        self.set_init_data(JsonData)
+        self.user_program_data = models.UserProgram.get_by_user_id(self.user_id)
+
+        if self.user_program_data:
+            self.user_program_data.status = (
+                models.UserProgram.UserProgramStatus.COMPLETE
+            )
+            db.session.commit()

config.py

@@ -38,3 +38,5 @@
 SECRET_KEY = os.environ.get("SECRET_KEY")
 
 DEFAULT_PROGRAM_ID = 2
+RETRY_LOGS_BATCH_LIMIT = os.environ.get("RETRY_LOGS_BATCH_LIMIT", 1000)
+MAX_RETRY_ATTEMPTS_FOR_LOGS = os.environ.get("MAX_RETRY_ATTEMPTS_FOR_LOGS", 3)

main.py

@@ -1,67 +1,111 @@
 from api import services
 from flask import jsonify
+import json
+
+from api.helpers import db_helper
 
 ### Endpoint for Cloud function
 def webhook(request):
     if request.method == "POST":
         try:
             jsonData = request.get_json()
-            transaction_log_service = services.TransactionLogService()
-            webhook_log = transaction_log_service.create_new_webhook_log(jsonData)
+        except:
+            return jsonify(message="Something went wrong!"), 400
+
+        transaction_log_service = services.TransactionLogService()
+
+        if jsonData and jsonData.get("type", None) != ("retry_failed_log"):
             if "contact" in jsonData:
-                # Conditions based on the flow categories
-                if (
-                    "flow_category" in jsonData
-                    and jsonData["flow_category"] != "dry_flow"
-                ):
-                    handle_flow_category_data(jsonData)
-
-                    # Handle call logs
-                    calllog_service = services.CallLogService()
-                    calllog_service.handle_call_log(jsonData)
-
-                # All the prompt responses are captured with results
-                if "results" in jsonData:
-                    handle_prompts(jsonData)
-
-                # Handle content details
-                user_module_content_id = None
-                if "content_id" in jsonData:
-                    content_service = services.ContentService()
-                    user_module_content_id = content_service.add_user_module_content(
-                        jsonData
-                    )
-                if user_module_content_id:
-                    calllog_service.update_user_module_content_id_in_call_log(
-                        user_module_content_id
-                    )
-
-                # Handle contact groups
-                if (
-                    "groups" in jsonData["contact"]
-                    and jsonData["contact"]["groups"] is not None
-                    and jsonData.get("flow_category", None) == "dry_flow"
-                ):
-                    handle_user_group_data(jsonData)
-
-                # Handle custom fields
-                if (
-                    "fields" in jsonData["contact"]
-                    and jsonData["contact"]["fields"] is not None
-                    and jsonData.get("flow_category", None) == "dry_flow"
-                ):
-                    handle_user_custom_field_data(jsonData)
+                webhook_log = transaction_log_service.create_new_webhook_log(jsonData)
+            processed = handle_payload(jsonData)
 
-                transaction_log_service.mark_webhook_log_as_processed(webhook_log)
-            else:
+            if processed is False:
+                return jsonify(message="Something went wrong!"), 400
+            elif processed == -1:
                 return jsonify(message="Contact"), 400
-        except:
-            return jsonify(message="Something went wrong!"), 400
+
+            if "contact" in jsonData:
+                transaction_log_service.mark_webhook_log_as_processed(webhook_log)
+        elif jsonData and jsonData.get("type", None) == "retry_failed_log":
+            retry_failed_webhook(transaction_log_service)
+
         return jsonify(message="Success"), 200
     else:
         return jsonify(message="Currently, the system do not accept a GET request"), 405
 
 
+def retry_failed_webhook(transaction_log_service):
+    failed_webhook_logs = transaction_log_service.get_failed_webhook_transaction_log()
+
+    for log in failed_webhook_logs:
+        log.attempts += 1
+        db_helper.save(log)
+
+        json_data = json.loads(log.payload)
+        json_data["log_created_on"] = log.created_on
+        processed = handle_payload(json_data, True)
+
+        if processed is not True:
+            continue
+
+        log.processed = True
+        db_helper.save(log)
+
+
+def handle_payload(jsonData, is_retry_payload=False):
+    try:
+        if "contact" in jsonData:
+            # Conditions based on the flow categories
+            if "flow_category" in jsonData and jsonData["flow_category"] != "dry_flow":
+                handle_flow_category_data(jsonData)
+
+                # Handle call logs
+                calllog_service = services.CallLogService()
+                calllog_service.handle_call_log(jsonData)
+
+            if jsonData.get("is_last_content", None) is True:
+                update_user_program(jsonData)
+
+            # All the prompt responses are captured with results
+            if "results" in jsonData:
+                handle_prompts(jsonData)
+
+            # Handle content details
+            user_module_content_id = None
+            if "content_id" in jsonData:
+                content_service = services.ContentService()
+                user_module_content_id = content_service.add_user_module_content(
+                    jsonData
+                )
+            if user_module_content_id:
+                calllog_service.update_user_module_content_id_in_call_log(
+                    user_module_content_id
+                )
+
+            # Handle contact groups
+            if (
+                "groups" in jsonData["contact"]
+                and jsonData["contact"]["groups"] is not None
+                and jsonData.get("flow_category", None) == "dry_flow"
+                and not is_retry_payload
+            ):
+                handle_user_group_data(jsonData)
+
+            # Handle custom fields
+            if (
+                "fields" in jsonData["contact"]
+                and jsonData["contact"]["fields"] is not None
+                and jsonData.get("flow_category", None) == "dry_flow"
+                and not is_retry_payload
+            ):
+                handle_user_custom_field_data(jsonData)
+        else:
+            return -1
+    except:
+        return False
+    return True
+
+
 def handle_flow_category_data(jsonData):
     registration_service = services.RegistrationService()
     if jsonData["flow_category"] == "registration":
@@ -81,3 +125,8 @@ def handle_user_custom_field_data(jsonData):
 def handle_prompts(jsonData):
     prompt_service = services.PromptService()
     prompt_service.handle_prompt_response(jsonData)
+
+
+def update_user_program(JsonData):
+    user_program_service = services.UserProgramService()
+    user_program_service.mark_user_program_as_completed(JsonData)