Using dedicated event to store working time

changelog.d/20240529_154845_boris_dedicated_event_for_wt.md

@@ -0,0 +1,4 @@
+### Changed
+
+- CVAT now stores user's working time in ad hoc event type
+  (<https://github.com/cvat-ai/cvat/pull/7958>)

cvat-ui/src/actions/annotation-actions.ts

@@ -925,14 +925,7 @@ export function getJobAsync({
                 throw new Error('Requested resource id is not valid');
             }
 
-            const loadJobEvent = await logger.log(
-                EventScope.loadJob,
-                {
-                    task_id: taskID,
-                    job_id: jobID,
-                },
-                true,
-            );
+            const loadJobEvent = await logger.log(EventScope.loadJob, {}, true);
 
             getCore().config.globalObjectsCounter = 0;
             const [job] = await cvat.jobs.get({ jobID });
@@ -979,7 +972,12 @@ export function getJobAsync({
                 }
             }
 
-            loadJobEvent.close(await jobInfoGenerator(job));
+            loadJobEvent.close({
+                ...await jobInfoGenerator(job),
+                jobID: job.id,
+                taskID: job.taskId,
+                projectID: job.projectId,
+            });
 
             const openTime = Date.now();
             dispatch({

cvat/apps/events/handlers.py

@@ -1,43 +1,37 @@
-# Copyright (C) 2023 CVAT.ai Corporation
+# Copyright (C) 2023-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
+import datetime
+import json
+import traceback
 from copy import deepcopy
 from typing import Optional, Union
-import traceback
-import rq
 
-from rest_framework.views import exception_handler
-from rest_framework.exceptions import NotAuthenticated
+import rq
+from crum import get_current_request, get_current_user
 from rest_framework import status
-from crum import get_current_user, get_current_request
-
-from cvat.apps.engine.models import (
-    Project,
-    Task,
-    Job,
-    User,
-    CloudStorage,
-    Issue,
-    Comment,
-    Label,
-)
-from cvat.apps.engine.serializers import (
-    ProjectReadSerializer,
-    TaskReadSerializer,
-    JobReadSerializer,
-    BasicUserSerializer,
-    CloudStorageReadSerializer,
-    IssueReadSerializer,
-    CommentReadSerializer,
-    LabelSerializer,
-)
-from cvat.apps.engine.models import ShapeType
-from cvat.apps.organizations.models import Membership, Organization, Invitation
-from cvat.apps.organizations.serializers import OrganizationReadSerializer, MembershipReadSerializer, InvitationReadSerializer
+from rest_framework.exceptions import NotAuthenticated
+from rest_framework.views import exception_handler
+
+from cvat.apps.engine.models import (CloudStorage, Comment, Issue, Job, Label,
+                                     Project, ShapeType, Task, User)
+from cvat.apps.engine.serializers import (BasicUserSerializer,
+                                          CloudStorageReadSerializer,
+                                          CommentReadSerializer,
+                                          IssueReadSerializer,
+                                          JobReadSerializer, LabelSerializer,
+                                          ProjectReadSerializer,
+                                          TaskReadSerializer)
+from cvat.apps.organizations.models import Invitation, Membership, Organization
+from cvat.apps.organizations.serializers import (InvitationReadSerializer,
+                                                 MembershipReadSerializer,
+                                                 OrganizationReadSerializer)
 
-from .event import event_scope, record_server_event
 from .cache import get_cache
+from .event import event_scope, record_server_event
+from .serializers import EventSerializer
+
 
 def project_id(instance):
     if isinstance(instance, Project):
@@ -141,7 +135,7 @@ def user_name(instance=None):
 
 def user_email(instance=None):
     current_user = get_user(instance)
-    return _get_value(current_user, "email")
+    return _get_value(current_user, "email") or None
 
 def organization_slug(instance):
     if isinstance(instance, Organization):
@@ -594,3 +588,96 @@ def handle_viewset_exception(exc, context):
     )
 
     return response
+
+def handle_client_events_push(request, data: dict):
+    TIME_THRESHOLD = datetime.timedelta(seconds=100)
+    WORKING_TIME_SCOPE = 'send:working_time'
+    WORKING_TIME_RESOLUTION = datetime.timedelta(milliseconds=1)
+    COLLAPSED_EVENT_SCOPES = frozenset(("change:frame",))
+    org = request.iam_context["organization"]
+
+    def read_ids(event: dict) -> tuple[int | None, int | None, int | None]:
+        return event.get("job_id"), event.get("task_id"), event.get("project_id")
+
+    def get_end_timestamp(event: dict) -> datetime.datetime:
+        if event["scope"] in COLLAPSED_EVENT_SCOPES:
+            return event["timestamp"] + datetime.timedelta(milliseconds=event["duration"])
+        return event["timestamp"]
+
+    def generate_wt_event(
+        ids: tuple[int | None, int | None, int | None],
+        wt: datetime.timedelta,
+        common: dict
+    ) -> EventSerializer | None:
+        job_id, task_id, project_id = ids
+        if wt.total_seconds():
+            value = wt // WORKING_TIME_RESOLUTION
+            event = EventSerializer(data={
+                **common,
+                "scope": WORKING_TIME_SCOPE,
+                "obj_name": "working_time",
+                "obj_val": value,
+                "source": "server",
+                "count": 1,
+                "project_id": project_id,
+                "task_id": task_id,
+                "job_id": job_id,
+                # keep it in payload for backward compatibility
+                # but in the future it is much better to use a dedicated field "obj_value"
+                # because parsing JSON in SQL query is very slow
+                "payload": json.dumps({ "working_time": value })
+            })
+
+            return event
+
+    if previous_event := data["previous_event"]:
+        previous_end_timestamp = get_end_timestamp(previous_event)
+        previous_ids = read_ids(previous_event)
+    elif data["events"]:
+        previous_end_timestamp = data["events"][0]["timestamp"]
+        previous_ids = read_ids(data["events"][0])
+
+    working_time_dict = {}
+    for event in data["events"]:
+        working_time = datetime.timedelta()
+        timestamp = event["timestamp"]
+
+        if timestamp > previous_end_timestamp:
+            t_diff = timestamp - previous_end_timestamp
+            if t_diff < TIME_THRESHOLD:
+                working_time += t_diff
+
+            previous_end_timestamp = timestamp
+
+        end_timestamp = get_end_timestamp(event)
+        if end_timestamp > previous_end_timestamp:
+            working_time += end_timestamp - previous_end_timestamp
+            previous_end_timestamp = end_timestamp
+
+        if previous_ids not in working_time_dict:
+            working_time_dict[previous_ids] = datetime.timedelta()
+
+        working_time_dict[previous_ids] += working_time
+        previous_ids = read_ids(event)
+
+    if data["events"]:
+        common = {
+            "timestamp": data["events"][0]["timestamp"],
+            "user_id": request.user.id,
+            "user_name": request.user.username,
+            "user_email": request.user.email or None,
+            "org_id": getattr(org, "id", None),
+            "org_slug": getattr(org, "slug", None),
+        }
+
+        for ids in working_time_dict:
+            event = generate_wt_event(ids, working_time_dict[ids], common)
+            if event:
+                event.is_valid(raise_exception=True)
+                record_server_event(
+                    scope=WORKING_TIME_SCOPE,
+                    request_id=request_id(),
+                    payload=json.loads(event.validated_data['payload']),
+                    timestamp=str(event.validated_data['timestamp'].timestamp()),
+                    **{key: value for key, value in event.validated_data.items() if key not in ['scope', 'payload', 'timestamp']}
+                )

cvat/apps/events/serializers.py

@@ -7,6 +7,7 @@
 
 from rest_framework import serializers
 
+
 class EventSerializer(serializers.Serializer):
     scope = serializers.CharField(required=True)
     obj_name = serializers.CharField(required=False, allow_null=True)
@@ -30,16 +31,6 @@ class ClientEventsSerializer(serializers.Serializer):
     events = EventSerializer(many=True, default=[])
     previous_event = EventSerializer(default=None, allow_null=True, write_only=True)
     timestamp = serializers.DateTimeField()
-    _TIME_THRESHOLD = datetime.timedelta(seconds=100)
-    _WORKING_TIME_RESOLUTION = datetime.timedelta(milliseconds=1)
-    _COLLAPSED_EVENT_SCOPES = frozenset(("change:frame",))
-
-    @classmethod
-    def _end_timestamp(cls, event: dict) -> datetime.datetime:
-        if event["scope"] in cls._COLLAPSED_EVENT_SCOPES:
-            return event["timestamp"] + datetime.timedelta(milliseconds=event["duration"])
-
-        return event["timestamp"]
 
     def to_internal_value(self, data):
         data = super().to_internal_value(data)
@@ -52,42 +43,26 @@ def to_internal_value(self, data):
         receive_time = datetime.datetime.now(datetime.timezone.utc)
         time_correction = receive_time - send_time
 
-        if previous_event := data["previous_event"]:
-            previous_end_timestamp = self._end_timestamp(previous_event)
-        elif data["events"]:
-            previous_end_timestamp = data["events"][0]["timestamp"]
+        if data["previous_event"]:
+            data["previous_event"].update({
+                "timestamp": data["previous_event"]["timestamp"] + time_correction,
+            })
 
         for event in data["events"]:
-            working_time = datetime.timedelta()
-
-            timestamp = event["timestamp"]
-            if timestamp > previous_end_timestamp:
-                t_diff = timestamp - previous_end_timestamp
-                if t_diff < self._TIME_THRESHOLD:
-                    working_time += t_diff
-
-                previous_end_timestamp = timestamp
-
-            end_timestamp = self._end_timestamp(event)
-            if end_timestamp > previous_end_timestamp:
-                working_time += end_timestamp - previous_end_timestamp
-                previous_end_timestamp = end_timestamp
-
-            payload = json.loads(event.get("payload", "{}"))
-            payload.update({
-                "working_time": working_time // self._WORKING_TIME_RESOLUTION,
-                "username": request.user.username,
-            })
+            try:
+                json_payload = json.loads(event.get("payload", "{}"))
+            except json.JSONDecodeError:
+                raise serializers.ValidationError("JSON payload is not valid in passed event")
 
             event.update({
-                "timestamp": str((timestamp + time_correction).timestamp()),
+                "timestamp": event["timestamp"] + time_correction,
                 "source": "client",
                 "org_id": org_id,
                 "org_slug": org_slug,
                 "user_id": request.user.id,
                 "user_name": request.user.username,
                 "user_email": request.user.email,
-                "payload": json.dumps(payload),
+                "payload": json.dumps(json_payload),
             })
 
         return data

cvat/apps/events/views.py

@@ -1,23 +1,41 @@
-# Copyright (C) 2023 CVAT.ai Corporation
+# Copyright (C) 2023-2024 CVAT.ai Corporation
 #
 # SPDX-License-Identifier: MIT
 
 from django.conf import settings
-from rest_framework import status, viewsets
-from rest_framework.response import Response
-from drf_spectacular.utils import OpenApiResponse, OpenApiParameter, extend_schema
 from drf_spectacular.types import OpenApiTypes
+from drf_spectacular.utils import (OpenApiParameter, OpenApiResponse,
+                                   extend_schema)
+from rest_framework import status, viewsets
 from rest_framework.renderers import JSONRenderer
+from rest_framework.response import Response
+from rest_framework.serializers import ValidationError
 
-
-from cvat.apps.iam.filters import ORGANIZATION_OPEN_API_PARAMETERS
+from cvat.apps.engine.log import vlogger
 from cvat.apps.events.permissions import EventsPermission
 from cvat.apps.events.serializers import ClientEventsSerializer
-from cvat.apps.engine.log import vlogger
+from cvat.apps.iam.filters import ORGANIZATION_OPEN_API_PARAMETERS
+
 from .export import export
+from .handlers import handle_client_events_push
+
 
 class EventsViewSet(viewsets.ViewSet):
     serializer_class = None
+    SERVER_ONLY_SCOPES = frozenset((
+        'create:project', 'update:project', 'delete:project',
+        'create:task', 'update:task', 'delete:task',
+        'create:job', 'update:job', 'delete:job',
+        'create:organization', 'update:organization', 'delete:organization',
+        'create:user', 'update:user', 'delete:user',
+        'create:cloudstorage', 'update:cloudstorage', 'delete:cloudstorage',
+        'create:issue', 'update:issue', 'delete:issue',
+        'create:comment', 'update:comment', 'delete:comment',
+        'create:annotations', 'update:annotations', 'delete:annotations',
+        'create:label', 'update:label', 'delete:label',
+        'export:dataset', 'import:dataset',
+        'send:working_time',
+    ))
 
     @extend_schema(summary='Log client events',
         methods=['POST'],
@@ -31,11 +49,20 @@ def create(self, request):
         serializer = ClientEventsSerializer(data=request.data, context={"request": request})
         serializer.is_valid(raise_exception=True)
 
-        for event in serializer.data["events"]:
-            message = JSONRenderer().render(event).decode('UTF-8')
+        for event in serializer.validated_data['events']:
+            scope = event['scope']
+            if scope in EventsViewSet.SERVER_ONLY_SCOPES:
+                raise ValidationError(f'Event scope **{scope}** is not allowed from client')
+
+        handle_client_events_push(request, serializer.validated_data)
+        for event in serializer.validated_data["events"]:
+            message = JSONRenderer().render({
+                **event,
+                'timestamp': str(event["timestamp"].timestamp())
+            }).decode('UTF-8')
             vlogger.info(message)
 
-        return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.validated_data, status=status.HTTP_201_CREATED)
 
     @extend_schema(summary='Get an event log',
         methods=['GET'],