Using dedicated event to store working time

cvat/apps/events/serializers.py

@@ -5,7 +5,9 @@
 import datetime
 import json
 
+from contextlib import suppress
 from rest_framework import serializers
+from cvat.apps.engine.models import Job
 
 class EventSerializer(serializers.Serializer):
     scope = serializers.CharField(required=True)
@@ -30,10 +32,43 @@ class ClientEventsSerializer(serializers.Serializer):
     events = EventSerializer(many=True, default=[])
     previous_event = EventSerializer(default=None, allow_null=True, write_only=True)
     timestamp = serializers.DateTimeField()
+    _WORKING_TIME_SCOPE = 'send:working_time'
     _TIME_THRESHOLD = datetime.timedelta(seconds=100)
     _WORKING_TIME_RESOLUTION = datetime.timedelta(milliseconds=1)
     _COLLAPSED_EVENT_SCOPES = frozenset(("change:frame",))
 
+    @classmethod
+    def _generate_wt_event(cls, job_id: int | None, wt: datetime.timedelta, common: dict) -> EventSerializer | None:
+        if wt.total_seconds():
+            task_id = None
+            project_id = None
+
+            if job_id is not None:
+                with suppress(Job.DoesNotExist):
+                    task_id, project_id = Job.objects.values_list(
+                        "segment__task__id", "segment__task__project__id"
+                    ).get(pk=job_id)
+
+            value = wt // cls._WORKING_TIME_RESOLUTION
+            event = EventSerializer(data={
+                **common,
+                "scope": cls._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 })
+            })
+
+            event.is_valid(raise_exception=True)
+            return event
+
     @classmethod
     def _end_timestamp(cls, event: dict) -> datetime.datetime:
         if event["scope"] in cls._COLLAPSED_EVENT_SCOPES:
@@ -54,10 +89,14 @@ def to_internal_value(self, data):
 
         if previous_event := data["previous_event"]:
             previous_end_timestamp = self._end_timestamp(previous_event)
+            previous_job_id = previous_event.get("job_id")
         elif data["events"]:
             previous_end_timestamp = data["events"][0]["timestamp"]
+            previous_job_id = data["events"][0].get("job_id")
 
+        working_time_per_job = {}
         for event in data["events"]:
+            job_id = event.get('job_id')
             working_time = datetime.timedelta()
 
             timestamp = event["timestamp"]
@@ -73,21 +112,39 @@ def to_internal_value(self, data):
                 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,
-            })
+            if previous_job_id not in working_time_per_job:
+                working_time_per_job[previous_job_id] = datetime.timedelta()
+            working_time_per_job[previous_job_id] += working_time
+            previous_job_id = job_id
+
+            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": str(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),
             })
 
+        common = {
+            "timestamp": str(receive_time),
+            "user_id": request.user.id,
+            "user_name": request.user.username,
+            "user_email": request.user.email,
+            "org_id": org_id,
+            "org_slug": org_slug,
+        }
+
+        for job_id in working_time_per_job:
+            event = ClientEventsSerializer._generate_wt_event(job_id, working_time_per_job[job_id], common)
+            if event:
+                data["events"].append(event.validated_data)
+
         return data

cvat/apps/events/views.py

@@ -5,6 +5,7 @@
 from django.conf import settings
 from rest_framework import status, viewsets
 from rest_framework.response import Response
+from rest_framework.exceptions import PermissionDenied
 from drf_spectacular.utils import OpenApiResponse, OpenApiParameter, extend_schema
 from drf_spectacular.types import OpenApiTypes
 from rest_framework.renderers import JSONRenderer