stats

stats/admin.py

@@ -0,0 +1,7 @@
+from django.contrib import admin
+
+from stats.models import Snapshot
+
+# Register your models here.
+
+admin.site.register(Snapshot)

stats/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class StatsConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'stats'

stats/management/commands/fetch_stats.py

@@ -0,0 +1,31 @@
+import datetime
+from time import sleep
+
+import croniter
+from django.core.management import BaseCommand
+from django.utils import timezone
+from requests import get
+
+from stats.models import Snapshot
+from web.models import Project
+
+
+class Command(BaseCommand):
+    def handle(self, *args, **options):
+        while True:
+            for project in Project.objects.filter(next_stats_fetch_time__lte=timezone.now()):
+                if project.stats_enabled:
+                    if not project.stats_link:
+                        continue
+                    try:
+                        response = get(project.stats_link)
+                    except:
+                        continue
+                    if response.status_code != 200:
+                        continue
+                    Snapshot.objects.create(project=project, data=response.json())
+                cron = croniter.croniter(project.stats_cron, timezone.now())
+                next_date = cron.get_next(datetime.datetime)
+                project.next_stats_fetch_time = next_date
+                project.save()
+            sleep(5 * 60)

stats/migrations/0001_initial.py

@@ -0,0 +1,30 @@
+# Generated by Django 4.1.7 on 2023-09-26 16:20
+
+from django.db import migrations, models
+import django.db.models.deletion
+import django.utils.timezone
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('web', '0004_alter_customuser_vk_id_alter_customuser_yandex_id'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Snapshot',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('data', models.JSONField()),
+                ('created_at', models.DateTimeField(default=django.utils.timezone.now)),
+                ('project', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='web.project')),
+            ],
+        ),
+        migrations.AddIndex(
+            model_name='snapshot',
+            index=models.Index(fields=['project', '-created_at'], name='stats_snaps_project_4ed188_idx'),
+        ),
+    ]

stats/models.py

@@ -0,0 +1,13 @@
+from django.db import models
+from django.utils import timezone
+
+
+class Snapshot(models.Model):
+    data = models.JSONField()
+    project = models.ForeignKey('web.Project', on_delete=models.CASCADE)
+    created_at = models.DateTimeField(default=timezone.now)
+
+    class Meta:
+        indexes = [
+            models.Index(fields=['project', '-created_at'])
+        ]

stats/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

stats/urls.py

@@ -0,0 +1,7 @@
+from django.urls import path
+
+from .views import *
+
+urlpatterns = [
+    path(*StatsView.as_path()),
+]

stats/views.py

@@ -0,0 +1,44 @@
+import croniter
+import validators
+
+from BaseLib.BaseView import BaseView
+from stats.models import Snapshot
+
+
+class StatsView(BaseView):
+    endpoint = ''
+    view_file = 'stats.html'
+    required_login = True
+
+    def get(self):
+        count = 20
+        snapshots = list(Snapshot.objects.filter(project=self.request.user.selected_project).order_by('-created_at')[:count])
+        keys = set()
+        for snapshot in snapshots:
+            for key in snapshot.data:
+                keys.add(key)
+        keys = list(keys)
+        rows = [[] for _ in keys]
+        for snapshot in snapshots:
+            for index, key in enumerate(keys):
+                rows[index].append(snapshot.data.get(key))
+        self.context['keys'] = keys
+        self.context['data'] = rows
+        self.context['err'] = 'err' in self.request.GET
+
+    def post(self):
+        self.request.user.selected_project.stats_enabled = 'enabled' in self.request.POST
+        if self.request.POST['link'] == "":
+            self.request.user.selected_project.stats_link = None
+        elif validators.url(self.request.POST['link']):
+            self.request.user.selected_project.stats_link = self.request.POST['link']
+        else:
+            return '/stats?err=true'
+        if self.request.POST['cron'] == "":
+            self.request.user.selected_project.stats_cron = None
+        elif croniter.croniter.is_valid(self.request.POST['cron']):
+            self.request.user.selected_project.stats_cron = self.request.POST['cron']
+        else:
+            return '/stats?err=true'
+        self.request.user.selected_project.save()
+        return '/stats'