refactor: rework events (stats)

config/settings/development.py

@@ -67,8 +67,8 @@
             "handlers": ["console"],
             "propagate": False,
         },
-        "django.request": {
-            "level": "INFO",
+        "django.server": {
+            "level": "WARNING",
             "handlers": ["console"],
             "propagate": False,
         },
@@ -87,8 +87,13 @@
             "handlers": ["console"],
             "propagate": False,
         },
+        "stats": {
+            "level": "DEBUG",
+            "handlers": ["console"],
+            "propagate": False,
+        },
         "sync": {
-            "level": "INFO",
+            "level": "DEBUG",
             "handlers": ["console"],
             "propagate": False,
         },

obr_core/stats/admin/event.py

@@ -1,4 +1,5 @@
 from django.contrib import admin
+from django.utils import timezone
 from django.utils.html import mark_safe
 
 from stats.models import PlayerEvent, StreamEvent
@@ -11,14 +12,17 @@ class PlayerEventAdmin(
     admin.ModelAdmin,
 ):
     list_display = [
-        "user_identity",
-        "user_display",
         "state",
+        "time_display",
+        "time_end_display",
+        "annotated_duration_display",
         "duration_display",
+        "max_duration_display",
+        "user_display",
         "obj_key",
         "source",
         # "user_identity",
-        "device_key",
+        # "device_key",
     ]
     list_filter = [
         "source",
@@ -34,13 +38,42 @@ class PlayerEventAdmin(
         "device_key",
     ]
 
-    @admin.display(description="User")
-    def user_display(self, obj):
-        return get_admin_link_for_user_identity(obj.user_identity)
+    def get_queryset(self, request):
+        return super().get_queryset(request).annotate_times_and_durations()
+
+    @admin.display(description="Time")
+    def time_display(self, obj):
+        timestamp_in_tz = obj.time.astimezone(timezone.get_default_timezone())
+        return timestamp_in_tz.strftime("%b %d, %Y, %H:%M:%S")
+
+    @admin.display(description="Time end")
+    def time_end_display(self, obj):
+        if not obj.time_end:
+            return None
+        timestamp_in_tz = obj.time_end.astimezone(timezone.get_default_timezone())
+        return timestamp_in_tz.strftime("%b %d, %Y, %H:%M:%S")
+
+    @admin.display(description="Duration (est)")
+    def annotated_duration_display(self, obj):
+        if obj.state == PlayerEvent.State.STOPPED:
+            return None
+        if not obj.annotated_duration:
+            return None
+        return obj.annotated_duration
 
     @admin.display(description="Duration")
     def duration_display(self, obj):
-        return obj.duration
+        if not obj.time_end:
+            return None
+        return obj.time_end - obj.time
+
+    @admin.display(description="Duration (max)")
+    def max_duration_display(self, obj):
+        return obj.max_duration
+
+    @admin.display(description="User")
+    def user_display(self, obj):
+        return get_admin_link_for_user_identity(obj.user_identity)
 
 
 @admin.register(StreamEvent)

obr_core/stats/api/serializers.py

@@ -34,7 +34,7 @@ class PlayerEventSerializer(
 ):
     time_start = serializers.DateTimeField(source="time")
     time_end = serializers.DateTimeField()
-    duration = DurationInSecondsSerializer()
+    duration = DurationInSecondsSerializer(source="annotated_duration")
 
     class Meta(CTUIDModelSerializer.Meta):
         model = PlayerEvent

obr_core/stats/api/views/event.py

@@ -45,9 +45,9 @@ class PlayerEventViewSet(
     filterset_class = PlayerEventFilter
 
     def get_queryset(self):
-        qs = PlayerEvent.objects.filter(
+        qs = PlayerEvent.objects.annotated_times_and_durations().filter(
             state="playing",
-            duration__gt=timedelta(seconds=5),
+            annotated_duration__gt=timedelta(seconds=5),
         )
 
         return qs
@@ -67,12 +67,15 @@ def override_media_durations(qs):
         for event in qs:
             media_uid = event.obj_key[-8:]
             media_duration = media_durations[media_uid]
-            if event.duration.total_seconds() > media_duration.total_seconds() + 5:
+            if (
+                event.annotated_duration.total_seconds()
+                > media_duration.total_seconds() + 5
+            ):
                 logger.debug(
-                    f"fix duration for {media_uid} - {event.duration.seconds} > {media_duration.seconds}",
+                    f"fix duration for {media_uid} - {event.annotated_duration.seconds} > {media_duration.seconds}",
                 )
                 # NOTE: yes - i know, this is ugly. see above...
-                event.duration = media_duration
+                event.annotated_duration = media_duration
 
         return qs
 

obr_core/stats/events.py

@@ -0,0 +1,85 @@
+from datetime import timedelta
+
+from django.db.models import F
+from django.db.models.functions import Now
+
+from stats.models import PlayerEvent
+
+
+def post_process_player_events(database="default"):
+    ###################################################################
+    # first pass: get unprocessed (no time_end) events where
+    # time + max_duration is in the past and older than 24 hours
+    ###################################################################
+    qs = (
+        PlayerEvent.objects.using(database)
+        .annotate(
+            annotated_time_end=F("time") + F("max_duration"),
+        )
+        .filter(
+            time_end__isnull=True,
+            annotated_time_end__lt=Now(),
+            state=PlayerEvent.State.PLAYING,
+            time__lt=Now() - timedelta(days=1),
+        )
+    )
+
+    # and set the time_end field
+    # qs.update(
+
+    print("pass 1", qs.count())
+
+    ###################################################################
+    # second pass: get unprocessed (no time_end) and update the
+    # time_end to the calculated / annotated time
+    ###################################################################
+    qs = (
+        PlayerEvent.objects.using(database)
+        .annotate_times_and_durations()
+        .annotate(
+            annotated_max_time_end=F("time") + F("max_duration"),
+        )
+        .filter(
+            time_end__isnull=True,
+            state__in=[
+                PlayerEvent.State.PLAYING,
+                PlayerEvent.State.PAUSED,
+                PlayerEvent.State.BUFFERING,
+            ],
+        )
+        .exclude(
+            time__lte=Now() - timedelta(days=10000),
+            annotated_time_end=None,
+        )
+    )
+
+    # and set the time_end field
+    # qs.update does not work here (in combination with window)
+    print("###")
+    # for event in [e for e in qs if e.annotated_time_end]:
+    for event in qs:
+        if (
+            event.state == PlayerEvent.State.PLAYING
+            and event.annotated_time_end
+            and event.annotated_max_time_end
+        ):
+            time_end = min(event.annotated_time_end, event.annotated_max_time_end)
+            print("TE", time_end)
+        else:
+            time_end = event.annotated_time_end
+
+        PlayerEvent.objects.using(database).filter(
+            pk=event.pk,
+        ).update(
+            time_end=time_end,
+        )
+
+    print("pass 2", qs.count())
+
+    return qs.count()
+
+    #
+    #
+    # # Window is disallowed in the filter clause
+    # # so filtering has to be done in python
+    #

obr_core/stats/management/commands/stats_ingest_events.py

@@ -4,7 +4,7 @@
 
 
 class Command(BaseCommand):
-    help = "Ingest events into big-query"
+    help = "Ingest events into big-query (NOT IN USE)"
 
     def add_arguments(self, parser):
         parser.add_argument(

obr_core/stats/management/commands/stats_postprocess_events.py

@@ -0,0 +1,18 @@
+from django.core.management.base import BaseCommand
+
+from stats import events
+
+
+class Command(BaseCommand):
+    help = "Post-process events (adding end-times, etc.)"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--database",
+            type=str,
+            default="default",
+        )
+
+    def handle(self, *args, **options):
+        num_processed = events.post_process_player_events(database=options["database"])
+        self.stdout.write(f"processed {num_processed} events")

obr_core/stats/migrations/0015_playerevent_max_duration_playerevent_time_end_and_more.py

@@ -0,0 +1,61 @@
+# Generated by Django 4.2.13 on 2024-06-26 14:44
+
+import datetime
+
+from django.db import migrations, models
+
+import django_countries.fields
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("stats", "0014_alter_streamevent_geoip_city_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="playerevent",
+            name="max_duration",
+            field=models.DurationField(
+                blank=True, db_index=True, default=datetime.timedelta(0)
+            ),
+        ),
+        migrations.AddField(
+            model_name="playerevent",
+            name="time_end",
+            field=models.DateTimeField(blank=True, db_index=True, null=True),
+        ),
+        migrations.AlterField(
+            model_name="streamevent",
+            name="geoip_city",
+            field=models.CharField(
+                blank=True,
+                db_index=True,
+                default="",
+                max_length=128,
+                verbose_name="city",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="streamevent",
+            name="geoip_country",
+            field=django_countries.fields.CountryField(
+                blank=True,
+                db_index=True,
+                default="",
+                max_length=2,
+                verbose_name="country",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="streamevent",
+            name="geoip_region",
+            field=models.CharField(
+                blank=True,
+                db_index=True,
+                default="",
+                max_length=128,
+                verbose_name="region",
+            ),
+        ),
+    ]

obr_core/stats/migrations/0016_alter_playerevent_max_duration.py

@@ -0,0 +1,17 @@
+# Generated by Django 4.2.13 on 2024-06-26 14:45
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("stats", "0015_playerevent_max_duration_playerevent_time_end_and_more"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="playerevent",
+            name="max_duration",
+            field=models.DurationField(blank=True, db_index=True, null=True),
+        ),
+    ]