Misago/misago/graphql/admin/analytics.py

from datetime import timedelta

from ariadne import QueryType
from django.contrib.auth import get_user_model
from django.core.cache import cache
from django.utils import timezone

from ...threads.models import Attachment, Post, Thread
from ...users.models import DataDownload

CACHE_KEY = "misago_admin_analytics"
CACHE_LENGTH = 3600 * 4  # 4 hours

User = get_user_model()

analytics = QueryType()


@analytics.field("analytics")
def resolve_analytics(_, info, *, span):
    span = clean_span(span)
    cache_key = "%s_%s" % (CACHE_KEY, span)
    data = cache.get(cache_key)
    if not data:
        data = get_data_from_db(span)
        cache.set(cache_key, data, CACHE_LENGTH)
    return data


def clean_span(span):
    if span > 360:
        return 360
    if span < 30:
        return 30
    return span


def get_data_from_db(span):
    analytics = Analytics(span)

    return {
        "users": analytics.get_data_for_model(User, "joined_on"),
        "threads": analytics.get_data_for_model(Thread, "started_on"),
        "posts": analytics.get_data_for_model(Post, "posted_on"),
        "attachments": analytics.get_data_for_model(Attachment, "uploaded_on"),
        "dataDownloads": analytics.get_data_for_model(DataDownload, "requested_on"),
    }


class Analytics:
    def __init__(self, span):
        self.today = timezone.now()
        self.span = span

        self.cutoff = self.today - timedelta(days=span * 2)
        self.legend = self.get_legend()

    def get_legend(self):
        legend = []
        for day in range(self.span * 2):
            date = self.today - timedelta(days=day)
            legend.append(date.strftime("%x"))
        return legend

    def get_empty_data(self):
        return {k: 0 for k in self.legend}

    def get_data_for_model(self, model, date_attr):
        filter_kwarg = {"%s__gte" % date_attr: self.cutoff}
        queryset = model.objects.filter(**filter_kwarg).order_by("-pk")

        data = self.get_empty_data()
        for item in queryset.values(date_attr).iterator():
            date = item[date_attr].strftime("%x")
            if date in data:
                data[date] += 1

        values = list(data.values())
        current = list(reversed(values[: self.span]))
        previous = list(reversed(values[self.span :]))

        return {
            "current": current,
            "currentCumulative": cumulate_data(current),
            "previous": previous,
            "previousCumulative": cumulate_data(previous),
        }


def cumulate_data(data_series):
    data = []
    for v in data_series:
        if not data:
            data.append(v)
        else:
            data.append(data[-1] + v)
    return data