Merge pull request #23 from DostEducation/feature/17-create-user-acti…

…vities-service

Functionality to capture user activities

.pre-commit-config.yaml

@@ -30,3 +30,7 @@ repos:
   rev: v1.7.5
   hooks:
     - id: docformatter
+-   repo: https://github.com/pre-commit/pygrep-hooks
+    rev: v1.10.0
+    hooks:
+    -   id: python-use-type-annotations

api/helpers/__init__.py

@@ -0,0 +1 @@
+from .common_helper import *

api/helpers/common_helper.py

@@ -0,0 +1,9 @@
+from datetime import datetime, timedelta
+
+
+def get_ist_timestamp() -> datetime:
+    return datetime.utcnow() + timedelta(minutes=330)
+
+
+def check_activity_key(activity_key: str, keyword: str, status: str):
+    return activity_key.startswith(keyword) and activity_key.endswith(status)

api/models/user_activities.py

@@ -1,18 +1,49 @@
+from datetime import date
+from flask_sqlalchemy.query import Query as BaseQuery
+from sqlalchemy import desc, func
+
 from api import db
 from api.mixins import TimestampMixin
 
 
+class UserActivitiesQuery(BaseQuery):
+    def get_todays_started_activity_for_user(self, user_id, user_phone):
+        return (
+            self.filter(
+                UserActivities.user_id == user_id,
+                UserActivities.user_phone == user_phone,
+                UserActivities.is_started.is_(True),
+                func.DATE(UserActivities.started_on) == date.today(),
+            )
+            .order_by(desc("started_on"))
+            .first()
+        )
+
+    def get_todays_succeeded_activity_for_user(self, user_id, user_phone):
+        return (
+            self.filter(
+                UserActivities.user_id == user_id,
+                UserActivities.user_phone == user_phone,
+                UserActivities.is_succeeded.is_(True),
+                func.DATE(UserActivities.succeeded_on) == date.today(),
+            )
+            .order_by(desc("succeeded_on"))
+            .first()
+        )
+
+
 class UserActivities(TimestampMixin, db.Model):
+    query_class = UserActivitiesQuery
 
     __tablename__ = "user_activities"
     id = db.Column(db.Integer, primary_key=True)
     user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
     user_phone = db.Column(db.BigInteger, nullable=False, index=True)
     user_flow_id = db.Column(db.Integer, db.ForeignKey("user_flows.id"))
     activity = db.Column(db.String(500))
-    is_started = db.Column(db.Boolean, default=False, nullable=False)
+    is_started = db.Column(db.Boolean)
     started_on = db.Column(db.DateTime)
-    is_succeeded = db.Column(db.Boolean, default=False, nullable=False)
+    is_succeeded = db.Column(db.Boolean)
     succeeded_on = db.Column(db.DateTime)
-    is_completed = db.Column(db.Boolean, default=False, nullable=False)
+    is_completed = db.Column(db.Boolean)
     completed_on = db.Column(db.DateTime)

api/models/user_flows.py

@@ -1,25 +1,29 @@
 from datetime import date
 from flask_sqlalchemy.query import Query as BaseQuery
-from sqlalchemy import func
+from sqlalchemy import desc, func
 
 from api import db
 from api.mixins import TimestampMixin
 
 
 class UserFlowsQuery(BaseQuery):
-    def get_by_flow_uuid_and_phone(self, flow_uuid, user_phone):
-        return self.filter(
-            UserFlows.flow_uuid == flow_uuid,
-            UserFlows.user_phone == user_phone,
-            func.DATE(UserFlows.flow_start_time) == date.today(),
-        ).first()
+    def get_todays_latest_user_flow(self, flow_uuid, user_phone):
+        return (
+            self.filter(
+                UserFlows.flow_uuid == flow_uuid,
+                UserFlows.user_phone == user_phone,
+                func.DATE(UserFlows.flow_start_time) == date.today(),
+            )
+            .order_by(desc("flow_start_time"))
+            .first()
+        )
 
 
 class UserFlows(TimestampMixin, db.Model):
     query_class = UserFlowsQuery
 
     class FlowRunStatus:
-        IN_PROGRESS = "in-progress"
+        STARTED = "started"
         COMPLETED = "completed"
         ENDED = "ended"
 
@@ -33,4 +37,4 @@ class FlowRunStatus:
     flow_run_status = db.Column(db.String(255))
     flow_start_time = db.Column(db.DateTime)
     flow_end_time = db.Column(db.DateTime)
-    is_active = db.Column(db.Boolean, default=False, nullable=False)
+    is_active = db.Column(db.Boolean)

api/services/__init__.py

@@ -1,4 +1,5 @@
 from .flow_run_log_service import *
+from .user_activities_service import *
 from .user_creation_service import *
 from .user_indicator_response_service import *
 from .webhook_transaction_log_service import *

api/services/flow_run_log_service.py

@@ -1,62 +1,78 @@
-from datetime import datetime, timedelta
+from typing import Any, Optional
 
 from api import models
+from api.helpers import common_helper
 from api.utils import db_utils
 from api.utils.loggingutils import logger
 
 
 class FlowRunLogService:
-    def __init__(self, user):
+    def __init__(self, user: models.Users):
         self.user = user
         self.class_model = models.UserFlows
 
-    def create_user_flow_log(self, json_data):
+    def create_user_flow_log(self, json_data: dict[str, Any]) -> models.UserFlows:
+        user_flow_log = None
+
         try:
             flow_uuid = json_data.get("flow_uuid")
             flow_name = json_data.get("flow_name")
             flow_type = json_data.get("flow_type")
-            flow_completed = json_data.get("flow_completed")
+            flow_status = json_data.get("flow_status")
 
-            today_flow_log = self.class_model.query.get_by_flow_uuid_and_phone(
-                flow_uuid, self.user.phone
+            latest_flow_log: models.UserFlows = (
+                self.class_model.query.get_todays_latest_user_flow(
+                    flow_uuid, self.user.phone
+                )
             )
 
-            user_flow_log = None
-
-            if not today_flow_log:
+            if flow_status == self.class_model.FlowRunStatus.STARTED:
                 user_flow_log = self.create_log(flow_uuid, flow_name, flow_type)
-            elif today_flow_log and flow_completed:
-                user_flow_log = self.update_log(today_flow_log)
+            elif flow_status == self.class_model.FlowRunStatus.COMPLETED:
+                user_flow_log = self.update_log(latest_flow_log)
+            else:
+                logger.error(
+                    f"Got unexpected flow status {flow_status}. Flow name {flow_name}."
+                )
 
-            return user_flow_log
         except Exception as e:
             logger.error(
                 f"Error while creating new user flow log. json data: {json_data}."
                 f"Error message: {e}"
             )
             raise
 
-    def create_log(self, flow_uuid, flow_name, flow_type):
-        user_flow_log = self.class_model(
+        if user_flow_log is None:
+            raise ValueError("Failed to create or update user flow log.")
+
+        return user_flow_log
+
+    def create_log(
+        self,
+        flow_uuid: Optional[str],
+        flow_name: Optional[str],
+        flow_type: Optional[str],
+    ) -> models.UserFlows:
+        user_flow_log: models.UserFlows = self.class_model(
             user_id=self.user.id,
             user_phone=self.user.phone,
             flow_uuid=flow_uuid,
             flow_name=flow_name,
             flow_type=flow_type,
-            flow_run_status=self.class_model.FlowRunStatus.IN_PROGRESS,
-            flow_start_time=datetime.utcnow() + timedelta(minutes=330),
+            flow_run_status=self.class_model.FlowRunStatus.STARTED,
+            flow_start_time=common_helper.get_ist_timestamp(),
             is_active=True,
         )
 
         db_utils.save(user_flow_log)
         logger.info(f"Created a user flow log for phone number {self.user.phone}.")
         return user_flow_log
 
-    def update_log(self, today_flow_log):
-        today_flow_log.flow_run_status = self.class_model.FlowRunStatus.COMPLETED
-        today_flow_log.flow_end_time = datetime.utcnow() + timedelta(minutes=330)
-        today_flow_log.is_active = False
+    def update_log(self, latest_flow_log: models.UserFlows) -> models.UserFlows:
+        latest_flow_log.flow_run_status = self.class_model.FlowRunStatus.COMPLETED
+        latest_flow_log.flow_end_time = common_helper.get_ist_timestamp()
+        latest_flow_log.is_active = False
 
-        db_utils.save(today_flow_log)
+        db_utils.save(latest_flow_log)
         logger.info(f"Updated user flow log for phone number {self.user.phone}.")
-        return today_flow_log
+        return latest_flow_log

api/services/user_activities_service.py

@@ -0,0 +1,99 @@
+from datetime import datetime
+from typing import Any, Optional
+
+from api import db, models
+from api.helpers import common_helper
+from api.utils.loggingutils import logger
+
+
+class UserActivitiesService:
+    def __init__(self, user, user_flow):
+        self.user_id = user.id
+        self.user_phone = user.phone
+        self.user_flow_id = user_flow.id
+        self.class_model = models.UserActivities
+
+    def handle_user_activities(self, json_data: dict[str, Any]):
+        try:
+            current_ist_time = common_helper.get_ist_timestamp()
+            contact_activities = json_data.get("contact", {}).get("fields", {})
+
+            for activity_key, _ in contact_activities.items():
+                if activity_key.strip() in contact_activities:
+                    user_activity = self.create_or_update_user_activity(
+                        activity_key, current_ist_time
+                    )
+                    if user_activity:
+                        db.session.add(user_activity)
+
+            db.session.commit()
+            logger.info(f"Captured user activity for {self.user_phone}.")
+        except Exception as e:
+            logger.error(
+                f"Error while capturing user activity for {self.user_phone}."
+                f"Error message: {e}"
+            )
+
+    def create_or_update_user_activity(
+        self, activity_key: str, current_ist_time: datetime
+    ) -> Optional[models.UserActivities]:
+        is_started = common_helper.check_activity_key(
+            activity_key, "activity_", "_started"
+        )
+        is_succeeded = common_helper.check_activity_key(
+            activity_key, "activity_", "_success"
+        )
+        is_completed = common_helper.check_activity_key(
+            activity_key, "activity_", "_completed"
+        )
+
+        if is_started:
+            return self.create_activity(activity_key, current_ist_time)
+        elif is_succeeded:
+            return self.update_succeeded_activity(current_ist_time)
+        elif is_completed:
+            return self.update_completed_activity(current_ist_time)
+        else:
+            return None
+
+    def create_activity(
+        self, activity_key: str, current_ist_time: datetime
+    ) -> models.UserActivities:
+        return self.class_model(
+            user_id=self.user_id,
+            user_phone=self.user_phone,
+            user_flow_id=self.user_flow_id,
+            activity=activity_key.strip(),
+            is_started=True,
+            started_on=current_ist_time,
+        )
+
+    def update_succeeded_activity(
+        self, current_ist_time: datetime
+    ) -> Optional[models.UserActivities]:
+        last_activity = self.class_model.query.get_todays_started_activity_for_user(
+            self.user_id, self.user_phone
+        )
+
+        if last_activity:
+            last_activity.is_succeeded = True
+            last_activity.succeeded_on = current_ist_time
+            return last_activity
+        else:
+            logger.error("No previous activity found for updating 'is_succeeded'.")
+            return None
+
+    def update_completed_activity(
+        self, current_ist_time: datetime
+    ) -> Optional[models.UserActivities]:
+        last_activity = self.class_model.query.get_todays_succeeded_activity_for_user(
+            self.user_id, self.user_phone
+        )
+
+        if last_activity:
+            last_activity.is_completed = True
+            last_activity.completed_on = current_ist_time
+            return last_activity
+        else:
+            logger.error("No previous activity found for updating 'is_completed'.")
+            return None

api/services/user_creation_service.py

@@ -1,3 +1,5 @@
+from typing import Any, Optional
+
 from api import models
 from api.utils import db_utils
 from api.utils.loggingutils import logger
@@ -7,11 +9,13 @@ class UserCreationService:
     def __init__(self):
         self.class_model = models.Users
 
-    def create_new_user(self, contact_data):
+    def create_new_user(self, contact_data: dict[str, Any]) -> Optional[models.Users]:
         try:
-            user_phone = contact_data["phone"]
-            formatted_user_phone = int(user_phone[-10:])
-            user = self.class_model.query.get_by_phone(formatted_user_phone)
+            user_phone: str = contact_data["phone"]
+            formatted_user_phone: int = int(user_phone[-10:])
+            user: Optional[models.Users] = self.class_model.query.get_by_phone(
+                formatted_user_phone
+            )
             if user:
                 logger.info(
                     f"Skipped user creation for user {user.phone}. "
@@ -29,12 +33,15 @@ def create_new_user(self, contact_data):
                 f"Error while creating new user. Contact data: {contact_data}."
                 f"Error message: {e}"
             )
+            return None
 
-    def create_user(self, contact_data, formatted_user_phone):
-        glific_user_id = contact_data["id"]
-        name = contact_data["name"]
+    def create_user(
+        self, contact_data: dict[str, Any], formatted_user_phone: int
+    ) -> models.Users:
+        glific_user_id: str = contact_data["id"]
+        name: str = contact_data["name"]
 
-        user = self.class_model(
+        user: models.Users = self.class_model(
             glific_user_id=glific_user_id,
             phone=formatted_user_phone,
             name=name,

api/services/user_indicator_response_service.py

@@ -1,16 +1,18 @@
+from typing import Any
+
 from api import db, models
 from api.utils.loggingutils import logger
 
 
 class UserIndicatorResponseService:
-    def __init__(self, user, user_flow):
-        self.key = "indicator_question"  # Prefix for the indicators key in payload
-        self.user_id = user.id
-        self.user_phone = user.phone
-        self.user_flow_id = user_flow.id
+    def __init__(self, user: models.Users, user_flow: models.UserFlows):
+        self.key: str = "indicator_question"  # Prefix for the indicators key in payload
+        self.user_id: int = user.id
+        self.user_phone: int = user.phone
+        self.user_flow_id: int = user_flow.id
         self.class_model = models.UserIndicatorResponses
 
-    def process_user_indicator_responses(self, data):
+    def process_user_indicator_responses(self, data: dict[str, Any]) -> None:
         try:
             indicators = [(key, data[key]) for key in data if key.startswith(self.key)]