feat: implement team activity timeline endpoint and enhance audit log… (#221)

* feat: implement team activity timeline endpoint and enhance audit logging

- Added TeamActivityTimelineView to provide a comprehensive timeline of activities related to tasks assigned to a team, including assignment, unassignment, executor changes, and status changes.
- Updated urls.py to include the new route for accessing the team activity timeline.
- Enhanced AuditLogModel to include additional fields for tracking status and assignment changes, improving the granularity of logged actions.
- Updated TaskAssignmentService and TaskService to log relevant actions to the audit log, ensuring better tracking of task assignments and status changes.

These changes improve team management and accountability by providing a detailed activity log for team-related tasks.

* refactor: remove unused AuditLogModel import in team views

- Deleted the unused import of AuditLogModel from team.py to clean up the code and improve maintainability. This change contributes to a more organized codebase by eliminating unnecessary dependencies.

* fix: correct formatting in TeamActivityTimelineView response

- Added a missing comma in the OpenApiResponse description for clarity.
- Reformatted the retrieval of previous executor names to enhance readability by breaking long lines into multiple lines.

These changes improve the consistency and maintainability of the API response structure.

* refactor: improve code formatting in task_service.py

- Standardized the use of double quotes for string literals in the TaskService class.
- This change enhances code consistency and readability without affecting functionality.

---------

Co-authored-by: Amit Prakash <[email protected]>

Author: amit-flx

todo/models/audit_log.py

@@ -8,10 +8,18 @@
 class AuditLogModel(Document):
     collection_name: ClassVar[str] = "audit_logs"
 
-    task_id: PyObjectId
-    team_id: PyObjectId
+    task_id: PyObjectId | None = None
+    team_id: PyObjectId | None = None
     previous_executor_id: PyObjectId | None = None
-    new_executor_id: PyObjectId
-    spoc_id: PyObjectId
-    action: str = "reassign_executor"
+    new_executor_id: PyObjectId | None = None
+    spoc_id: PyObjectId | None = None
+    action: str  # e.g., "assigned_to_team", "unassigned_from_team", "status_changed", "reassign_executor"
     timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    # For status changes
+    status_from: str | None = None
+    status_to: str | None = None
+    # For assignment changes
+    assignee_from: PyObjectId | None = None
+    assignee_to: PyObjectId | None = None
+    # For general user reference (who performed the action)
+    performed_by: PyObjectId | None = None

todo/repositories/audit_log_repository.py

@@ -14,3 +14,9 @@ def create(cls, audit_log: AuditLogModel) -> AuditLogModel:
         insert_result = collection.insert_one(audit_log_dict)
         audit_log.id = insert_result.inserted_id
         return audit_log
+
+    @classmethod
+    def get_by_team_id(cls, team_id: str) -> list[AuditLogModel]:
+        collection = cls.get_collection()
+        logs = collection.find({"team_id": team_id}).sort("timestamp", -1)
+        return [AuditLogModel(**log) for log in logs]

todo/services/task_assignment_service.py

@@ -11,6 +11,8 @@
 from todo.exceptions.task_exceptions import TaskNotFoundException
 from todo.models.task_assignment import TaskAssignmentModel
 from todo.dto.task_assignment_dto import TaskAssignmentDTO
+from todo.models.audit_log import AuditLogModel
+from todo.repositories.audit_log_repository import AuditLogRepository
 
 
 class TaskAssignmentService:
@@ -39,13 +41,22 @@ def create_task_assignment(cls, dto: CreateTaskAssignmentDTO, user_id: str) -> C
         # Check if task already has an active assignment
         existing_assignment = TaskAssignmentRepository.get_by_task_id(dto.task_id)
         if existing_assignment:
+            # If previous assignment was to a team, log unassignment
+            if existing_assignment.user_type == "team":
+                AuditLogRepository.create(
+                    AuditLogModel(
+                        task_id=existing_assignment.task_id,
+                        team_id=existing_assignment.assignee_id,
+                        action="unassigned_from_team",
+                        performed_by=PyObjectId(user_id),
+                    )
+                )
             # Update existing assignment
             updated_assignment = TaskAssignmentRepository.update_assignment(
                 dto.task_id, dto.assignee_id, dto.user_type, user_id
             )
             if not updated_assignment:
                 raise ValueError("Failed to update task assignment")
-
             assignment = updated_assignment
         else:
             # Create new assignment
@@ -56,9 +67,19 @@ def create_task_assignment(cls, dto: CreateTaskAssignmentDTO, user_id: str) -> C
                 created_by=PyObjectId(user_id),
                 updated_by=None,
             )
-
             assignment = TaskAssignmentRepository.create(task_assignment)
 
+        # If new assignment is to a team, log assignment
+        if assignment.user_type == "team":
+            AuditLogRepository.create(
+                AuditLogModel(
+                    task_id=assignment.task_id,
+                    team_id=assignment.assignee_id,
+                    action="assigned_to_team",
+                    performed_by=PyObjectId(user_id),
+                )
+            )
+
         # Also insert into assignee_task_details if this is a team assignment (legacy, can be removed if not needed)
         # if dto.user_type == "team":
         #     TaskAssignmentRepository.create(

todo/services/task_service.py

@@ -42,6 +42,8 @@
 from todo.repositories.user_repository import UserRepository
 from todo.repositories.watchlist_repository import WatchlistRepository
 import math
+from todo.models.audit_log import AuditLogModel
+from todo.repositories.audit_log_repository import AuditLogRepository
 
 
 @dataclass
@@ -303,6 +305,10 @@ def update_task(cls, task_id: str, validated_data: dict, user_id: str) -> TaskDT
                 if not team_data:
                     raise ValueError(f"Team not found: {assignee_id}")
 
+        # Track status change for audit log
+        old_status = getattr(current_task, "status", None)
+        new_status = validated_data.get("status")
+
         update_payload = {}
         enum_fields = {"priority": TaskPriority, "status": TaskStatus}
 
@@ -332,6 +338,18 @@ def update_task(cls, task_id: str, validated_data: dict, user_id: str) -> TaskDT
         update_payload["updatedBy"] = user_id
         updated_task = TaskRepository.update(task_id, update_payload)
 
+        # Audit log for status change
+        if old_status and new_status and old_status != new_status:
+            AuditLogRepository.create(
+                AuditLogModel(
+                    task_id=current_task.id,
+                    action="status_changed",
+                    status_from=old_status,
+                    status_to=new_status,
+                    performed_by=PyObjectId(user_id),
+                )
+            )
+
         if not updated_task:
             raise TaskNotFoundException(task_id)
 

todo/urls.py

@@ -11,6 +11,7 @@
     JoinTeamByInviteCodeView,
     AddTeamMembersView,
     TeamInviteCodeView,
+    TeamActivityTimelineView,
 )
 from todo.views.watchlist import WatchlistListView, WatchlistDetailView, WatchlistCheckView
 from todo.views.task_assignment import TaskAssignmentView, TaskAssignmentDetailView
@@ -22,6 +23,7 @@
     path("teams/<str:team_id>", TeamDetailView.as_view(), name="team_detail"),
     path("teams/<str:team_id>/members", AddTeamMembersView.as_view(), name="add_team_members"),
     path("teams/<str:team_id>/invite-code", TeamInviteCodeView.as_view(), name="team_invite_code"),
+    path("teams/<str:team_id>/activity-timeline", TeamActivityTimelineView.as_view(), name="team_activity_timeline"),
     path("tasks", TaskListView.as_view(), name="tasks"),
     path("tasks/<str:task_id>", TaskDetailView.as_view(), name="task_detail"),
     path("tasks/<str:task_id>/update", TaskUpdateView.as_view(), name="update_task_and_assignee"),

todo/views/team.py

@@ -19,6 +19,9 @@
 from todo.dto.team_dto import TeamDTO
 from todo.services.user_service import UserService
 from todo.repositories.team_repository import TeamRepository
+from todo.repositories.audit_log_repository import AuditLogRepository
+from todo.repositories.user_repository import UserRepository
+from todo.repositories.task_repository import TaskRepository
 
 
 class TeamListView(APIView):
@@ -367,3 +370,85 @@ def get(self, request: Request, team_id: str):
             {"detail": "You are not authorized to view the invite code for this team."},
             status=status.HTTP_403_FORBIDDEN,
         )
+
+
+class TeamActivityTimelineView(APIView):
+    @extend_schema(
+        operation_id="get_team_activity_timeline",
+        summary="Get team activity timeline",
+        description="Return a timeline of all activities related to tasks assigned to the team, including assignment, unassignment, executor changes, and status changes. All IDs are replaced with names.",
+        tags=["teams"],
+        parameters=[
+            OpenApiParameter(
+                name="team_id",
+                type=OpenApiTypes.STR,
+                location=OpenApiParameter.PATH,
+                description="Unique identifier of the team",
+                required=True,
+            ),
+        ],
+        responses={
+            200: OpenApiResponse(
+                response={
+                    "type": "object",
+                    "properties": {
+                        "timeline": {
+                            "type": "array",
+                            "items": {"type": "object"},
+                        }
+                    },
+                },
+                description="Team activity timeline returned successfully",
+            ),
+            404: OpenApiResponse(description="Team not found"),
+        },
+    )
+    def get(self, request: Request, team_id: str):
+        team = TeamRepository.get_by_id(team_id)
+        if not team:
+            return Response({"detail": "Team not found."}, status=status.HTTP_404_NOT_FOUND)
+        logs = AuditLogRepository.get_by_team_id(team_id)
+        # Pre-fetch team name
+        team_name = team.name
+        # Pre-fetch all user and task names needed
+        user_ids = set()
+        task_ids = set()
+        for log in logs:
+            if log.performed_by:
+                user_ids.add(str(log.performed_by))
+            if log.spoc_id:
+                user_ids.add(str(log.spoc_id))
+            if log.previous_executor_id:
+                user_ids.add(str(log.previous_executor_id))
+            if log.new_executor_id:
+                user_ids.add(str(log.new_executor_id))
+            if log.task_id:
+                task_ids.add(str(log.task_id))
+        user_map = {str(u.id): u.name for u in UserRepository.get_by_ids(list(user_ids))}
+        task_map = {str(t.id): t.title for t in TaskRepository.get_by_ids(list(task_ids))}
+        timeline = []
+        for log in logs:
+            entry = {
+                "action": log.action,
+                "timestamp": log.timestamp,
+            }
+            if log.task_id:
+                entry["task_title"] = task_map.get(str(log.task_id), str(log.task_id))
+            if log.team_id:
+                entry["team_name"] = team_name
+            if log.performed_by:
+                entry["performed_by_name"] = user_map.get(str(log.performed_by), str(log.performed_by))
+            if log.spoc_id:
+                entry["spoc_name"] = user_map.get(str(log.spoc_id), str(log.spoc_id))
+            if log.previous_executor_id:
+                entry["previous_executor_name"] = user_map.get(
+                    str(log.previous_executor_id), str(log.previous_executor_id)
+                )
+            if log.new_executor_id:
+                entry["new_executor_name"] = user_map.get(str(log.new_executor_id), str(log.new_executor_id))
+            if log.status_from:
+                entry["status_from"] = log.status_from
+            if log.status_to:
+                entry["status_to"] = log.status_to
+            timeline.append(entry)
+        return Response({"timeline": timeline}, status=status.HTTP_200_OK)